restart

CalvinKirs · Dec 13, 2024 · d1e2998 · d1e2998
1 parent 409c21d
commit d1e2998
Show file tree

Hide file tree

Showing 2 changed files with 18 additions and 6 deletions.
diff --git a/fe/fe-core/src/main/java/org/apache/doris/job/base/AbstractJob.java b/fe/fe-core/src/main/java/org/apache/doris/job/base/AbstractJob.java
@@ -461,15 +461,26 @@ public void onReplayCreate() throws JobException {
     public void onReplayEnd(AbstractJob<?, C> replayJob) throws JobException {
         log.info(new LogBuilder(LogKey.SCHEDULER_JOB, getJobId()).add("msg", "replay delete scheduler job").build());
     }
-
-    private void updateTaskStatusAfterRestart() {
+
+    public void updateTaskStatusAfterRestart() {
+        List<T> tasks = queryAllTasks();
+        if (CollectionUtils.isEmpty(tasks)) {
+            return;
+        }
+        List<T> runningTasks = tasks.stream().filter(task -> task.getStatus().equals(TaskStatus.RUNNING) || task.getStatus().equals(TaskStatus.PENDING))
+                .collect(Collectors.toList());
         if (CollectionUtils.isEmpty(runningTasks)) {
             return;
         }
+
         runningTasks.forEach(task -> {
-            if (task.getStatus().equals(TaskStatus.RUNNING)) {
-                task.setStatus(TaskStatus.PENDING);
+            try {
+                task.onFail("task failed because of restart");
+            } catch (JobException e) {
+                log.warn("task failed because of restart, job id is {}, task id is {}",
+                        jobId, task.getTaskId(), e);
             }
         });
+
     }
 }
diff --git a/fe/fe-core/src/main/java/org/apache/doris/job/manager/JobManager.java b/fe/fe-core/src/main/java/org/apache/doris/job/manager/JobManager.java
@@ -89,14 +89,15 @@ private void writeUnlock() {
 
     public void start() {
         jobScheduler = new JobScheduler<T, C>(jobMap);
+        clearTaskStatusWhenFeRestart();
         jobScheduler.start();
     }
 
-    private void clearTaskStatusWhenFeRestart() {
+    public void clearTaskStatusWhenFeRestart() {
         List<T> runningJobs = jobMap.values().stream()
                 .filter(job -> job.getJobStatus().equals(JobStatus.RUNNING)).collect(Collectors.toList());
         for (T job : runningJobs) {
-
+            job.updateTaskStatusAfterRestart();
         }
     }