[FLINK-38940] Start checkpointing when all tasks are running and job is running#27441
Merged
rkhachatryan merged 1 commit intoapache:masterfrom Jan 23, 2026
Merged
[FLINK-38940] Start checkpointing when all tasks are running and job is running#27441rkhachatryan merged 1 commit intoapache:masterfrom
rkhachatryan merged 1 commit intoapache:masterfrom
Conversation
Collaborator
davidradl
reviewed
Jan 19, 2026
| <td><h5>execution.checkpointing.pause-checkpoints-if-tasks-not-running</h5></td> | ||
| <td style="word-wrap: break-word;">false</td> | ||
| <td>Boolean</td> | ||
| <td>When enabled, checkpoints will be paused as long as some tasks are not running (and not finished), even if the job status is running. This help to prevent an additional delay on job startup</td> |
Contributor
There was a problem hiding this comment.
I am struggling with these sentences.
- nit it would be simpler to say if any tasks rather than as long as some tasks
- What does it mean to say a task is not running, but has a status of running? I think we need to talk about non blocking output, maybe more words outside of the config definition to say how to recognise non blocking jobs - as this relates to the subset of jobs we are effecting with this option.
- nit: This help -> This helps
- the text says "prevent an additional delay on job startup". But the logic does not seem startup specific? Can we get into any other non startup related scenarios where this might be helpful
- Can we guide the user as to when they should consider using this option and any downside to using it.
Contributor
Author
There was a problem hiding this comment.
We discussed this PR with @1996fanrui offline and agreed to remove the option because the new behavior is the only desirable (until FLIP-457). After FLIP-457, this logic will be changed.
davidradl
reviewed
Jan 19, 2026
| this.coordinator = checkNotNull(coordinator); | ||
| this.allTasksOutputNonBlocking = allTasksOutputNonBlocking; | ||
| return (jobId, newJobStatus, timestamp) -> { | ||
| if (newJobStatus == JobStatus.RUNNING && allTasksOutputNonBlocking) { |
Contributor
There was a problem hiding this comment.
I an curious - should we be checking the new config option here? I thought this is the condition we wanted to pause checkpointing for? Am I missing something?
Contributor
Author
There was a problem hiding this comment.
The check is in CheckpointCoordinator.createActivatorDeactivator, but as mentioned above, I'm going to remove this option.
davidradl
reviewed
Jan 19, 2026
| new CheckpointCoordinatorDeActivator(this, allTasksOutputNonBlocking); | ||
| if (deActivator == null) { | ||
| if (pauseCheckpointsIfTasksNotRunning && allTasksOutputNonBlocking) { | ||
| deActivator = new ExecutionTrackingCheckpointCoordinatorDeActivator(this); |
Contributor
There was a problem hiding this comment.
Maybe a comment to indicate why this if body results in a pause.
github-actions
bot
added
the
community-reviewed
rkhachatryan
force-pushed
the
f38940
branch
3 times, most recently
from
2012346 to
bb34b02
Compare
1996fanrui
reviewed
Jan 21, 2026
Comment on lines
+34
to
+43
| static CheckpointCoordinatorDeActivator forJobStatus( | ||
| CheckpointCoordinator coordinator, boolean allTasksOutputNonBlocking) { | ||
| this.coordinator = checkNotNull(coordinator); | ||
| this.allTasksOutputNonBlocking = allTasksOutputNonBlocking; | ||
| return (jobId, newJobStatus, timestamp) -> { | ||
| if (newJobStatus == JobStatus.RUNNING && allTasksOutputNonBlocking) { | ||
| // start the checkpoint scheduler if there is no blocking edge | ||
| coordinator.startCheckpointScheduler(); | ||
| } else { | ||
| // anything else should stop the trigger for now | ||
| coordinator.stopCheckpointScheduler(); | ||
| } |
Member
There was a problem hiding this comment.
forJobStatus is only called once, and allTasksOutputNonBlocking is false. so forJobStatus can be simplified.
Contributor
Author
There was a problem hiding this comment.
Right, I've simplified the code, PTAL.
1996fanrui
approved these changes
Jan 23, 2026
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
4 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
No description provided.