[Feature][Zeta] Support the tolerable-failed configuration for checkpoints

docs/en/seatunnel-engine/hybrid-cluster-deployment.md

@@ -93,6 +93,10 @@ The timeout for checkpoints. If the checkpoint cannot be completed within the ti
 
 The minimum pause (in milliseconds) between consecutive checkpoints. This ensures that checkpoints are not triggered too frequently.
 
+**tolerable-failed**
+
+The number of consecutive checkpoint failures that can be tolerated before the job fails.
+
 Example
 
 ```yaml
@@ -106,6 +110,7 @@ seatunnel:
             interval: 300000
             timeout: 10000
             min-pause: 5000
+            tolerable-failed: 3
 
 ```
 

docs/en/seatunnel-engine/separated-cluster-deployment.md

@@ -134,6 +134,10 @@ The timeout time of the checkpoint. If the checkpoint cannot be completed within
 
 The minimum pause (in milliseconds) between consecutive checkpoints. This ensures that checkpoints are not triggered too frequently.
 
+**tolerable-failed**
+
+The number of consecutive checkpoint failures that can be tolerated before the job fails.
+
 Example
 
 ```yaml
@@ -147,6 +151,7 @@ seatunnel:
             interval: 300000
             timeout: 10000
             min-pause: 5000
+            tolerable-failed: 3
 ```
 
 **checkpoint storage**

docs/zh/seatunnel-engine/hybrid-cluster-deployment.md

@@ -93,6 +93,10 @@ seatunnel:
 
 连续检查点之间的最小暂停时间(以毫秒为单位)，确保检查点不会频繁触发。
 
+**tolerable-failed**
+
+可容忍的连续检查点失败次数，超过该次数后作业将失败。
+
 示例
 
 ```yaml
@@ -106,6 +110,7 @@ seatunnel:
             interval: 300000
             timeout: 10000
             min-pause: 5000
+            tolerable-failed: 3
 ```
 
 **checkpoint storage**

docs/zh/seatunnel-engine/separated-cluster-deployment.md

@@ -137,6 +137,11 @@ seatunnel:
 
 连续检查点之间的最小暂停时间(以毫秒为单位)，确保检查点不会频繁触发。
 
+
+**tolerable-failed**
+
+可容忍的连续检查点失败次数，超过该次数后作业将失败。
+
 示例
 
 ```yaml
@@ -150,6 +155,7 @@ seatunnel:
             interval: 300000
             timeout: 10000
             min-pause: 5000
+            tolerable-failed: 3
 ```
 
 **checkpoint storage**

seatunnel-api/src/main/java/org/apache/seatunnel/api/options/EnvCommonOptions.java

@@ -92,6 +92,14 @@ public class EnvCommonOptions {
                             "The minimum pause (in milliseconds) between consecutive checkpoints. "
                                     + "This ensures that checkpoints are not triggered too frequently and provides.");
 
+    public static final Option<Integer> CHECKPOINT_TOLERABLE_FAILED =
+            Options.key("tolerable-failed")
+                    .intType()
+                    .defaultValue(0)
+                    .withDescription(
+                            "The number of consecutive checkpoint failures that can be tolerated before the job fails. "
+                                    + "Default is 0, which means any checkpoint failure will cause the job to fail.");
+
     public static Option<SaveModeExecuteLocation> SAVEMODE_EXECUTE_LOCATION =
             Options.key("savemode.execute.location")
                     .enumType(SaveModeExecuteLocation.class)

seatunnel-api/src/main/java/org/apache/seatunnel/api/options/EnvOptionRule.java

