[FLINK-40210][checkpoint] Add Per-Region timeout handling and bounded source forced global checkpoint

docs/layouts/shortcodes/generated/checkpointing_configuration.html

@@ -140,6 +140,24 @@
             <td>Integer</td>
             <td>The maximum number of completed checkpoints to retain.</td>
         </tr>
+        <tr>
+            <td><h5>execution.checkpointing.region.enabled</h5></td>
+            <td style="word-wrap: break-word;">false</td>
+            <td>Boolean</td>
+            <td>Global switch for Regional Checkpoint. When enabled, partial region failures during checkpoint will not abort the entire checkpoint. Historical state will be used for failed regions.</td>
+        </tr>
+        <tr>
+            <td><h5>execution.checkpointing.region.max-consecutive-failures</h5></td>
+            <td style="word-wrap: break-word;">2</td>
+            <td>Integer</td>
+            <td>Maximum number of consecutive checkpoints that may reference historical checkpoint state.</td>
+        </tr>
+        <tr>
+            <td><h5>execution.checkpointing.region.max-failure-ratio</h5></td>
+            <td style="word-wrap: break-word;">0.3</td>
+            <td>Double</td>
+            <td>Maximum ratio of regions that may fail within a single checkpoint and still allow commit.</td>
+        </tr>
         <tr>
             <td><h5>execution.checkpointing.savepoint-dir</h5></td>
             <td style="word-wrap: break-word;">(none)</td>

docs/layouts/shortcodes/generated/expert_checkpointing_section.html

@@ -20,6 +20,24 @@
             <td>MemorySize</td>
             <td>The minimum size of state data files. All state chunks smaller than that are stored inline in the root checkpoint metadata file. The max memory threshold for this configuration is 1MB.</td>
         </tr>
+        <tr>
+            <td><h5>execution.checkpointing.region.enabled</h5></td>
+            <td style="word-wrap: break-word;">false</td>
+            <td>Boolean</td>
+            <td>Global switch for Regional Checkpoint. When enabled, partial region failures during checkpoint will not abort the entire checkpoint. Historical state will be used for failed regions.</td>
+        </tr>
+        <tr>
+            <td><h5>execution.checkpointing.region.max-consecutive-failures</h5></td>
+            <td style="word-wrap: break-word;">2</td>
+            <td>Integer</td>
+            <td>Maximum number of consecutive checkpoints that may reference historical checkpoint state.</td>
+        </tr>
+        <tr>
+            <td><h5>execution.checkpointing.region.max-failure-ratio</h5></td>
+            <td style="word-wrap: break-word;">0.3</td>
+            <td>Double</td>
+            <td>Maximum ratio of regions that may fail within a single checkpoint and still allow commit.</td>
+        </tr>
         <tr>
             <td><h5>execution.checkpointing.write-buffer-size</h5></td>
             <td style="word-wrap: break-word;">4096</td>

flink-core/src/main/java/org/apache/flink/api/common/state/CheckpointListener.java

@@ -19,6 +19,7 @@
 package org.apache.flink.api.common.state;
 
 import org.apache.flink.annotation.Public;
+import org.apache.flink.annotation.PublicEvolving;
 
 /**
  * This interface is typically only needed for transactional interaction with the "outside world",
@@ -121,6 +122,60 @@ public interface CheckpointListener {
      */
     void notifyCheckpointComplete(long checkpointId) throws Exception;
 
