fix(dlm): task status is wrong (#3491)

* fix the issue of wrong task status * format * fix the issue of throw not found exception when database is deleted. * simplify code
oceanbase · Sep 23, 2024 · acf70aa · acf70aa
1 parent 2da088b
commit acf70aa
Show file tree

Hide file tree

Showing 4 changed files with 35 additions and 21 deletions.
diff --git a/server/odc-service/src/main/java/com/oceanbase/odc/service/dlm/DLMService.java b/server/odc-service/src/main/java/com/oceanbase/odc/service/dlm/DLMService.java
@@ -116,21 +116,28 @@ public List<DlmTableUnit> findByScheduleTaskId(Long scheduleTaskId) {
                         Collectors.toList());
     }
 
+    /**
+     * generate final task status by scheduleTaskId when the task is finished
+     */
     @SkipAuthorize("odc internal usage")
-    public TaskStatus getTaskStatus(Long scheduleTaskId) {
-        return getTaskStatus(findByScheduleTaskId(scheduleTaskId));
-    }
-
-    public TaskStatus getTaskStatus(List<DlmTableUnit> dlmTableUnits) {
+    public TaskStatus getFinalTaskStatus(Long scheduleTaskId) {
+        List<DlmTableUnit> dlmTableUnits = findByScheduleTaskId(scheduleTaskId);
         Set<TaskStatus> collect = dlmTableUnits.stream().map(DlmTableUnit::getStatus).collect(
                 Collectors.toSet());
+        // If any table fails, the task is considered a failure.
         if (collect.contains(TaskStatus.FAILED)) {
             return TaskStatus.FAILED;
         }
-        if (collect.contains(TaskStatus.DONE) && collect.size() == 1) {
-            return TaskStatus.DONE;
+        // If any table is canceled, the task is considered canceled.
+        if (collect.contains(TaskStatus.CANCELED)) {
+            return TaskStatus.CANCELED;
+        }
+        // The task is considered failed if any table is still preparing or running when the task is
+        // finished.
+        if (collect.contains(TaskStatus.PREPARING) || collect.contains(TaskStatus.RUNNING)) {
+            return TaskStatus.FAILED;
         }
-        return TaskStatus.CANCELED;
+        return TaskStatus.DONE;
     }
 
 }
diff --git a/...c/main/java/com/oceanbase/odc/service/schedule/factory/ScheduleResponseMapperFactory.java b/...c/main/java/com/oceanbase/odc/service/schedule/factory/ScheduleResponseMapperFactory.java
@@ -34,6 +34,7 @@
 import com.oceanbase.odc.common.json.JsonUtils;
 import com.oceanbase.odc.core.shared.constant.FlowStatus;
 import com.oceanbase.odc.core.shared.constant.TaskType;
+import com.oceanbase.odc.core.shared.exception.NotFoundException;
 import com.oceanbase.odc.metadb.flow.FlowInstanceEntity;
 import com.oceanbase.odc.metadb.flow.FlowInstanceRepository;
 import com.oceanbase.odc.metadb.flow.UserTaskInstanceEntity;
@@ -195,11 +196,7 @@ public ScheduleDetailRespHist generateHistoryScheduleDetail(Schedule schedule) {
         resp.setUpdateTime(schedule.getUpdateTime());
         resp.setDescription(schedule.getDescription());
         resp.setJobParameters(detailParameters(schedule));
-
-        List<Database> databaseByIds = getDatabaseByIds(Collections.singleton(schedule.getDatabaseId()));
-        if (!databaseByIds.isEmpty()) {
-            resp.setDatabase(databaseByIds.get(0));
-        }
+        resp.setDatabase(detailDatabaseOrNull(schedule.getDatabaseId()));
 
         Set<Long> approvableFlowInstanceIds = approvalPermissionService.getApprovableApprovalInstances()
                 .stream()
@@ -387,29 +384,37 @@ private ScheduleTaskParameters detailParameters(Schedule schedule) {
         switch (schedule.getType()) {
             case DATA_ARCHIVE: {
                 DataArchiveParameters parameters = (DataArchiveParameters) schedule.getParameters();
-                parameters.setSourceDatabase(databaseService.detail(parameters.getSourceDatabaseId()));
-                parameters.setTargetDatabase(databaseService.detail(parameters.getTargetDataBaseId()));
+                parameters.setSourceDatabase(detailDatabaseOrNull(parameters.getSourceDatabaseId()));
+                parameters.setTargetDatabase(detailDatabaseOrNull(parameters.getTargetDataBaseId()));
                 limiterService.findByScheduleId(schedule.getId()).ifPresent(parameters::setRateLimit);
                 return parameters;
             }
             case DATA_DELETE: {
                 DataDeleteParameters parameters = (DataDeleteParameters) schedule.getParameters();
                 if (parameters.getTargetDatabaseId() != null) {
-                    parameters.setTargetDatabase(databaseService.detail(parameters.getTargetDatabaseId()));
+                    parameters.setTargetDatabase(detailDatabaseOrNull(parameters.getTargetDatabaseId()));
                 }
-                parameters.setDatabase(databaseService.detail(parameters.getDatabaseId()));
+                parameters.setDatabase(detailDatabaseOrNull(parameters.getDatabaseId()));
                 limiterService.findByScheduleId(schedule.getId()).ifPresent(parameters::setRateLimit);
                 return parameters;
             }
             case SQL_PLAN: {
                 SqlPlanParameters parameters = (SqlPlanParameters) schedule.getParameters();
                 if (parameters.getDatabaseId() != null) {
-                    parameters.setDatabaseInfo(databaseService.detail(parameters.getDatabaseId()));
+                    parameters.setDatabaseInfo(detailDatabaseOrNull(parameters.getDatabaseId()));
                 }
                 return parameters;
             }
             default:
                 return schedule.getParameters();
         }
     }
+
+    private Database detailDatabaseOrNull(Long databaseId) {
+        try {
+            return databaseService.detail(databaseId);
+        } catch (NotFoundException e) {
+            return null;
+        }
+    }
 }
diff --git a/...ce/src/main/java/com/oceanbase/odc/service/task/listener/DefaultJobTerminateListener.java b/...ce/src/main/java/com/oceanbase/odc/service/task/listener/DefaultJobTerminateListener.java
@@ -63,7 +63,7 @@ public class DefaultJobTerminateListener extends AbstractEventListener<JobTermin
     public void onEvent(JobTerminateEvent event) {
         JobEntity jobEntity = taskFrameworkService.find(event.getJi().getId());
         scheduleTaskService.findByJobId(jobEntity.getId()).ifPresent(o -> {
-            TaskStatus taskStatus = "DLM".equals(jobEntity.getJobType()) ? dlmService.getTaskStatus(o.getId())
+            TaskStatus taskStatus = "DLM".equals(jobEntity.getJobType()) ? dlmService.getFinalTaskStatus(o.getId())
                     : event.getStatus().convertTaskStatus();
             scheduleTaskService.updateStatusById(o.getId(), taskStatus);
             log.info("Update schedule task status to {} succeed,scheduleTaskId={}", taskStatus, o.getId());

diff --git a/...dc-service/src/main/java/com/oceanbase/odc/service/task/processor/DLMResultProcessor.java b/...dc-service/src/main/java/com/oceanbase/odc/service/task/processor/DLMResultProcessor.java
@@ -60,8 +60,10 @@ public void process(TaskResult result) {
             log.info("Create or update dlm tableUnits success,jobIdentity={},scheduleTaskId={}",
                     result.getJobIdentity(),
                     dlmTableUnits.get(0).getScheduleTaskId());
-            TaskStatus taskStatus = dlmService.getTaskStatus(dlmTableUnits);
-            taskService.updateStatusById(dlmTableUnits.get(0).getScheduleTaskId(), taskStatus);
+            TaskStatus taskStatus = taskService.nullSafeGetById(dlmTableUnits.get(0).getScheduleTaskId()).getStatus();
+            if (taskStatus != TaskStatus.RUNNING) {
+                taskService.updateStatusById(dlmTableUnits.get(0).getScheduleTaskId(), TaskStatus.RUNNING);
+            }
             log.info("Update schedule task status to {} success", taskStatus);
         } catch (Exception e) {
             log.warn("Refresh result failed.", e);