From b266031e362b59ae9ed2caadd0263100bd1ab7fb Mon Sep 17 00:00:00 2001
From: Rui Fan <1996fanrui@gmail.com>
Date: Sun, 25 Oct 2026 21:40:22 +0100
Subject: [PATCH] [FLINK-41193][checkpointing] Disable checkpointing during
 recovery when aligned-checkpoint-timeout > 0

Generated-by: Claude Opus 5.5
---
 .../generated/checkpointing_configuration.html       |  2 +-
 .../generated/common_checkpointing_section.html      |  2 +-
 .../flink/configuration/CheckpointingOptions.java    | 12 +++++++++++-
 .../configuration/CheckpointingOptionsTest.java      |  9 +++++++++
 4 files changed, 22 insertions(+), 3 deletions(-)

diff --git a/docs/layouts/shortcodes/generated/checkpointing_configuration.html b/docs/layouts/shortcodes/generated/checkpointing_configuration.html
index e5fb53d40ca78d..cc6a1b6d5adafe 100644
--- a/docs/layouts/shortcodes/generated/checkpointing_configuration.html
+++ b/docs/layouts/shortcodes/generated/checkpointing_configuration.html
@@ -54,7 +54,7 @@
             <td><h5>execution.checkpointing.during-recovery.enabled</h5></td>
             <td style="word-wrap: break-word;">false</td>
             <td>Boolean</td>
-            <td>Whether to enable checkpointing during recovery from an unaligned checkpoint. When enabled, the job can take checkpoints while still recovering channel state (inflight data) from a previous unaligned checkpoint. This avoids the need to wait for full recovery before the first checkpoint can be triggered, which reduces the window of vulnerability to failures during recovery.<br /><br />This option requires <code class="highlighter-rouge">execution.checkpointing.unaligned.recover-output-on-downstream.enabled</code> to be enabled. It does not require unaligned checkpoints to be currently enabled, because a job may restore from an unaligned checkpoint while having unaligned checkpoints disabled for the new execution.</td>
+            <td>Whether to enable checkpointing during recovery from an unaligned checkpoint. When enabled, the job can take checkpoints while still recovering channel state (inflight data) from a previous unaligned checkpoint. This avoids the need to wait for full recovery before the first checkpoint can be triggered, which reduces the window of vulnerability to failures during recovery.<br /><br />This option requires <code class="highlighter-rouge">execution.checkpointing.unaligned.recover-output-on-downstream.enabled</code> to be enabled. It does not require unaligned checkpoints to be currently enabled, because a job may restore from an unaligned checkpoint while having unaligned checkpoints disabled for the new execution.<br /><br />Checkpointing during recovery is currently not supported when <code class="highlighter-rouge">execution.checkpointing.aligned-checkpoint-timeout</code> is greater than 0, and will be disabled in this case.</td>
         </tr>
         <tr>
             <td><h5>execution.checkpointing.externalized-checkpoint-retention</h5></td>
diff --git a/docs/layouts/shortcodes/generated/common_checkpointing_section.html b/docs/layouts/shortcodes/generated/common_checkpointing_section.html
index 1a5159022eb7eb..40e12a1890f7cf 100644
--- a/docs/layouts/shortcodes/generated/common_checkpointing_section.html
+++ b/docs/layouts/shortcodes/generated/common_checkpointing_section.html
@@ -36,7 +36,7 @@
             <td><h5>execution.checkpointing.during-recovery.enabled</h5></td>
             <td style="word-wrap: break-word;">false</td>
             <td>Boolean</td>
-            <td>Whether to enable checkpointing during recovery from an unaligned checkpoint. When enabled, the job can take checkpoints while still recovering channel state (inflight data) from a previous unaligned checkpoint. This avoids the need to wait for full recovery before the first checkpoint can be triggered, which reduces the window of vulnerability to failures during recovery.<br /><br />This option requires <code class="highlighter-rouge">execution.checkpointing.unaligned.recover-output-on-downstream.enabled</code> to be enabled. It does not require unaligned checkpoints to be currently enabled, because a job may restore from an unaligned checkpoint while having unaligned checkpoints disabled for the new execution.</td>
+            <td>Whether to enable checkpointing during recovery from an unaligned checkpoint. When enabled, the job can take checkpoints while still recovering channel state (inflight data) from a previous unaligned checkpoint. This avoids the need to wait for full recovery before the first checkpoint can be triggered, which reduces the window of vulnerability to failures during recovery.<br /><br />This option requires <code class="highlighter-rouge">execution.checkpointing.unaligned.recover-output-on-downstream.enabled</code> to be enabled. It does not require unaligned checkpoints to be currently enabled, because a job may restore from an unaligned checkpoint while having unaligned checkpoints disabled for the new execution.<br /><br />Checkpointing during recovery is currently not supported when <code class="highlighter-rouge">execution.checkpointing.aligned-checkpoint-timeout</code> is greater than 0, and will be disabled in this case.</td>
         </tr>
         <tr>
             <td><h5>execution.checkpointing.incremental</h5></td>