+    /**
+     * Notifies the listener that the checkpoint with the given {@code checkpointId} completed and
+     * was committed, providing additional context about whether this is a regional checkpoint.
+     *
+     * <p>This method is called instead of {@link #notifyCheckpointComplete(long)} when the
+     * framework has Regional Checkpoint information available. The default implementation delegates
+     * to {@link #notifyCheckpointComplete(long)}, so existing implementations are unaffected.
+     *
+     * <p>Implementations that need to distinguish between global checkpoints (all tasks
+     * acknowledged) and regional checkpoints (some tasks fell back to historical state) can
+     * override this method to inspect the {@link RegionalCheckpointInfo}.
+     *
+     * <p>Per FLIP-600, this method is called on <b>healthy-region tasks</b> only. Tasks in failed
+     * regions receive {@link #notifyRegionalCheckpointFallback(long, long)} instead.
+     *
+     * @param checkpointId The ID of the checkpoint that has been completed.
+     * @param regionalCheckpointInfo Context about which subtasks used historical state. Use {@link
+     *     RegionalCheckpointInfo#isGlobalCheckpoint()} to check if all tasks acknowledged.
+     * @throws Exception This method can propagate exceptions, which leads to a failure/recovery for
+     *     the task. Note that this will NOT lead to the checkpoint being revoked.
+     */
+    @PublicEvolving
+    default void notifyRegionalCheckpointComplete(
+            long checkpointId, RegionalCheckpointInfo regionalCheckpointInfo) throws Exception {
+        notifyCheckpointComplete(checkpointId);
+    }
+
+    /**
+     * Notifies the listener that a regional checkpoint has completed but this task's region fell
+     * back to a historical checkpoint. Sent to tasks in failed regions so they can clean up stale
+     * local state from the aborted attempt.
+     *
+     * <p>Per FLIP-600, this method is called on <b>failed-region tasks</b> only. Tasks in healthy
+     * regions receive {@link #notifyRegionalCheckpointComplete(long, RegionalCheckpointInfo)}
+     * instead.
+     *
+     * <p>When a regional checkpoint completes, the framework may have already cancelled/restarted
+     * the failed-region tasks (decline path) or they may still be running but did not finish the
+     * checkpoint (timeout path). This notification is delivered via the same task-side
+     * checkpoint-complete RPC path so that it survives task restarts and is applied after the task
+     * is recovered. Implementations that maintain local checkpoint state (e.g. {@code
+     * TaskLocalStateStore}) should override this method to discard the stale local state of the
+     * failed checkpoint attempt.
+     *
+     * <p>Default: no-op for backward compatibility.
+     *
+     * @param checkpointId the completed regional checkpoint id
+     * @param fallbackCheckpointId the historical checkpoint this task fell back to
+     */
+    @PublicEvolving
+    default void notifyRegionalCheckpointFallback(long checkpointId, long fallbackCheckpointId) {
+        // no-op for backward compatibility
+    }
+
     /**
      * This method is called as a notification once a distributed checkpoint has been aborted.
      *

flink-core/src/main/java/org/apache/flink/api/common/state/RegionalCheckpointInfo.java

@@ -0,0 +1,103 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.api.common.state;
+
+import org.apache.flink.annotation.PublicEvolving;
+
+import java.util.Collections;
+import java.util.Map;
+import java.util.Set;
+
+/**
+ * Provides context about a completed checkpoint, allowing {@link CheckpointListener}
+ * implementations to distinguish between a global checkpoint and a regional checkpoint.
+ *
+ * <p>A <b>global checkpoint</b> is one where all tasks acknowledged successfully. A <b>regional
+ * checkpoint</b> is one where some tasks failed to acknowledge and their state was replaced by
+ * state from a previous successful checkpoint.
+ *
+ * <p>This class provides:
+ *
+ * <ul>
+ *   <li>{@link #isGlobalCheckpoint()} — whether all tasks contributed current state
+ *   <li>{@link #getFallbackCheckpointSubtasks()} — which subtasks (by operator name and subtask
+ *       index) used historical state, grouped by the fallback checkpoint ID they reference
+ * </ul>
+ *
+ * <p>For a global checkpoint, {@link #isGlobalCheckpoint()} returns {@code true} and the fallback
+ * map is empty.
+ */
+@PublicEvolving
+public class RegionalCheckpointInfo {
+
+    /** Singleton instance representing a global checkpoint (no fallback subtasks). */
+    private static final RegionalCheckpointInfo GLOBAL =
+            new RegionalCheckpointInfo(Collections.emptyMap());
+
+    /**
+     * Mapping from fallback checkpointId to the set of operator-subtask identifiers whose state
+     * originates from that historical checkpoint rather than the current one.
+     *
+     * <p>Each entry in the set is formatted as "operatorName#subtaskIndex" (e.g., "Source:
+     * my_source -> Sink: my_sink#0"). In practice, implementations typically only need to check
+     * {@link #isGlobalCheckpoint()} or use {@link #getFallbackCheckpointIds()} to determine which
+     * historical checkpoints are referenced.
+     */
+    private final Map<Long, Set<String>> fallbackCheckpointSubtasks;
+
+    public RegionalCheckpointInfo(Map<Long, Set<String>> fallbackCheckpointSubtasks) {
+        this.fallbackCheckpointSubtasks = Collections.unmodifiableMap(fallbackCheckpointSubtasks);
+    }
+
+    /** Returns a {@link RegionalCheckpointInfo} representing a global checkpoint. */
+    public static RegionalCheckpointInfo globalCheckpoint() {
+        return GLOBAL;
+    }
+
+    /**
+     * Returns {@code true} if this is a global checkpoint where all tasks acknowledged
+     * successfully.
+     */
+    public boolean isGlobalCheckpoint() {
+        return fallbackCheckpointSubtasks.isEmpty();
+    }
+
+    /**
+     * Returns the set of fallback checkpoint IDs referenced by this regional checkpoint.
+     *
+     * <p>For a global checkpoint, this returns an empty set. For a regional checkpoint, each ID in
+     * the returned set represents a historical checkpoint whose state is used by some subtasks in
+     * this completed checkpoint.
+     */
+    public Set<Long> getFallbackCheckpointIds() {
+        return fallbackCheckpointSubtasks.keySet();
+    }
+
+    /**
+     * Returns the full mapping from fallback checkpoint IDs to the set of subtask identifiers whose
+     * state originates from that historical checkpoint.
+     *
+     * <p>Each subtask identifier is a string in the format "operatorName#subtaskIndex".
+     *
+     * <p>For a global checkpoint, this returns an empty map.
+     */
+    public Map<Long, Set<String>> getFallbackCheckpointSubtasks() {
+        return fallbackCheckpointSubtasks;
+    }
+}

