[FLINK-41193][checkpointing] Disable checkpointing during recovery when aligned-checkpoint-timeout > 0

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>

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>

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);
     }

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(