private void preProcessLabelFileTasks(TaskRequest taskRequest) {
        List<Integer> needProcessedStatusList = Arrays.asList(
                SourceStatus.SOURCE_NORMAL.getCode(),
                SourceStatus.SOURCE_FAILED.getCode(),
                SourceStatus.SOURCE_STOP.getCode(),
                SourceStatus.TO_BE_ISSUED_ADD.getCode(),
                SourceStatus.TO_BE_ISSUED_STOP.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");

        InlongClusterNodeEntity clusterNodeEntity = selectByIpAndCluster(agentClusterName, agentIp);
        List<StreamSourceEntity> sourceEntities = sourceMapper.selectByAgentIpAndCluster(needProcessedStatusList,
                Lists.newArrayList(SourceType.FILE), agentIp, agentClusterName);

        sourceEntities.forEach(sourceEntity -> {
            // case: agent tag unbind and mismatch source task
            Set<SourceStatus> exceptedUnmatchedStatus = Sets.newHashSet(
                    SourceStatus.SOURCE_STOP,
                    SourceStatus.TO_BE_ISSUED_STOP);
            if (!matchGroup(sourceEntity, clusterNodeEntity)
                    && !exceptedUnmatchedStatus.contains(SourceStatus.forCode(sourceEntity.getStatus()))) {
                LOGGER.info("Transform task({}) from {} to {} because tag mismatch "
                        + "for agent({}) in cluster({})", sourceEntity.getAgentIp(),
                        sourceEntity.getStatus(), SourceStatus.TO_BE_ISSUED_STOP.getCode(),
                        agentIp, agentClusterName);
                sourceMapper.updateStatus(
                        sourceEntity.getId(), SourceStatus.TO_BE_ISSUED_STOP.getCode(), false);
            }

            // case: agent tag rebind and match source task again and stream is not in 'SUSPENDED' status
            InlongGroupEntity groupEntity = groupMapper.selectByGroupId(sourceEntity.getInlongGroupId());
            Set<SourceStatus> exceptedMatchedSourceStatus = Sets.newHashSet(
                    SourceStatus.SOURCE_NORMAL,
                    SourceStatus.TO_BE_ISSUED_ADD,
                    SourceStatus.TO_BE_ISSUED_ACTIVE);
            Set<GroupStatus> matchedGroupStatus = Sets.newHashSet(GroupStatus.CONFIG_SUCCESSFUL);
            if (matchGroup(sourceEntity, clusterNodeEntity)
                    && groupEntity != null
                    && !exceptedMatchedSourceStatus.contains(SourceStatus.forCode(sourceEntity.getStatus()))
                    && matchedGroupStatus.contains(GroupStatus.forCode(groupEntity.getStatus()))) {
                LOGGER.info("Transform task({}) from {} to {} because tag rematch "
                        + "for agent({}) in cluster({})", sourceEntity.getAgentIp(),
                        sourceEntity.getStatus(), SourceStatus.TO_BE_ISSUED_ACTIVE.getCode(),
                        agentIp, agentClusterName);
                sourceMapper.updateStatus(
                        sourceEntity.getId(), SourceStatus.TO_BE_ISSUED_ACTIVE.getCode(), false);
            }
        });
    }
Find file collecting task match those condition:
1.agent ip match
2.cluster name match
Send the corresponding task action request according to the matching state of the tag and the current state
inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/core/impl/AgentServiceImpl.java