diff --git a/flink-core/src/main/java/org/apache/flink/configuration/CheckpointingOptions.java b/flink-core/src/main/java/org/apache/flink/configuration/CheckpointingOptions.java
index 2c9f3834177dba..9068c786885b38 100644
--- a/flink-core/src/main/java/org/apache/flink/configuration/CheckpointingOptions.java
+++ b/flink-core/src/main/java/org/apache/flink/configuration/CheckpointingOptions.java
@@ -754,6 +754,11 @@ public class CheckpointingOptions {
                                                     + "checkpoints disabled for the new execution.",
                                             TextElement.code(
                                                     UNALIGNED_RECOVER_OUTPUT_ON_DOWNSTREAM.key()))
+                                    .linebreak()
+                                    .linebreak()
+                                    .text(
+                                            "Checkpointing during recovery is currently not supported when %s is greater than 0, and will be disabled in this case.",
+                                            TextElement.code(ALIGNED_CHECKPOINT_TIMEOUT.key()))
                                     .build());
 
     /**
@@ -857,7 +862,8 @@ public static boolean isUnalignedCheckpointInterruptibleTimersEnabled(Configurat
      *
      * <p>Requires both {@link #UNALIGNED_RECOVER_OUTPUT_ON_DOWNSTREAM} and unaligned checkpoints to
      * be enabled, because checkpointing during recovery is only supported on the unaligned
-     * barrier-handler path.
+     * barrier-handler path. It is also disabled when {@link #ALIGNED_CHECKPOINT_TIMEOUT} is
+     * positive.
      *
      * @param config the configuration to check
      * @return {@code true} if unaligned checkpointing during recovery is enabled, {@code false}
@@ -868,6 +874,10 @@ public static boolean isCheckpointingDuringRecoveryEnabled(Configuration config)
         if (!config.get(UNALIGNED_RECOVER_OUTPUT_ON_DOWNSTREAM)) {
             return false;
         }
+        // TODO FLINK-40526: Support checkpointing during recovery with aligned checkpoint timeout.
+        if (!config.get(ALIGNED_CHECKPOINT_TIMEOUT).isZero()) {
+            return false;
+        }
         return config.get(CHECKPOINTING_DURING_RECOVERY_ENABLED)
                 && isUnalignedCheckpointEnabled(config);
     }
diff --git a/flink-core/src/test/java/org/apache/flink/configuration/CheckpointingOptionsTest.java b/flink-core/src/test/java/org/apache/flink/configuration/CheckpointingOptionsTest.java
index 9c8940eb2462d6..c1192204a0317b 100644
--- a/flink-core/src/test/java/org/apache/flink/configuration/CheckpointingOptionsTest.java
+++ b/flink-core/src/test/java/org/apache/flink/configuration/CheckpointingOptionsTest.java
@@ -386,6 +386,15 @@ void testIsCheckpointingDuringRecoveryEnabled() {
                 .as("During-recovery should be disabled when unaligned checkpoints are disabled")
                 .isFalse();
 
+        // Test when all three prerequisites are enabled but aligned checkpoint timeout is positive
+        Configuration alignedTimeoutConfig = new Configuration(allEnabledConfig);
+        alignedTimeoutConfig.set(
+                CheckpointingOptions.ALIGNED_CHECKPOINT_TIMEOUT, Duration.ofSeconds(1));
+        assertThat(CheckpointingOptions.isCheckpointingDuringRecoveryEnabled(alignedTimeoutConfig))
+                .as(
+                        "During-recovery should be disabled when aligned checkpoint timeout is positive")
+                .isFalse();
+
         // Test when recover-output-on-downstream is explicitly false and during-recovery is true
         Configuration explicitlyDisabledConfig = new Configuration();
         explicitlyDisabledConfig.set(