fix: file-worker任务状态更新请求无序到达导致第三方源文件偶现分发失败 TencentBlueKing#2434

统计子任务完成状态时加共享锁，防止应用统计结果的过程中数据被修改
jsonwan · Sep 15, 2023 · 3d3a640 · 3d3a640
1 parent 935de87
commit 3d3a640
Show file tree

Hide file tree

Showing 9 changed files with 29 additions and 28 deletions.
diff --git a/...teway/src/main/java/com/tencent/bk/job/file_gateway/dao/filesource/FileSourceTaskDAO.java b/...teway/src/main/java/com/tencent/bk/job/file_gateway/dao/filesource/FileSourceTaskDAO.java
@@ -41,7 +41,7 @@ public interface FileSourceTaskDAO {
 
     FileSourceTaskDTO getFileSourceTaskByIdForUpdate(String id);
 
-    Long countFileSourceTasksByBatchTaskId(String batchTaskId, Byte status);
+    Long countFileSourceTasksByBatchTaskIdForUpdate(String batchTaskId, Byte status);
 
     List<FileSourceTaskDTO> listByBatchTaskId(String batchTaskId);
 }
diff --git a/...ile-gateway/src/main/java/com/tencent/bk/job/file_gateway/dao/filesource/FileTaskDAO.java b/...ile-gateway/src/main/java/com/tencent/bk/job/file_gateway/dao/filesource/FileTaskDAO.java
@@ -47,5 +47,5 @@ List<String> listTimeoutFileSourceTaskIds(Long expireTimeMills, Collection<Byte>
 
     List<FileTaskDTO> listFileTasks(String fileSourceTaskId);
 
-    Long countFileTask(String fileSourceTaskId, Byte status);
+    Long countFileTaskForUpdate(String fileSourceTaskId, Byte status);
 }
diff --git a/.../main/java/com/tencent/bk/job/file_gateway/dao/filesource/impl/FileSourceTaskDAOImpl.java b/.../main/java/com/tencent/bk/job/file_gateway/dao/filesource/impl/FileSourceTaskDAOImpl.java
@@ -190,23 +190,23 @@ val record = dslContext.selectFrom(defaultTable).where(
     }
 
     @Override
