fix some checkpoint err

apache · Sep 14, 2023 · 8399565 · 8399565
1 parent b975648
commit 8399565
Showing 1 changed file with 8 additions and 7 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
@@ -406,24 +406,25 @@ protected void tryTriggerPendingCheckpoint(CheckpointType checkpointType) {
                                 shutdown));
                 return;
             }
-            if (checkpointType.isFinalCheckpoint() || checkpointType.isSchemaChangeCheckpoint()) {
-                if (pendingCounter.get() > 0) {
-                    scheduleTriggerPendingCheckpoint(checkpointType, 500L);
-                    return;
-                }
-            }
 
             if (schemaChanging.get() && checkpointType.isGeneralCheckpoint()) {
                 LOG.info("skip trigger generic-checkpoint because schema change in progress");
                 return;
             }
 
+            if (pendingCounter.get() > 0) {
+                scheduleTriggerPendingCheckpoint(checkpointType, 500L);
+                LOG.info(
+                        "skip trigger checkpoint because there is already a pending checkpoint.");
+                return;
+            }
+
             CompletableFuture<PendingCheckpoint> pendingCheckpoint =
                     createPendingCheckpoint(currentTimestamp, checkpointType);
             startTriggerPendingCheckpoint(pendingCheckpoint);
             pendingCounter.incrementAndGet();
             // if checkpoint type are final type, we don't need to trigger next checkpoint
-            if (checkpointType.notFinalCheckpoint() && checkpointType.notSchemaChangeCheckpoint()) {
+            if (checkpointType.notFinalCheckpoint()) {
                 scheduleTriggerPendingCheckpoint(coordinatorConfig.getCheckpointInterval());
             }
         }