[AMORO-2279] Retry failed tasks when restarting AMS (#2280)

apache · Nov 13, 2023 · 596574a · 596574a
1 parent 6f834ef
commit 596574a
Showing 1 changed file with 17 additions and 9 deletions.
diff --git a/ams/server/src/main/java/com/netease/arctic/server/optimizing/OptimizingQueue.java b/ams/server/src/main/java/com/netease/arctic/server/optimizing/OptimizingQueue.java
@@ -122,15 +122,23 @@ private void initTableRuntime(TableRuntimeMeta tableRuntimeMeta) {
         schedulingPolicy.addTable(tableRuntime);
       } else if (tableRuntime.getOptimizingStatus() != OptimizingStatus.COMMITTING) {
         TableOptimizingProcess process = new TableOptimizingProcess(tableRuntimeMeta);
-        process.getTaskMap().entrySet().stream()
-            .filter(
-                entry ->
-                    entry.getValue().getStatus() == TaskRuntime.Status.SCHEDULED
-                        || entry.getValue().getStatus() == TaskRuntime.Status.ACKED)
-            .forEach(entry -> executingTaskMap.put(entry.getKey(), entry.getValue()));
-        process.getTaskMap().values().stream()
-            .filter(task -> task.getStatus() == TaskRuntime.Status.PLANNED)
-            .forEach(taskQueue::offer);
+        process
+            .getTaskMap()
+            .forEach(
+                (taskId, taskRuntime) -> {
+                  switch (taskRuntime.getStatus()) {
+                    case SCHEDULED:
+                    case ACKED:
+                      executingTaskMap.put(taskId, taskRuntime);
+                      break;
+                    case PLANNED:
+                      taskQueue.offer(taskRuntime);
+                      break;
+                    case FAILED:
+                      retryTask(taskRuntime, false);
+                      break;
+                  }
+                });
       }
     } else {
       OptimizingProcess process = tableRuntime.getOptimizingProcess();