[Improve] Improve CheckpointCoordinator notify complete when restore

apache · Jul 22, 2023 · 5e68b5a · 5e68b5a
1 parent 5e8d982
commit 5e68b5a
Show file tree

Hide file tree

Showing 4 changed files with 30 additions and 8 deletions.
diff --git a/...er/src/main/java/org/apache/seatunnel/engine/server/checkpoint/CheckpointCloseReason.java b/...er/src/main/java/org/apache/seatunnel/engine/server/checkpoint/CheckpointCloseReason.java
@@ -26,7 +26,8 @@ public enum CheckpointCloseReason {
     CHECKPOINT_COORDINATOR_RESET("CheckpointCoordinator reset."),
     CHECKPOINT_INSIDE_ERROR("CheckpointCoordinator inside have error."),
     AGGREGATE_COMMIT_ERROR("Aggregate commit error."),
-    TASK_NOT_ALL_READY_WHEN_SAVEPOINT("Task not all ready, savepoint error");
+    TASK_NOT_ALL_READY_WHEN_SAVEPOINT("Task not all ready, savepoint error"),
+    CHECKPOINT_NOTIFY_COMPLETE_FAILED("Checkpoint notify complete failed");
 
     private final String message;
 

diff --git a/...er/src/main/java/org/apache/seatunnel/engine/server/checkpoint/CheckpointCoordinator.java b/...er/src/main/java/org/apache/seatunnel/engine/server/checkpoint/CheckpointCoordinator.java
@@ -272,7 +272,8 @@ private void handleCoordinatorError(CheckpointCloseReason reason, Throwable e) {
         checkpointCoordinatorFuture.complete(
                 new CheckpointCoordinatorState(
                         CheckpointCoordinatorStatus.FAILED, errorByPhysicalVertex.get()));
-        checkpointManager.handleCheckpointError(pipelineId);
+        checkpointManager.handleCheckpointError(
+                pipelineId, reason.equals(CheckpointCloseReason.CHECKPOINT_NOTIFY_COMPLETE_FAILED));
     }
 
     private void restoreTaskState(TaskLocation taskLocation) {
@@ -316,9 +317,26 @@ private void allTaskReady() {
         isAllTaskReady = true;
         InvocationFuture<?>[] futures = notifyTaskStart();
         CompletableFuture.allOf(futures).join();
+        notifyCompleted(latestCompletedCheckpoint);
         scheduleTriggerPendingCheckpoint(coordinatorConfig.getCheckpointInterval());
     }
 
+    private void notifyCompleted(CompletedCheckpoint completedCheckpoint) {
+        if (completedCheckpoint != null) {
+            try {
+                LOG.info("start notify checkpoint completed, checkpoint:{}", completedCheckpoint);
+                InvocationFuture<?>[] invocationFutures =
+                        notifyCheckpointCompleted(completedCheckpoint.getCheckpointId());
+                CompletableFuture.allOf(invocationFutures).join();
+            } catch (Throwable e) {
+                handleCoordinatorError(
+                        "notify checkpoint completed failed",
+                        e,
+                        CheckpointCloseReason.CHECKPOINT_NOTIFY_COMPLETE_FAILED);
+            }
+        }
+    }
+
     public InvocationFuture<?>[] notifyTaskStart() {
         return plan.getPipelineSubtasks().stream()
                 .map(NotifyTaskStartOperation::new)
@@ -358,6 +376,7 @@ protected void restoreCoordinator(boolean alreadyStarted) {
         shutdown = false;
         if (alreadyStarted) {
             isAllTaskReady = true;
+            notifyCompleted(latestCompletedCheckpoint);
             tryTriggerPendingCheckpoint(CHECKPOINT_TYPE);
         } else {
             isAllTaskReady = false;
@@ -719,10 +738,9 @@ public synchronized void completePendingCheckpoint(CompletedCheckpoint completed
                 completedCheckpoint.getCheckpointId(),
                 completedCheckpoint.getPipelineId(),
                 completedCheckpoint.getJobId());
-        InvocationFuture<?>[] invocationFutures = notifyCheckpointCompleted(checkpointId);
-        CompletableFuture.allOf(invocationFutures).join();
-        // TODO: notifyCheckpointCompleted fail
+
         latestCompletedCheckpoint = completedCheckpoint;
+        notifyCompleted(completedCheckpoint);
         if (isCompleted()) {
             cleanPendingCheckpoint(CheckpointCloseReason.CHECKPOINT_COORDINATOR_COMPLETED);
             if (latestCompletedCheckpoint.getCheckpointType().equals(SAVEPOINT_TYPE)) {

diff --git a/...server/src/main/java/org/apache/seatunnel/engine/server/checkpoint/CheckpointManager.java b/...server/src/main/java/org/apache/seatunnel/engine/server/checkpoint/CheckpointManager.java
@@ -174,8 +174,8 @@ public void reportedPipelineRunning(int pipelineId, boolean alreadyStarted) {
         getCheckpointCoordinator(pipelineId).restoreCoordinator(alreadyStarted);
     }
 
-    protected void handleCheckpointError(int pipelineId) {
-        jobMaster.handleCheckpointError(pipelineId);
+    protected void handleCheckpointError(int pipelineId, boolean neverRestore) {
+        jobMaster.handleCheckpointError(pipelineId, neverRestore);
     }
 
     private CheckpointCoordinator getCheckpointCoordinator(TaskLocation taskLocation) {

diff --git a/...nnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/master/JobMaster.java b/...nnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/master/JobMaster.java
@@ -325,7 +325,10 @@ public void run() {
         }
     }
 
-    public void handleCheckpointError(long pipelineId) {
+    public void handleCheckpointError(long pipelineId, boolean neverRestore) {
+        if (neverRestore) {
+            this.neverNeedRestore();
+        }
         this.physicalPlan
                 .getPipelineList()
                 .forEach(