Skip to content

[FLINK-38544] Add spill-to-disk for channel state recovery with filtering #27905

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Draft
1996fanrui wants to merge 2 commits into
base: master
Choose a base branch
from
Draft

[FLINK-38544] Add spill-to-disk for channel state recovery with filtering #27905

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
884e214
[FLINK-38541][checkpoint] Expose checkpointing during recovery config…
1996fanrui Mar 18, 2026
bd3a6cb
[FLINK-38541][checkpoint] Randomize UNALIGNED_DURING_RECOVERY_ENABLED…
1996fanrui Jan 15, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
6 changes: 6 additions & 0 deletions docs/layouts/shortcodes/generated/checkpointing_configuration.html
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -158,6 +158,12 @@
<td>Integer</td>
<td>The tolerable checkpoint consecutive failure number. If set to 0, that means we do not tolerance any checkpoint failure. This only applies to the following failure reasons: IOException on the Job Manager, failures in the async phase on the Task Managers and checkpoint expiration due to a timeout. Failures originating from the sync phase on the Task Managers are always forcing failover of an affected task. Other types of checkpoint failures (such as checkpoint being subsumed) are being ignored.</td>
</tr>
<tr>
<td><h5>execution.checkpointing.unaligned.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>
</tr>
<tr>
<td><h5>execution.checkpointing.unaligned.enabled</h5></td>
<td style="word-wrap: break-word;">false</td>
Expand Down
6 changes: 6 additions & 0 deletions docs/layouts/shortcodes/generated/common_checkpointing_section.html
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -56,6 +56,12 @@
<td>Integer</td>
<td>The maximum number of completed checkpoints to retain.</td>
</tr>
<tr>
<td><h5>execution.checkpointing.unaligned.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>
</tr>
<tr>
<td><h5>execution.checkpointing.unaligned.recover-output-on-downstream.enabled</h5></td>
<td style="word-wrap: break-word;">false</td>
Expand Down
3 changes: 1 addition & 2 deletions flink-core/src/main/java/org/apache/flink/configuration/CheckpointingOptions.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -670,8 +670,7 @@ public class CheckpointingOptions {
+ "when job restores from the unaligned checkpoint.");

@Experimental
@Documentation.ExcludeFromDocumentation(
"This option is not yet ready for public use, will be documented in a follow-up commit")
@Documentation.Section(Documentation.Sections.COMMON_CHECKPOINTING)
public static final ConfigOption<Boolean> UNALIGNED_DURING_RECOVERY_ENABLED =
ConfigOptions.key("execution.checkpointing.unaligned.during-recovery.enabled")
.booleanType()
Expand Down
1 change: 1 addition & 0 deletions ...flink-test-utils/src/main/java/org/apache/flink/streaming/util/TestStreamEnvironment.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -148,6 +148,7 @@ private static void randomizeConfiguration(MiniCluster miniCluster, Configuratio
randomize(conf, CheckpointingOptions.ENABLE_UNALIGNED, true, false);
randomize(
conf, CheckpointingOptions.UNALIGNED_RECOVER_OUTPUT_ON_DOWNSTREAM, true, false);
randomize(conf, CheckpointingOptions.UNALIGNED_DURING_RECOVERY_ENABLED, true, false);
randomize(
conf,
CheckpointingOptions.ALIGNED_CHECKPOINT_TIMEOUT,
Expand Down