[AMORO-2623] Avoid deadlock in task cancellation (apache#2684)

* [AMORO-2623] Avoid deadlock in task cancellation * fix: update cancelTasks invoke order
huyuanfeng2018 · Mar 28, 2024 · 326c1fe · 326c1fe
1 parent 3bf7104
commit 326c1fe
Showing 1 changed file with 26 additions and 30 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
@@ -421,6 +421,7 @@ public void close() {
       } finally {
         lock.unlock();
       }
+      cancelTasks();
     }
 
     @Override
@@ -467,6 +468,10 @@ public void acceptResult(TaskRuntime taskRuntime) {
       } finally {
         lock.unlock();
       }
+      // the cleanup of task should be done after unlock to avoid deadlock
+      if (this.status == OptimizingProcess.Status.FAILED) {
+        cancelTasks();
+      }
     }
 
     @Override
@@ -559,6 +564,9 @@ public void commit() {
       } finally {
         clearProcess(this);
         lock.unlock();
+        if (this.status == OptimizingProcess.Status.FAILED) {
+          cancelTasks();
+        }
       }
     }
 
@@ -624,36 +632,24 @@ private void beginAndPersistProcess() {
     }
 
     private void persistProcessCompleted(boolean success) {
-      if (!success) {
-        doAsTransaction(
-            () -> taskMap.values().forEach(TaskRuntime::tryCanceling),
-            () ->
-                doAs(
-                    OptimizingMapper.class,
-                    mapper ->
-                        mapper.updateOptimizingProcess(
-                            tableRuntime.getTableIdentifier().getId(),
-                            processId,
-                            status,
-                            endTime,
-                            getSummary(),
-                            getFailedReason())),
-            () -> tableRuntime.completeProcess(false));
-      } else {
-        doAsTransaction(
-            () ->
-                doAs(
-                    OptimizingMapper.class,
-                    mapper ->
-                        mapper.updateOptimizingProcess(
-                            tableRuntime.getTableIdentifier().getId(),
-                            processId,
-                            status,
-                            endTime,
-                            getSummary(),
-                            getFailedReason())),
-            () -> tableRuntime.completeProcess(true));
-      }
+      doAsTransaction(
+          () ->
+              doAs(
+                  OptimizingMapper.class,
+                  mapper ->
+                      mapper.updateOptimizingProcess(
+                          tableRuntime.getTableIdentifier().getId(),
+                          processId,
+                          status,
+                          endTime,
+                          getSummary(),
+                          getFailedReason())),
+          () -> tableRuntime.completeProcess(success));
+    }
+
+    /** The cancellation should be invoked outside the process lock to avoid deadlock. */
+    private void cancelTasks() {
+      taskMap.values().forEach(TaskRuntime::tryCanceling);
     }
 
     private void loadTaskRuntimes(OptimizingProcess optimizingProcess) {