[Improve] [SeaTunnel-Engine] Cancel CheckpointCoordinator First Befor…

…e Cancel Task (apache#3838) * [Improve] [SeaTunnel-Engine] Cancel CheckpointCoordinator First Before Cancel Task
lhyundeadsoul · Jan 3, 2023 · 871f0b8 · 871f0b8
1 parent e60f529
commit 871f0b8
Show file tree

Hide file tree

Showing 3 changed files with 17 additions and 15 deletions.
diff --git a/...-engine-server/src/main/java/org/apache/seatunnel/engine/server/TaskExecutionService.java b/...-engine-server/src/main/java/org/apache/seatunnel/engine/server/TaskExecutionService.java
@@ -268,8 +268,7 @@ private void notifyTaskStatusToMaster(TaskGroupLocation taskGroupLocation, TaskE
                 invoke.get();
                 notifyStateSuccess = true;
             } catch (InterruptedException e) {
-                logger.severe(e);
-                Thread.interrupted();
+                logger.severe("send notify task status failed", e);
             } catch (ExecutionException e) {
                 logger.warning(ExceptionUtils.getMessage(e));
                 logger.warning(String.format("notify the job of the task(%s) status failed, retry in %s millis",
@@ -278,7 +277,6 @@ private void notifyTaskStatusToMaster(TaskGroupLocation taskGroupLocation, TaskE
                     Thread.sleep(sleepTime);
                 } catch (InterruptedException ex) {
                     logger.severe(e);
-                    Thread.interrupted();
                 }
             }
         }

diff --git a/...-server/src/main/java/org/apache/seatunnel/engine/server/dag/physical/PhysicalVertex.java b/...-server/src/main/java/org/apache/seatunnel/engine/server/dag/physical/PhysicalVertex.java
@@ -306,20 +306,23 @@ private TaskGroupImmutableInformation getTaskGroupImmutableInformation() {
 
     private boolean turnToEndState(@NonNull ExecutionState endState) {
         synchronized (this) {
-            // consistency check
-            ExecutionState currentState = (ExecutionState) runningJobStateIMap.get(taskGroupLocation);
-            if (currentState.isEndState()) {
-                String message = String.format("Task %s is already in terminal state %s", taskFullName, currentState);
-                LOGGER.warning(message);
-                return false;
-            }
             if (!endState.isEndState()) {
                 String message =
                     String.format("Turn task %s state to end state need gave a end state, not %s", taskFullName,
                         endState);
                 LOGGER.warning(message);
                 return false;
             }
+            // consistency check
+            ExecutionState currentState = (ExecutionState) runningJobStateIMap.get(taskGroupLocation);
+            if (currentState.equals(endState)) {
+                return true;
+            }
+            if (currentState.isEndState()) {
+                String message = String.format("Task %s is already in terminal state %s", taskFullName, currentState);
+                LOGGER.warning(message);
+                return false;
+            }
 
             updateStateTimestamps(endState);
             runningJobStateIMap.set(taskGroupLocation, endState);
@@ -372,10 +375,6 @@ public boolean updateTaskState(@NonNull ExecutionState current, @NonNull Executi
         }
     }
 
-    public TaskGroupDefaultImpl getTaskGroup() {
-        return taskGroup;
-    }
-
     public void cancel() {
         if (updateTaskState(ExecutionState.CREATED, ExecutionState.CANCELED) ||
             updateTaskState(ExecutionState.SCHEDULED, ExecutionState.CANCELED) ||
@@ -392,6 +391,7 @@ private void noticeTaskExecutionServiceCancel() {
         if (!checkTaskGroupIsExecuting(taskGroupLocation)){
             updateTaskState(ExecutionState.CANCELING, ExecutionState.CANCELED);
             taskFuture.complete(new TaskExecutionState(this.taskGroupLocation, ExecutionState.CANCELED, null));
+            return;
         }
         int i = 0;
         // In order not to generate uncontrolled tasks, We will try again until the taskFuture is completed
@@ -415,7 +415,6 @@ private void noticeTaskExecutionServiceCancel() {
                 }
             }
         }
-        this.taskFuture.complete(new TaskExecutionState(taskGroupLocation, ExecutionState.CANCELED, null));
     }
 
     private void updateStateTimestamps(@NonNull ExecutionState targetState) {

diff --git a/...-engine-server/src/main/java/org/apache/seatunnel/engine/server/dag/physical/SubPlan.java b/...-engine-server/src/main/java/org/apache/seatunnel/engine/server/dag/physical/SubPlan.java
@@ -247,9 +247,14 @@ public void cancelPipeline() {
         if (!PipelineStatus.CANCELING.equals(runningJobStateIMap.get(pipelineLocation))) {
             updatePipelineState(getPipelineState(), PipelineStatus.CANCELING);
         }
+        cancelCheckpointCoordinator();
         cancelPipelineTasks();
     }
 
+    private void cancelCheckpointCoordinator() {
+        jobMaster.getCheckpointManager().listenPipelineRetry(pipelineId, PipelineStatus.CANCELING).join();
+    }
+
     private void cancelPipelineTasks() {
         List<CompletableFuture<Void>> coordinatorCancelList =
             coordinatorVertexList.stream().map(this::cancelTask).filter(Objects::nonNull)