private void preProcessTemplateFileTask(TaskRequest taskRequest) {
        List<Integer> needCopiedStatusList = Arrays.asList(SourceStatus.TO_BE_ISSUED_ADD.getCode(),
                SourceStatus.TO_BE_ISSUED_ACTIVE.getCode());
        final String agentIp = taskRequest.getAgentIp();
        final String agentClusterName = taskRequest.getClusterName();
        Preconditions.expectTrue(StringUtils.isNotBlank(agentIp) || StringUtils.isNotBlank(agentClusterName),
                "both agent ip and cluster name are blank when fetching file task");

        // find those node whose tag match stream_source tag and agent ip match stream_source agent ip
        List<StreamSourceEntity> sourceEntities = sourceMapper.selectTemplateSourceByCluster(needCopiedStatusList,
                Lists.newArrayList(SourceType.FILE), agentClusterName);
        Set<GroupStatus> noNeedAddTask = Sets.newHashSet(
                GroupStatus.CONFIG_OFFLINE_SUCCESSFUL, GroupStatus.CONFIG_OFFLINE_ING, GroupStatus.CONFIG_DELETING,
                GroupStatus.CONFIG_DELETED);
        sourceEntities.stream()
                .forEach(sourceEntity -> {
                    InlongGroupEntity groupEntity = groupMapper.selectByGroupId(sourceEntity.getInlongGroupId());
                    if (groupEntity != null && noNeedAddTask.contains(GroupStatus.forCode(groupEntity.getStatus()))) {
                        return;
                    }
                    StreamSourceEntity subSource = sourceMapper.selectOneByTaskMapIdAndAgentIp(sourceEntity.getId(),
                            agentIp);
                    if (subSource == null) {
                        InlongClusterNodeEntity clusterNodeEntity = selectByIpAndCluster(agentClusterName, agentIp);
                        // if stream_source match node_group with node, clone a subtask for this Agent.
                        // note: a new source name with random suffix is generated to adhere to the unique constraint
                        if (matchGroup(sourceEntity, clusterNodeEntity)) {
                            StreamSourceEntity fileEntity =
                                    CommonBeanUtils.copyProperties(sourceEntity, StreamSourceEntity::new);
                            fileEntity.setSourceName(fileEntity.getSourceName() + "-"
                                    + RandomStringUtils.randomAlphanumeric(10).toLowerCase(Locale.ROOT));
                            fileEntity.setTaskMapId(sourceEntity.getId());
                            fileEntity.setAgentIp(agentIp);
                            fileEntity.setStatus(SourceStatus.TO_BE_ISSUED_ADD.getCode());
                            // create new sub source task
                            sourceMapper.insert(fileEntity);
                            LOGGER.info("Transform new template task({}) for agent({}) in cluster({}).",
                                    fileEntity.getId(), taskRequest.getAgentIp(), taskRequest.getClusterName());
                        }
                    }
                });
    }
Add subtasks to template tasks.
(Template task are agent_ip is null and task_map_id is null)
inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/core/impl/AgentServiceImpl.java