@Override
    @Transactional(rollbackFor = Throwable.class, isolation = Isolation.REPEATABLE_READ)
    public Integer addDataAddTask(DataAddTaskRequest request, String operator) {
        FileDataAddTaskRequest sourceRequest = (FileDataAddTaskRequest) request;
        StreamSourceEntity sourceEntity = sourceMapper.selectById(request.getSourceId());
        try {
            List<StreamSourceEntity> dataAddTaskList = sourceMapper.selectByTaskMapId(sourceEntity.getId());
            FileSourceDTO dto = FileSourceDTO.getFromJson(sourceEntity.getExtParams());
            dto.setDataTimeFrom(sourceRequest.getDataTimeFrom());
            dto.setDataTimeTo(sourceRequest.getDataTimeTo());
            dto.setRetry(true);
            if (request.getIncreaseAuditVersion()) {
                dto.setAuditVersion(request.getAuditVersion());
            }
            dto.setFilterStreams(sourceRequest.getFilterStreams());
            StreamSourceEntity dataAddTaskEntity =
                    CommonBeanUtils.copyProperties(sourceEntity, StreamSourceEntity::new);
            dataAddTaskEntity.setId(null);
            dataAddTaskEntity.setSourceName(
                    sourceEntity.getSourceName() + "-" + request.getAuditVersion() + "-" + sourceEntity.getId());
            dataAddTaskEntity.setExtParams(objectMapper.writeValueAsString(dto));
            dataAddTaskEntity.setTaskMapId(sourceEntity.getId());
            Integer id = sourceMapper.insert(dataAddTaskEntity);
            SourceRequest dataAddTaskRequest =
                    CommonBeanUtils.copyProperties(dataAddTaskEntity, SourceRequest::new, true);
            updateAgentTaskConfig(dataAddTaskRequest, operator);
            return id;
        } catch (Exception e) {
            LOGGER.error("serialize extParams of File SourceDTO failure: ", e);
            throw new BusinessException(ErrorCodeEnum.SOURCE_INFO_INCORRECT,
                    String.format("serialize extParams of File SourceDTO failure: %s", e.getMessage()));
        }
    }
File source operator, such as get or set file source info.
inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/source/file/FileSourceOperator.java