flink-core/src/main/java/org/apache/flink/configuration/CheckpointingOptions.java

@@ -286,6 +286,41 @@ public class CheckpointingOptions {
                                     + ". By default, local backup is deactivated. Local backup currently only "
                                     + "covers keyed state backends (including both the EmbeddedRocksDBStateBackend and the HashMapStateBackend).");
 
+    // ------------------------------------------------------------------------
+    //  Options related to regional checkpoint
+    // ------------------------------------------------------------------------
+
+    @Experimental
+    @Documentation.Section(Documentation.Sections.EXPERT_CHECKPOINTING)
+    public static final ConfigOption<Boolean> REGIONAL_CHECKPOINT_ENABLED =
+            ConfigOptions.key("execution.checkpointing.region.enabled")
+                    .booleanType()
+                    .defaultValue(false)
+                    .withDescription(
+                            "Global switch for Regional Checkpoint. When enabled, partial "
+                                    + "region failures during checkpoint will not abort the entire "
+                                    + "checkpoint. Historical state will be used for failed regions.");
+
+    @Experimental
+    @Documentation.Section(Documentation.Sections.EXPERT_CHECKPOINTING)
+    public static final ConfigOption<Double> REGIONAL_CHECKPOINT_MAX_FAILURE_RATIO =
+            ConfigOptions.key("execution.checkpointing.region.max-failure-ratio")
+                    .doubleType()
+                    .defaultValue(0.3)
+                    .withDescription(
+                            "Maximum ratio of regions that may fail within a single checkpoint "
+                                    + "and still allow commit.");
+
+    @Experimental
+    @Documentation.Section(Documentation.Sections.EXPERT_CHECKPOINTING)
+    public static final ConfigOption<Integer> REGIONAL_CHECKPOINT_MAX_CONSECUTIVE_FAILURES =
+            ConfigOptions.key("execution.checkpointing.region.max-consecutive-failures")
+                    .intType()
+                    .defaultValue(2)
+                    .withDescription(
+                            "Maximum number of consecutive checkpoints that may reference "
+                                    + "historical checkpoint state.");
+
     // ------------------------------------------------------------------------
     //  Options related to file merging
     // ------------------------------------------------------------------------