private DataConfig getDataConfig(StreamSourceEntity entity, int op) {
        DataConfig dataConfig = new DataConfig();
        dataConfig.setIp(entity.getAgentIp());
        dataConfig.setUuid(entity.getUuid());
        dataConfig.setOp(String.valueOf(op));
        dataConfig.setTaskId(entity.getId());
        dataConfig.setTaskType(getTaskType(entity));
        dataConfig.setTaskName(entity.getSourceName());
        dataConfig.setSnapshot(entity.getSnapshot());
        dataConfig.setTimeZone(entity.getDataTimeZone());
        dataConfig.setVersion(entity.getVersion());

        String groupId = entity.getInlongGroupId();
        String streamId = entity.getInlongStreamId();
        dataConfig.setInlongGroupId(groupId);
        dataConfig.setInlongStreamId(streamId);

        InlongGroupEntity groupEntity = groupMapper.selectByGroupIdWithoutTenant(groupId);
        InlongStreamEntity streamEntity = streamMapper.selectByIdentifier(groupId, streamId);
        if (operatorFactory == null) {
            operatorFactory = new SourceOperatorFactory();
            autowireCapableBeanFactory.autowireBean(operatorFactory);
        }
        StreamSourceOperator sourceOperator = operatorFactory.getInstance(entity.getSourceType());
        String extParams = sourceOperator.getExtParams(entity);
        if (groupEntity != null && streamEntity != null) {
            dataConfig.setState(
                    SourceStatus.NORMAL_STATUS_SET.contains(SourceStatus.forCode(entity.getStatus()))
                            ? TaskStateEnum.RUNNING.getType()
                            : TaskStateEnum.FROZEN.getType());
            dataConfig.setSyncSend(streamEntity.getSyncSend());
            extParams = sourceOperator.updateDataConfig(extParams, streamEntity, dataConfig);
            InlongStreamInfo streamInfo = CommonBeanUtils.copyProperties(streamEntity, InlongStreamInfo::new);
            // Processing extParams
            unpackExtParams(streamEntity.getExtParams(), streamInfo);
            dataConfig.setPredefinedFields(streamInfo.getPredefinedFields());

            int dataReportType = groupEntity.getDataReportType();
            dataConfig.setDataReportType(dataReportType);
            if (InlongConstants.REPORT_TO_MQ_RECEIVED == dataReportType) {
                // add mq cluster setting
                List<MQClusterInfo> mqSet = new ArrayList<>();
                List<String> clusterTagList = Collections.singletonList(groupEntity.getInlongClusterTag());
                ClusterPageRequest pageRequest = ClusterPageRequest.builder()
                        .type(groupEntity.getMqType())
                        .clusterTagList(clusterTagList)
                        .build();
                List<InlongClusterEntity> mqClusterList = clusterMapper.selectByCondition(pageRequest);
                for (InlongClusterEntity cluster : mqClusterList) {
                    MQClusterInfo clusterInfo = new MQClusterInfo();
                    clusterInfo.setUrl(cluster.getUrl());
                    clusterInfo.setToken(cluster.getToken());
                    clusterInfo.setMqType(cluster.getType());
                    clusterInfo.setParams(JsonUtils.parseObject(cluster.getExtParams(), HashMap.class));
                    mqSet.add(clusterInfo);
                }
                dataConfig.setMqClusters(mqSet);

                // add topic setting
                String mqResource = groupEntity.getMqResource();
                String mqType = groupEntity.getMqType();
                if (MQType.PULSAR.equals(mqType) || MQType.TDMQ_PULSAR.equals(mqType)) {
                    // first get the tenant from the InlongGroup, and then get it from the PulsarCluster.
                    InlongPulsarDTO pulsarDTO = InlongPulsarDTO.getFromJson(groupEntity.getExtParams());
                    String tenant = pulsarDTO.getPulsarTenant();
                    if (StringUtils.isBlank(tenant)) {
                        // If there are multiple Pulsar clusters, take the first one.
                        // Note that the tenants in multiple Pulsar clusters must be identical.
                        PulsarClusterDTO pulsarCluster = PulsarClusterDTO.getFromJson(
                                mqClusterList.get(0).getExtParams());
                        tenant = pulsarCluster.getPulsarTenant();
                    }

                    String topic = String.format(InlongConstants.PULSAR_TOPIC_FORMAT,
                            tenant, mqResource, streamEntity.getMqResource());
                    DataProxyTopicInfo topicConfig = new DataProxyTopicInfo();
                    topicConfig.setInlongGroupId(groupId + "/" + streamId);
                    topicConfig.setTopic(topic);
                    dataConfig.setTopicInfo(topicConfig);
                } else if (MQType.TUBEMQ.equals(mqType)) {
                    DataProxyTopicInfo topicConfig = new DataProxyTopicInfo();
                    topicConfig.setInlongGroupId(groupId);
                    topicConfig.setTopic(mqResource);
                    dataConfig.setTopicInfo(topicConfig);
                } else if (MQType.KAFKA.equals(mqType)) {
                    DataProxyTopicInfo topicConfig = new DataProxyTopicInfo();
                    topicConfig.setInlongGroupId(groupId);
                    topicConfig.setTopic(groupEntity.getMqResource() + DOT + streamEntity.getMqResource());
                    dataConfig.setTopicInfo(topicConfig);
                }
            } else {
                LOGGER.warn("set syncSend=[0] as the stream not exists for groupId={}, streamId={}", groupId, streamId);
            }
        }
        dataConfig.setExtParams(extParams);
        return dataConfig;
    }
If the stream source can only use one data type, return the data type that has been set.

@param streamSource stream source
@param streamDataType stream data type
@return serialization type
inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/source/AbstractSourceOperator.java