@@ -43,6 +43,7 @@ public OptionRule optionRule() {
                         EnvCommonOptions.CHECKPOINT_INTERVAL,
                         EnvCommonOptions.CHECKPOINT_TIMEOUT,
                         EnvCommonOptions.CHECKPOINT_MIN_PAUSE,
+                        EnvCommonOptions.CHECKPOINT_TOLERABLE_FAILED,
                         EnvCommonOptions.READ_LIMIT_ROW_PER_SECOND,
                         EnvCommonOptions.READ_LIMIT_BYTES_PER_SECOND,
                         EnvCommonOptions.SAVEMODE_EXECUTE_LOCATION,

seatunnel-core/seatunnel-flink-starter/seatunnel-flink-starter-common/src/main/java/org/apache/seatunnel/core/starter/flink/execution/AbstractFlinkRuntimeEnvironment.java

@@ -114,6 +114,10 @@ protected void setCheckpoint() {
         } else if (config.hasPath(EnvCommonOptions.CHECKPOINT_MIN_PAUSE.key())) {
             long minPause = config.getLong(EnvCommonOptions.CHECKPOINT_MIN_PAUSE.key());
             checkpointConfig.setMinPauseBetweenCheckpoints(minPause);
+        } else if (config.hasPath(EnvCommonOptions.CHECKPOINT_TOLERABLE_FAILED.key())) {
+            int tolerableFailedCheckpointsNums =
+                    config.getInt(EnvCommonOptions.CHECKPOINT_TOLERABLE_FAILED.key());
+            checkpointConfig.setTolerableCheckpointFailureNumber(tolerableFailedCheckpointsNums);
         }
 
         if (EnvironmentUtil.hasPathAndWaring(config, ConfigKeyName.CHECKPOINT_MODE)) {

seatunnel-engine/seatunnel-engine-common/src/main/java/org/apache/seatunnel/engine/common/config/YamlSeaTunnelDomConfigProcessor.java

@@ -299,6 +299,15 @@ private CheckpointConfig parseCheckpointConfig(Node checkpointNode) {
                                 ServerConfigOptions.MasterServerConfigOptions.CHECKPOINT_MIN_PAUSE
                                         .key(),
                                 getTextContent(node)));
+            } else if (ServerConfigOptions.MasterServerConfigOptions.CHECKPOINT_TOLERABLE_FAILED
+                    .key()
+                    .equals(name)) {
+                checkpointConfig.setCheckpointTolerableFailed(
+                        getIntegerValue(
+                                ServerConfigOptions.MasterServerConfigOptions
+                                        .CHECKPOINT_TOLERABLE_FAILED
+                                        .key(),
+                                getTextContent(node)));
             } else if (ServerConfigOptions.MasterServerConfigOptions
                     .SCHEMA_CHANGE_CHECKPOINT_TIMEOUT
                     .key()

seatunnel-engine/seatunnel-engine-common/src/main/java/org/apache/seatunnel/engine/common/config/server/CheckpointConfig.java

@@ -34,6 +34,9 @@ public class CheckpointConfig implements Serializable {
             ServerConfigOptions.MasterServerConfigOptions.CHECKPOINT_TIMEOUT.defaultValue();
     private long checkpointMinPause =
             ServerConfigOptions.MasterServerConfigOptions.CHECKPOINT_MIN_PAUSE.defaultValue();
+    private int checkpointTolerableFailed =
+            ServerConfigOptions.MasterServerConfigOptions.CHECKPOINT_TOLERABLE_FAILED
+                    .defaultValue();
     private long schemaChangeCheckpointTimeout =
             ServerConfigOptions.MasterServerConfigOptions.SCHEMA_CHANGE_CHECKPOINT_TIMEOUT
                     .defaultValue();
@@ -67,4 +70,11 @@ public void setSchemaChangeCheckpointTimeout(long checkpointTimeout) {
                 "The minimum checkpoint timeout is 10 ms.");
         this.schemaChangeCheckpointTimeout = checkpointTimeout;
     }
+
+    public void setCheckpointTolerableFailed(int tolerableFailedCheckpoints) {
+        checkArgument(
+                tolerableFailedCheckpoints >= 0,
+                "The tolerable failed checkpoints must be non-negative.");
+        this.checkpointTolerableFailed = tolerableFailedCheckpoints;
+    }
 }

seatunnel-engine/seatunnel-engine-common/src/main/java/org/apache/seatunnel/engine/common/config/server/ServerConfigOptions.java

@@ -135,6 +135,14 @@ public static class MasterServerConfigOptions {
                                 "The minimum pause (in milliseconds) between consecutive checkpoints. "
                                         + "This ensures that checkpoints are not triggered too frequently and provides.");
 
+        public static final Option<Integer> CHECKPOINT_TOLERABLE_FAILED =
+                Options.key("tolerable-failed")
+                        .intType()
+                        .defaultValue(0)
+                        .withDescription(
+                                "The number of consecutive checkpoint failures that can be tolerated before the job fails. "
+                                        + "Default is 0, which means any checkpoint failure will cause the job to fail.");
+
         public static final Option<String> CHECKPOINT_STORAGE_TYPE =
                 Options.key("type")
                         .stringType()

seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/checkpoint/CheckpointCoordinator.java

@@ -98,7 +98,7 @@ public class CheckpointCoordinator {
 
     private final CheckpointManager checkpointManager;
 
-    private final CheckpointStorage checkpointStorage;
+    @Getter private CheckpointStorage checkpointStorage;
 
     @Getter private final CheckpointIDCounter checkpointIdCounter;
 
@@ -132,6 +132,8 @@ public class CheckpointCoordinator {
 
     private final AtomicInteger pendingCounter = new AtomicInteger(0);
 
+    private final AtomicInteger consecutiveFailedCounter = new AtomicInteger(0);
+
     private final AtomicBoolean schemaChanging = new AtomicBoolean(false);
 
     private final Object lock = new Object();
@@ -153,7 +155,7 @@ public class CheckpointCoordinator {
     // processed with one savepoint operation in the same time.
     private PendingCheckpoint savepointPendingCheckpoint;
 
-    private final String checkpointStateImapKey;
+    @Getter private final String checkpointStateImapKey;
 
     @SneakyThrows
     public CheckpointCoordinator(
@@ -287,6 +289,25 @@ private void handleCoordinatorError(CheckpointCloseReason reason, Throwable e) {
         if (checkpointCoordinatorFuture.isDone()) {
             return;
         }
+
+        int failedCount = consecutiveFailedCounter.incrementAndGet();
+        int tolerableFailures = coordinatorConfig.getCheckpointTolerableFailed();
+
+        if (tolerableFailures > 0 && failedCount <= tolerableFailures) {
+            LOG.warn(
+                    "Checkpoint failed (consecutive failures: {}/{}): {}",
+                    failedCount,
+                    tolerableFailures,
+                    ExceptionUtils.getMessage(checkpointException));
+            cleanFailedCheckpoint(reason);
+            return;
+        }
+
+        LOG.error(
+                "Checkpoint failures exceeded tolerable limit ({}/{}), failing the job",
+                failedCount,
+                tolerableFailures);
+
         updateStatus(CheckpointCoordinatorStatus.FAILED);
         checkpointCoordinatorFuture.complete(
                 new CheckpointCoordinatorState(
@@ -630,7 +651,8 @@ private PassiveCompletableFuture<CompletedCheckpoint> completableFutureWithError
         return new PassiveCompletableFuture<>(future);
     }
 
-    private void startTriggerPendingCheckpoint(
+    @VisibleForTesting
+    protected void startTriggerPendingCheckpoint(
             CompletableFuture<PendingCheckpoint> pendingCompletableFuture) {
         pendingCompletableFuture.thenAccept(
                 pendingCheckpoint -> {
@@ -717,7 +739,8 @@ private void startTriggerPendingCheckpoint(
         pendingCounter.incrementAndGet();
     }
 
-    private CompletableFuture<PendingCheckpoint> createPendingCheckpoint(
+    @VisibleForTesting
+    protected CompletableFuture<PendingCheckpoint> createPendingCheckpoint(
             long triggerTimestamp, CheckpointType checkpointType) {
         synchronized (lock) {
             CompletableFuture<Long> idFuture;
@@ -951,6 +974,13 @@ public synchronized void completePendingCheckpoint(CompletedCheckpoint completed
         notifyCompleted(completedCheckpoint);
         pendingCheckpoints.remove(checkpointId).abortCheckpointTimeoutFutureWhenIsCompleted();
         pendingCounter.decrementAndGet();
+        int lastestFailedCount = consecutiveFailedCounter.getAndSet(0);
+        if (lastestFailedCount > 0) {
+            LOG.info(
+                    "Reset consecutive failed counter from {} to 0 after checkpoint {} completed",
+                    lastestFailedCount,
+                    completedCheckpoint.getCheckpointId());
+        }
 
         if (isCompleted()) {
             cleanPendingCheckpoint(CheckpointCloseReason.CHECKPOINT_COORDINATOR_COMPLETED);
@@ -1112,8 +1142,32 @@ protected void completeSchemaChangeAfterCheckpoint(CompletedCheckpoint checkpoin
         }
     }
 
-    public String getCheckpointStateImapKey() {
-        return checkpointStateImapKey;
+    /**
+     * Clean only the failed checkpoint(s) without shutting down the coordinator. This is used for
+     * tolerable checkpoint failures to allow subsequent checkpoints to continue.
+     */
+    protected void cleanFailedCheckpoint(CheckpointCloseReason closedReason) {
+        synchronized (lock) {
+            LOG.info("start clean failed checkpoint cause {}", closedReason.message());
+            if (!pendingCheckpoints.isEmpty()) {
+                pendingCheckpoints
+                        .values()
+                        .forEach(
+                                pendingCheckpoint ->
+                                        pendingCheckpoint.abortCheckpoint(closedReason, null));
+                pendingCheckpoints.clear();
+            }
+            pendingCounter.set(0);
+        }
+    }
+
+    public int getConsecutiveFailedCounter() {
+        return consecutiveFailedCounter.get();
+    }
+
+    @VisibleForTesting
+    public void setCheckpointStorage(CheckpointStorage checkpointStorage) {
+        this.checkpointStorage = checkpointStorage;
     }
 
     /** Only for test */

seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/checkpoint/CheckpointManager.java

@@ -17,6 +17,8 @@
 
 package org.apache.seatunnel.engine.server.checkpoint;
 
+import org.apache.seatunnel.shade.com.google.common.annotations.VisibleForTesting;
+
 import org.apache.seatunnel.api.tracing.MDCTracer;
 import org.apache.seatunnel.engine.checkpoint.storage.PipelineState;
 import org.apache.seatunnel.engine.checkpoint.storage.api.CheckpointStorage;
@@ -183,6 +185,11 @@ public CheckpointCoordinator getCheckpointCoordinator(int pipelineId) {
         return coordinator;
     }
 
+    @VisibleForTesting
+    protected void setCheckpointCoordinator(int pipelineId, CheckpointCoordinator coordinator) {
+        coordinatorMap.put(pipelineId, coordinator);
+    }
+
     /**
      * Called by the {@link Task}. <br>
      * used by Task to report the {@link SeaTunnelTaskState} of the state machine.

seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/checkpoint/PendingCheckpoint.java

@@ -170,7 +170,8 @@ private CompletedCheckpoint toCompletedCheckpoint() {
 
     public void abortCheckpoint(CheckpointCloseReason closedReason, @Nullable Throwable cause) {
         if (closedReason.equals(CheckpointCloseReason.CHECKPOINT_COORDINATOR_RESET)
-                || closedReason.equals(CheckpointCloseReason.PIPELINE_END)) {
+                || closedReason.equals(CheckpointCloseReason.PIPELINE_END)
+                || closedReason.equals(CheckpointCloseReason.CHECKPOINT_EXPIRED)) {
             completableFuture.complete(null);
         } else {
             this.failureCause = new CheckpointException(closedReason, cause);

seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/master/JobMaster.java

@@ -334,6 +334,8 @@ private CheckpointConfig createJobCheckpointConfig(
         jobCheckpointConfig.setCheckpointTimeout(defaultCheckpointConfig.getCheckpointTimeout());
         jobCheckpointConfig.setCheckpointInterval(defaultCheckpointConfig.getCheckpointInterval());
         jobCheckpointConfig.setCheckpointMinPause(defaultCheckpointConfig.getCheckpointMinPause());
+        jobCheckpointConfig.setCheckpointTolerableFailed(
+                defaultCheckpointConfig.getCheckpointTolerableFailed());
 
         CheckpointStorageConfig jobCheckpointStorageConfig = new CheckpointStorageConfig();
         jobCheckpointStorageConfig.setStorage(defaultCheckpointConfig.getStorage().getStorage());
@@ -363,6 +365,12 @@ private CheckpointConfig createJobCheckpointConfig(
                     Long.parseLong(
                             jobEnv.get(EnvCommonOptions.CHECKPOINT_MIN_PAUSE.key()).toString()));
         }
+        if (jobEnv.containsKey(EnvCommonOptions.CHECKPOINT_TOLERABLE_FAILED.key())) {
+            jobCheckpointConfig.setCheckpointTolerableFailed(
+                    Integer.parseInt(
+                            jobEnv.get(EnvCommonOptions.CHECKPOINT_TOLERABLE_FAILED.key())
+                                    .toString()));
+        }
         return jobCheckpointConfig;
     }