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

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

@@ -121,6 +121,58 @@ 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.
+     */
+    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
+     */
+    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
     // ------------------------------------------------------------------------

flink-core/src/test/java/org/apache/flink/api/common/state/CheckpointListenerRegionalTest.java

@@ -0,0 +1,153 @@
+/*
+ * 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.junit.jupiter.api.Test;
+
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicLong;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+class CheckpointListenerRegionalTest {
+
+    @Test
+    void testGlobalCheckpointInfoIsGlobal() {
+        RegionalCheckpointInfo info = RegionalCheckpointInfo.globalCheckpoint();
+        assertThat(info.isGlobalCheckpoint()).isTrue();
+        assertThat(info.getFallbackCheckpointSubtasks()).isEmpty();
+        assertThat(info.getFallbackCheckpointIds()).isEmpty();
+    }
+
+    @Test
+    void testRegionalCheckpointInfoNotGlobal() {
+        Map<Long, Set<String>> fallback = new HashMap<>();
+        fallback.put(99L, Set.of("Source: kafka_source#0", "Map#0", "Sink#0"));
+        RegionalCheckpointInfo info = new RegionalCheckpointInfo(fallback);
+
+        assertThat(info.isGlobalCheckpoint()).isFalse();
+        assertThat(info.getFallbackCheckpointIds()).containsExactly(99L);
+        assertThat(info.getFallbackCheckpointSubtasks().get(99L))
+                .containsExactlyInAnyOrder("Source: kafka_source#0", "Map#0", "Sink#0");
+    }
+
+    @Test
+    void testDefaultMethodDelegatesToOriginal() throws Exception {
+        AtomicLong receivedId = new AtomicLong(-1);
+
+        CheckpointListener listener =
+                new CheckpointListener() {
+                    @Override
+                    public void notifyCheckpointComplete(long checkpointId) {
+                        receivedId.set(checkpointId);
+                    }
+                };
+
+        Map<Long, Set<String>> fallback = new HashMap<>();
+        fallback.put(99L, Set.of("Source#0"));
+        RegionalCheckpointInfo info = new RegionalCheckpointInfo(fallback);
+
+        listener.notifyRegionalCheckpointComplete(100L, info);
+        assertThat(receivedId.get()).isEqualTo(100L);
+    }
+
+    @Test
+    void testOverriddenMethodReceivesRegionalInfo() throws Exception {
+        AtomicBoolean wasRegional = new AtomicBoolean(false);
+
+        CheckpointListener listener =
+                new CheckpointListener() {
+                    @Override
+                    public void notifyCheckpointComplete(long checkpointId) {}
+
+                    @Override
+                    public void notifyRegionalCheckpointComplete(
+                            long checkpointId, RegionalCheckpointInfo info) {
+                        wasRegional.set(!info.isGlobalCheckpoint());
+                    }
+                };
+
+        Map<Long, Set<String>> fallback = new HashMap<>();
+        fallback.put(99L, Set.of("Source#0"));
+        listener.notifyRegionalCheckpointComplete(100L, new RegionalCheckpointInfo(fallback));
+        assertThat(wasRegional.get()).isTrue();
+
+        listener.notifyRegionalCheckpointComplete(101L, RegionalCheckpointInfo.globalCheckpoint());
+        assertThat(wasRegional.get()).isFalse();
+    }
+
+    @Test
+    void testNotifyRegionalCheckpointFallbackDefaultNoOp() {
+        // Default implementation should be no-op and not throw
+        CheckpointListener listener =
+                new CheckpointListener() {
+                    @Override
+                    public void notifyCheckpointComplete(long checkpointId) {}
+                };
+        listener.notifyRegionalCheckpointFallback(100L, 99L);
+    }
+
+    @Test
+    void testNotifyRegionalCheckpointFallbackOverridden() {
+        AtomicLong receivedCheckpointId = new AtomicLong(-1);
+        AtomicLong receivedFallbackId = new AtomicLong(-1);
+
+        CheckpointListener listener =
+                new CheckpointListener() {
+                    @Override
+                    public void notifyCheckpointComplete(long checkpointId) {}
+
+                    @Override
+                    public void notifyRegionalCheckpointFallback(
+                            long checkpointId, long fallbackCheckpointId) {
+                        receivedCheckpointId.set(checkpointId);
+                        receivedFallbackId.set(fallbackCheckpointId);
+                    }
+                };
+
+        listener.notifyRegionalCheckpointFallback(100L, 99L);
+        assertThat(receivedCheckpointId.get()).isEqualTo(100L);
+        assertThat(receivedFallbackId.get()).isEqualTo(99L);
+    }
+
+    @Test
+    void testFallbackMapIsUnmodifiable() {
+        Map<Long, Set<String>> fallback = new HashMap<>();
+        fallback.put(99L, Set.of("Source#0"));
+        RegionalCheckpointInfo info = new RegionalCheckpointInfo(fallback);
+
+        assertThatThrownBy(() -> info.getFallbackCheckpointSubtasks().put(98L, Set.of("Map#1")))
+                .isInstanceOf(UnsupportedOperationException.class);
+    }
+
+    @Test
+    void testMultipleFallbackCheckpoints() {
+        Map<Long, Set<String>> fallback = new HashMap<>();
+        fallback.put(99L, Set.of("Source#0", "Map#0"));
+        fallback.put(98L, Set.of("Source#1"));
+        RegionalCheckpointInfo info = new RegionalCheckpointInfo(fallback);
+
+        assertThat(info.isGlobalCheckpoint()).isFalse();
+        assertThat(info.getFallbackCheckpointIds()).containsExactlyInAnyOrder(99L, 98L);
+    }
+}