[Improve][Zeta] Move checkpoint notify complete in checkpoint stage (#…

…5185)
apache · Jul 31, 2023 · bc28eb1 · bc28eb1
1 parent 1db9f45
commit bc28eb1
Showing 1 changed file with 8 additions and 8 deletions.
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
@@ -695,14 +695,6 @@ public synchronized void completePendingCheckpoint(CompletedCheckpoint completed
                 completedCheckpoint.getCheckpointTimestamp(),
                 completedCheckpoint.getCompletedTimestamp());
         final long checkpointId = completedCheckpoint.getCheckpointId();
-        pendingCheckpoints.remove(checkpointId);
-        pendingCounter.decrementAndGet();
-        if (pendingCheckpoints.size() + 1 == coordinatorConfig.getMaxConcurrentCheckpoints()) {
-            // latest checkpoint completed time > checkpoint interval
-            if (notFinalCheckpoint(completedCheckpoint.getCheckpointType())) {
-                scheduleTriggerPendingCheckpoint(0L);
-            }
-        }
         completedCheckpoints.addLast(completedCheckpoint);
         try {
             byte[] states = serializer.serialize(completedCheckpoint);
@@ -742,6 +734,14 @@ public synchronized void completePendingCheckpoint(CompletedCheckpoint completed
                 completedCheckpoint.getJobId());
         latestCompletedCheckpoint = completedCheckpoint;
         notifyCompleted(completedCheckpoint);
+        pendingCheckpoints.remove(checkpointId);
+        pendingCounter.decrementAndGet();
+        if (pendingCheckpoints.size() + 1 == coordinatorConfig.getMaxConcurrentCheckpoints()) {
+            // latest checkpoint completed time > checkpoint interval
+            if (notFinalCheckpoint(completedCheckpoint.getCheckpointType())) {
+                scheduleTriggerPendingCheckpoint(0L);
+            }
+        }
         if (isCompleted()) {
             cleanPendingCheckpoint(CheckpointCloseReason.CHECKPOINT_COORDINATOR_COMPLETED);
             if (latestCompletedCheckpoint.getCheckpointType().equals(SAVEPOINT_TYPE)) {