[INLONG-11103][Manager] Data add task supports filtering based on stream

fuweng11 · Sep 14, 2024 · 5f82188 · 5f82188
1 parent 807717a
commit 5f82188
Show file tree

Hide file tree

Showing 8 changed files with 29 additions and 12 deletions.
diff --git a/...pojo/src/main/java/org/apache/inlong/manager/pojo/source/file/FileDataAddTaskRequest.java b/...pojo/src/main/java/org/apache/inlong/manager/pojo/source/file/FileDataAddTaskRequest.java
@@ -27,13 +27,18 @@
 import lombok.EqualsAndHashCode;
 import lombok.ToString;
 
+import java.util.List;
+
 @Data
 @ToString(callSuper = true)
 @EqualsAndHashCode(callSuper = true)
 @JsonTypeDefine(value = SourceType.FILE)
 @ApiModel(value = "File data add task request")
 public class FileDataAddTaskRequest extends DataAddTaskRequest {
 
+    @ApiModelProperty("filterStreams")
+    private List<String> filterStreams;
+
     @ApiModelProperty("Start time")
     private Long startTime;
 

diff --git a/...ger/manager-pojo/src/main/java/org/apache/inlong/manager/pojo/source/file/FileSource.java b/...ger/manager-pojo/src/main/java/org/apache/inlong/manager/pojo/source/file/FileSource.java
@@ -95,6 +95,9 @@ public class FileSource extends StreamSource {
     @ApiModelProperty("End time")
     private Long endTime;
 
+    @ApiModelProperty("filterStreams")
+    private List<String> filterStreams;
+
     public FileSource() {
         this.setSourceType(SourceType.FILE);
     }

diff --git a/.../manager-pojo/src/main/java/org/apache/inlong/manager/pojo/source/file/FileSourceDTO.java b/.../manager-pojo/src/main/java/org/apache/inlong/manager/pojo/source/file/FileSourceDTO.java
@@ -101,6 +101,9 @@ public class FileSourceDTO {
     @ApiModelProperty(value = "Audit version")
     private String auditVersion;
 
+    @ApiModelProperty("filterStreams")
+    private List<String> filterStreams;
+
     public static FileSourceDTO getFromRequest(@NotNull FileSourceRequest fileSourceRequest, String extParams) {
         FileSourceDTO dto = StringUtils.isNotBlank(extParams)
                 ? FileSourceDTO.getFromJson(extParams)

diff --git a/...ager-pojo/src/main/java/org/apache/inlong/manager/pojo/source/file/FileSourceRequest.java b/...ager-pojo/src/main/java/org/apache/inlong/manager/pojo/source/file/FileSourceRequest.java
@@ -90,6 +90,9 @@ public class FileSourceRequest extends SourceRequest {
     @ApiModelProperty("End time")
     private Long endTime;
 
+    @ApiModelProperty("filterStreams")
+    private List<String> filterStreams;
+
     public FileSourceRequest() {
         this.setSourceType(SourceType.FILE);
         this.setSerializationType(DataFormat.CSV.getName());

diff --git a/...r-service/src/main/java/org/apache/inlong/manager/service/source/StreamSourceService.java b/...r-service/src/main/java/org/apache/inlong/manager/service/source/StreamSourceService.java
@@ -220,6 +220,6 @@ default Boolean updateAfterApprove(String operator) {
      * @param operator Operator's name.
      * @return source id list after saving.
      */
-    List<Integer> batchAddDataAddTask(String groupId, String streamId, List<DataAddTaskRequest> requestList,
+    List<Integer> batchAddDataAddTask(String groupId, List<DataAddTaskRequest> requestList,
             String operator);
 }
diff --git a/...rvice/src/main/java/org/apache/inlong/manager/service/source/StreamSourceServiceImpl.java b/...rvice/src/main/java/org/apache/inlong/manager/service/source/StreamSourceServiceImpl.java
@@ -537,10 +537,10 @@ public Integer addDataAddTask(DataAddTaskRequest request, String operator) {
     }
 
     @Override
-    public List<Integer> batchAddDataAddTask(String groupId, String streamId, List<DataAddTaskRequest> requestList,
+    public List<Integer> batchAddDataAddTask(String groupId, List<DataAddTaskRequest> requestList,
             String operator) {
         List<Integer> result = new ArrayList<>();
-        String auditVersion = String.valueOf(sourceMapper.selectDataAddTaskCount(groupId, streamId));
+        String auditVersion = String.valueOf(sourceMapper.selectDataAddTaskCount(groupId, null));
         for (DataAddTaskRequest request : requestList) {
             request.setAuditVersion(auditVersion);
             int id = addDataAddTask(request, operator);

diff --git a/...rvice/src/main/java/org/apache/inlong/manager/service/source/file/FileSourceOperator.java b/...rvice/src/main/java/org/apache/inlong/manager/service/source/file/FileSourceOperator.java
@@ -119,13 +119,19 @@ public Integer addDataAddTask(DataAddTaskRequest request, String operator) {
             dto.setStartTime(sourceRequest.getStartTime());
             dto.setEndTime(sourceRequest.getEndTime());
             dto.setRetry(true);
+            dto.setAuditVersion(request.getAuditVersion());
+            dto.setFilterStreams(sourceRequest.getFilterStreams());
             StreamSourceEntity dataAddTaskEntity =
                     CommonBeanUtils.copyProperties(sourceEntity, StreamSourceEntity::new);
             dataAddTaskEntity.setId(null);
             dataAddTaskEntity.setSourceName(sourceEntity.getSourceName() + "-" + (dataAddTaskSize + 1));
             dataAddTaskEntity.setExtParams(objectMapper.writeValueAsString(dto));
             dataAddTaskEntity.setTaskMapId(sourceEntity.getId());
-            return sourceMapper.insert(dataAddTaskEntity);
+            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,

diff --git a/...er-web/src/main/java/org/apache/inlong/manager/web/controller/StreamSourceController.java b/...er-web/src/main/java/org/apache/inlong/manager/web/controller/StreamSourceController.java
@@ -130,16 +130,13 @@ public Response<Boolean> forceDelete(@RequestParam String inlongGroupId, @Reques
                 sourceService.forceDelete(inlongGroupId, inlongStreamId, LoginUserUtils.getLoginUser().getName()));
     }
 
-    @RequestMapping(value = "/source/addDataAddTask/{groupId}/{streamId}", method = RequestMethod.POST)
+    @RequestMapping(value = "/source/addDataAddTask/{groupId}", method = RequestMethod.POST)
     @ApiOperation(value = "Add supplementary recording task for stream source")
-    @ApiImplicitParams({
-            @ApiImplicitParam(name = "groupId", dataTypeClass = String.class, required = true),
-            @ApiImplicitParam(name = "streamId", dataTypeClass = String.class, required = true)
-    })
-    public Response<List<Integer>> addSub(@PathVariable String groupId, @PathVariable String streamId,
+    @ApiImplicitParam(name = "groupId", dataTypeClass = String.class, required = true)
+    public Response<List<Integer>> addSub(@PathVariable String groupId,
             @RequestBody List<DataAddTaskRequest> requestList) {
-        return Response.success(sourceService.batchAddDataAddTask(groupId, streamId, requestList,
-                LoginUserUtils.getLoginUser().getName()));
+        return Response.success(
+                sourceService.batchAddDataAddTask(groupId, requestList, LoginUserUtils.getLoginUser().getName()));
     }
 
 }