@Override
    public void updateAgentTaskConfig(SourceRequest request, String operator) {
        try {
            if (SourceType.AUTO_PUSH.equals(request.getSourceType())) {
                return;
            }
            final String clusterName = request.getInlongClusterName();
            final String ip = request.getAgentIp();
            final String uuid = request.getUuid();
            if (StringUtils.isBlank(clusterName) || StringUtils.isBlank(ip)) {
                LOGGER.warn("skip update agent task config where cluster name or ip is null for request={}", request);
                return;
            }
            AgentTaskConfigEntity existEntity = agentTaskConfigEntityMapper.selectByIdentifier(ip, clusterName);
            AgentTaskConfigEntity agentTaskConfigEntity = new AgentTaskConfigEntity();
            if (existEntity != null) {
                agentTaskConfigEntity = CommonBeanUtils.copyProperties(existEntity, AgentTaskConfigEntity::new, true);
            }

            LOGGER.debug("begin to get agent config info for {}", request);
            Set<String> tagSet = new HashSet<>(16);
            InlongClusterEntity agentClusterInfo = clusterMapper.selectByNameAndType(request.getInlongClusterName(),
                    ClusterType.AGENT);
            if (agentClusterInfo == null) {
                agentTaskConfigEntity.setIsDeleted(agentTaskConfigEntity.getId());
                agentTaskConfigEntityMapper.updateByIdSelective(agentTaskConfigEntity);
                return;
            }
            String clusterTag = agentClusterInfo.getClusterTags();
            AgentConfigInfo agentConfigInfo = AgentConfigInfo.builder()
                    .cluster(AgentConfigInfo.AgentClusterInfo.builder()
                            .parentId(agentClusterInfo.getId())
                            .clusterName(agentClusterInfo.getName())
                            .build())
                    .build();
            if (StringUtils.isNotBlank(clusterTag)) {
                tagSet.addAll(Arrays.asList(clusterTag.split(InlongConstants.COMMA)));
                List<String> clusterTagList = new ArrayList<>(tagSet);
                ClusterPageRequest pageRequest = ClusterPageRequest.builder()
                        .type(ClusterType.AGENT_ZK)
                        .clusterTagList(clusterTagList)
                        .build();
                List<InlongClusterEntity> agentZkCluster = clusterMapper.selectByCondition(pageRequest);
                if (CollectionUtils.isNotEmpty(agentZkCluster)) {
                    agentConfigInfo.setZkUrl(agentZkCluster.get(0).getUrl());
                }
            }

            String jsonStr = GSON.toJson(agentConfigInfo);
            String configMd5 = DigestUtils.md5Hex(jsonStr);
            agentConfigInfo.setMd5(configMd5);
            agentConfigInfo.setCode(AgentResponseCode.SUCCESS);
            agentTaskConfigEntity.setConfigParams(objectMapper.writeValueAsString(agentConfigInfo));

            List<StreamSourceEntity> normalSourceEntities = sourceMapper.selectByStatusAndCluster(
                    SourceStatus.NORMAL_STATUS_SET.stream().map(SourceStatus::getCode)
                            .collect(Collectors.toList()),
                    clusterName, ip, uuid);
            List<StreamSourceEntity> taskLists = new ArrayList<>(normalSourceEntities);
            List<StreamSourceEntity> stopSourceEntities = sourceMapper.selectByStatusAndCluster(
                    SourceStatus.STOP_STATUS_SET.stream().map(SourceStatus::getCode)
                            .collect(Collectors.toList()),
                    clusterName, ip, uuid);
            taskLists.addAll(stopSourceEntities);
            LOGGER.debug("success to add task : {}", taskLists.size());
            List<DataConfig> runningTaskConfig = Lists.newArrayList();
            List<CmdConfig> cmdConfigs = sourceCmdConfigMapper.queryCmdByAgentIp(request.getAgentIp()).stream()
                    .map(cmd -> {
                        CmdConfig cmdConfig = new CmdConfig();
                        cmdConfig.setDataTime(cmd.getSpecifiedDataTime());
                        cmdConfig.setOp(cmd.getCmdType());
                        cmdConfig.setId(cmd.getId());
                        cmdConfig.setTaskId(cmd.getTaskId());
                        return cmdConfig;
                    }).collect(Collectors.toList());
            if (CollectionUtils.isEmpty(taskLists)) {
                agentTaskConfigEntity.setTaskParams("");
                agentTaskConfigEntityMapper.updateByIdSelective(agentTaskConfigEntity);
                return;
            }
            for (StreamSourceEntity sourceEntity : taskLists) {
                int op = sourceEntity.getStatus() % MODULUS_100;
                DataConfig dataConfig = getDataConfig(sourceEntity, op);
                runningTaskConfig.add(dataConfig);
            }
            TaskResult taskResult =
                    TaskResult.builder().dataConfigs(runningTaskConfig).cmdConfigs(cmdConfigs).build();
            String md5 = DigestUtils.md5Hex(GSON.toJson(taskResult));
            taskResult.setMd5(md5);
            taskResult.setCode(AgentResponseCode.SUCCESS);
            agentTaskConfigEntity.setAgentIp(request.getAgentIp());
            agentTaskConfigEntity.setClusterName(request.getInlongClusterName());
            agentTaskConfigEntity.setTaskParams(objectMapper.writeValueAsString(taskResult));

            agentClusterInfo.setModifier(operator);
            if (existEntity == null) {
                agentTaskConfigEntity.setCreator(operator);
                agentTaskConfigEntityMapper.insert(agentTaskConfigEntity);
            } else {
                agentTaskConfigEntityMapper.updateByIdSelective(agentTaskConfigEntity);
            }
            LOGGER.debug("success to update agent config info for: {}, result: {}", request, agentConfigInfo);
        } catch (Exception e) {
            String errMsg = String.format("update agent task config failed for groupId=%s, streamId=%s, ip=%s",
                    request.getInlongGroupId(), request.getInlongStreamId(), request.getAgentIp());
            LOGGER.error(errMsg, e);
            throw new BusinessException(errMsg);
        }
    }
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