-    public Long countFileSourceTasksByBatchTaskId(String batchTaskId, Byte status) {
+    public Long countFileSourceTasksByBatchTaskIdForUpdate(String batchTaskId, Byte status) {
         List<Condition> conditions = new ArrayList<>();
         if (batchTaskId != null) {
             conditions.add(defaultTable.BATCH_TASK_ID.eq(batchTaskId));
         }
         if (status != null) {
             conditions.add(defaultTable.STATUS.eq(status));
         }
-        return countFileSourceTasksByConditions(conditions);
+        return countFileSourceTasksByConditionsForUpdate(conditions);
     }
 
-    public Long countFileSourceTasksByConditions(Collection<Condition> conditions) {
+    public Long countFileSourceTasksByConditionsForUpdate(Collection<Condition> conditions) {
         val query = dslContext.select(
             DSL.countDistinct(defaultTable.ID)
         ).from(defaultTable)
             .where(conditions);
-        return query.fetchOne(0, Long.class);
+        return query.forUpdate().fetchOne(0, Long.class);
     }
 
     public List<FileSourceTaskDTO> listByConditions(Collection<Condition> conditions,

diff --git a/...ay/src/main/java/com/tencent/bk/job/file_gateway/dao/filesource/impl/FileTaskDAOImpl.java b/...ay/src/main/java/com/tencent/bk/job/file_gateway/dao/filesource/impl/FileTaskDAOImpl.java
@@ -149,12 +149,12 @@ val record = dslContext.selectFrom(defaultTable).where(
         }
     }
 
-    public Long countFileTasksByConditions(Collection<Condition> conditions) {
+    public Long countFileTasksByConditionsForUpdate(Collection<Condition> conditions) {
         val query = dslContext.select(
             DSL.countDistinct(defaultTable.ID)
         ).from(defaultTable)
             .where(conditions);
-        return query.fetchOne(0, Long.class);
+        return query.forUpdate().fetchOne(0, Long.class);
     }
 
     @Override
@@ -196,15 +196,15 @@ public List<FileTaskDTO> listFileTasks(String fileSourceTaskId) {
     }
 
     @Override
-    public Long countFileTask(String fileSourceTaskId, Byte status) {
+    public Long countFileTaskForUpdate(String fileSourceTaskId, Byte status) {
         List<Condition> conditions = new ArrayList<>();
         if (fileSourceTaskId != null) {
             conditions.add(defaultTable.FILE_SOURCE_TASK_ID.eq(fileSourceTaskId));
         }
         if (status != null) {
             conditions.add(defaultTable.STATUS.eq(status));
         }
-        return countFileTasksByConditions(conditions);
+        return countFileTasksByConditionsForUpdate(conditions);
     }
 
     private FileTaskDTO convertRecordToDto(FileTaskRecord record) {

diff --git a/...src/main/java/com/tencent/bk/job/file_gateway/service/impl/FileSourceTaskServiceImpl.java b/...src/main/java/com/tencent/bk/job/file_gateway/service/impl/FileSourceTaskServiceImpl.java
@@ -228,8 +228,10 @@ private void writeLog(FileSourceTaskDTO fileSourceTaskDTO,
         redisTemplate.expireAt(PREFIX_REDIS_TASK_LOG + taskId, new Date(System.currentTimeMillis() + 3600 * 1000));
     }
 
-    private void notifyFileTaskStatusChangeListeners(FileTaskDTO fileTaskDTO, FileSourceTaskDTO fileSourceTaskDTO,
-                                                     FileWorkerDTO fileWorkerDTO, TaskStatusEnum previousStatus,
+    private void notifyFileTaskStatusChangeListeners(FileTaskDTO fileTaskDTO,
+                                                     FileSourceTaskDTO fileSourceTaskDTO,
+                                                     FileWorkerDTO fileWorkerDTO,
+                                                     TaskStatusEnum previousStatus,
                                                      TaskStatusEnum currentStatus) {
         TaskContext context = new DefaultTaskContext(fileTaskDTO, fileSourceTaskDTO, fileWorkerDTO);
         if (!fileTaskStatusChangeListenerList.isEmpty()) {
@@ -293,8 +295,7 @@ public String updateFileSourceTask(String taskId, String filePath, String downlo
         }
         // 通知关注者
         if (status != previousStatus) {
-            notifyFileTaskStatusChangeListeners(fileTaskDTO, fileSourceTaskDTO, fileWorkerDTO,
-                TaskStatusEnum.valueOf(fileTaskDTO.getStatus()), status);
+            notifyFileTaskStatusChangeListeners(fileTaskDTO, fileSourceTaskDTO, fileWorkerDTO, previousStatus, status);
         }
         // 进度上报
         writeLog(fileSourceTaskDTO, fileWorkerDTO, filePath, fileSize, speed, progress, content);

diff --git a/.../com/tencent/bk/job/file_gateway/service/listener/FileSourceTaskStatusChangeListener.java b/.../com/tencent/bk/job/file_gateway/service/listener/FileSourceTaskStatusChangeListener.java
@@ -29,12 +29,12 @@
 
 public interface FileSourceTaskStatusChangeListener {
     /**
-     * Invoked when FileSourceTask status change
+     * 在单个文件源任务状态改变时被调用
      *
-     * @param context
-     * @param previousStatus
-     * @param currentStatus
-     * @return true if need to stop event dispatch to later listeners
+     * @param context        任务上下文
+     * @param previousStatus 上一个状态
+     * @param currentStatus  当前状态
+     * @return true 如果需要终止事件向后续的Listener传递则返回true
      */
     boolean onStatusChange(TaskContext context, TaskStatusEnum previousStatus, TaskStatusEnum currentStatus);
 }
diff --git a/...n/java/com/tencent/bk/job/file_gateway/service/listener/FileTaskStatusChangeListener.java b/...n/java/com/tencent/bk/job/file_gateway/service/listener/FileTaskStatusChangeListener.java
@@ -29,12 +29,12 @@
 
 public interface FileTaskStatusChangeListener {
     /**
-     * Invoked when FileTask status change
+     * 在单个文件任务状态改变时被调用
      *
-     * @param context
-     * @param previousStatus
-     * @param currentStatus
-     * @return true if need to stop event dispatch to later listeners
+     * @param context        任务上下文
+     * @param previousStatus 上一个状态
+     * @param currentStatus  当前状态
+     * @return true 如果需要终止事件向后续的Listener传递则返回true
      */
     boolean onStatusChange(TaskContext context, TaskStatusEnum previousStatus, TaskStatusEnum currentStatus);
 }
diff --git a/...in/java/com/tencent/bk/job/file_gateway/service/listener/impl/BatchTaskStatusUpdater.java b/...in/java/com/tencent/bk/job/file_gateway/service/listener/impl/BatchTaskStatusUpdater.java
@@ -65,8 +65,8 @@ public boolean onStatusChange(TaskContext context, TaskStatusEnum previousStatus
         }
         if (TaskStatusEnum.SUCCESS.equals(currentStatus)) {
             //检查批量任务是否成功
-            if (fileSourceTaskDAO.countFileSourceTasksByBatchTaskId(batchTaskId, null)
-                .equals(fileSourceTaskDAO.countFileSourceTasksByBatchTaskId(
+            if (fileSourceTaskDAO.countFileSourceTasksByBatchTaskIdForUpdate(batchTaskId, null)
+                .equals(fileSourceTaskDAO.countFileSourceTasksByBatchTaskIdForUpdate(
                     batchTaskId, TaskStatusEnum.SUCCESS.getStatus()))) {
                 // 批量任务成功
                 fileSourceBatchTaskDTO.setStatus(TaskStatusEnum.SUCCESS.getStatus());

diff --git a/...va/com/tencent/bk/job/file_gateway/service/listener/impl/FileSourceTaskStatusUpdater.java b/...va/com/tencent/bk/job/file_gateway/service/listener/impl/FileSourceTaskStatusUpdater.java
@@ -93,8 +93,8 @@ public boolean onStatusChange(TaskContext context, TaskStatusEnum previousStatus
             }
         } else if (TaskStatusEnum.SUCCESS.equals(currentStatus)) {
             //检查主任务是否成功
-            if (fileTaskDAO.countFileTask(fileSourceTaskId, null).equals(fileTaskDAO.countFileTask(fileSourceTaskId,
-                TaskStatusEnum.SUCCESS.getStatus()))) {
+            if (fileTaskDAO.countFileTaskForUpdate(fileSourceTaskId, null).equals(
+                fileTaskDAO.countFileTaskForUpdate(fileSourceTaskId, TaskStatusEnum.SUCCESS.getStatus()))) {
                 // 主任务成功
                 fileSourceTaskDTO.setStatus(TaskStatusEnum.SUCCESS.getStatus());
                 fileSourceTaskDAO.updateFileSourceTask(fileSourceTaskDTO);