[FLINK-38939][runtime] Pause sources until the 1st checkpoint to prioritize processing recovered records #27440

rkhachatryan · 2026-01-19T09:32:16Z

No description provided.

flinkbot · 2026-01-19T09:38:20Z

CI report:

7729a99 Azure: FAILURE

Bot commands

The @flinkbot bot supports the following commands:

@flinkbot run azure re-run the last Azure build

davidradl · 2026-01-20T09:47:19Z

docs/layouts/shortcodes/generated/pipeline_configuration.html

+            <td><h5>pipeline.sources.pause-until-first-checkpoint</h5></td>
+            <td style="word-wrap: break-word;">true</td>
+            <td>Boolean</td>
+            <td>Don't pull any data from sources until the first checkpoint is triggered. This might be helpful in reducing recovery times. Incompatible 0 value for execution.checkpointing.interval-during-backlog</td>


I suggest expanding on "This might be helpful in reducing recovery times." to be specific about the scenarios that this would be helpful in. Are there scenarios it would not be helpful in? If there are no downsides to this change - then do we need a config option?

I am not sure what Incompatible 0 value for execution.checkpointing.interval-during-backlog means .

davidradl · 2026-01-20T10:05:14Z

...k-runtime/src/main/java/org/apache/flink/streaming/api/graph/StreamingJobGraphGenerator.java

+                && checkpointConfig.isPauseSourcesUntilFirstCheckpoint()) {
+            throw new IllegalArgumentException(
+                    "Pausing sources until first checkpoint is incompatible with disabling checkpoints during backlog processing. "
+                            + "Please consult "


nit: "Please consult ...." -> "Please review and choose whether you require + CheckpointingOptions.PAUSE_SOURCES_UNTIL_FIRST_CHECKPOINT.key()
+ " or"
+ CheckpointingOptions.CHECKPOINTING_INTERVAL_DURING_BACKLOG.key());

1996fanrui

Hey @rkhachatryan , thanks for the PR, I have several questions about this approach.

Pause sources until the 1st checkpoint to prioritize processing recovered records
Don't pull any data from sources until the first checkpoint is triggered.

If so, the source does not work even if all recovered buffers are consumed, right?

Let me understand the existing issue and the current approach:

The task will be switched from INITIALIZATION to RUNNING once all recovered input buffers and output buffers are consumed.
The recovered buffer of some input channels are fully consumed, and there are some new buffers is coming. The recovered buffer of rest of channels are not fully consumed.

Issue: If task starts consume new buffers before all recovered buffer are consumed, it will be switched to running later.

IIUC,, the purpose of pause source is avoid new buffers are generated. Is it correct?

If so, I do not think it works perfect since new buffers can be generated from the recovered buffers of upstream task. Of course, pause source could avoid new buffers from outside of flink during recovery.

Blocking channels whose recovered buffers are fully consumed maybe more fine-grained that pausing source, it allows task consumes recovered buffers before new buffers, as well as the upstream tasks and source are not blocked as early as possible.

Also, FLIP-547 part 4.6 will introduce fine-grained blocking mechanism. Not sure whether pausing source is still needed if new mechanism will be introduced in the near future?

Looking forward to your opinion, thanks

rkhachatryan · 2026-01-21T12:24:38Z

Yes @1996fanrui, you're right.

The purpose of this change is to prevent new input records from delaying the switch of the downstream tasks to RUNNING.
This doesn't help in every case; an example where it is useful is two JOINed sources, where one of the channels has a lot of checkpointed buffers; and the other one is fast in producing new data (from source).

In a sense, this is a lightweight alternative to FLIP-547.
To my understanding the timeline of implementing and stabilizing FLIP-547 is relatively long (PCMIIW), so this feature still makes sense.

1996fanrui

Thanks @rkhachatryan for the comment!

Sounds make sense for considering this approach as a lightweight alternative first. I only left one comment, please take a look when you are available, thanks

flink-core/src/main/java/org/apache/flink/configuration/CheckpointingOptions.java

1996fanrui

LGTM assuming CI is green

…eived This allows to prioritize processing of recovered records (when recovering from an unaligned checkpoint)

…inting interval The check doesn't make sense because checkpointing might be disabled before recovery; or there might be a manual checkpoint.

…orConfiguration

rkhachatryan · 2026-01-24T17:26:13Z

flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/AdaptiveScheduler.java

    private JobAllocationsInformation getJobAllocationsInformationFromGraphAndState(
            @Nullable final ExecutionGraph previousExecutionGraph) {

-        CompletedCheckpoint latestCompletedCheckpoint = null;
-        if (jobGraph.isCheckpointingEnabled()) {
-            latestCompletedCheckpoint = completedCheckpointStore.getLatestCheckpoint();
-        }


@1996fanrui , @Izeren

After after minimizing the delay for automatic checkpoints, LocalRecoveryTest#testStateSizeIsConsideredForLocalRecoveryOnRestart starts failing because there's a race condition between a manual and automatic checkpoints.

So I disabled automatic checkpoints in test and removed if (jobGraph.isCheckpointingEnabled()) check (in prod code).
This should be fine - completedCheckpointStore should never be null.

Besides of test, I think checkpointing can be disabled before recovering the job, and this branch should still be executed.

Oh, it can be DeactivatedCheckpointCompletedCheckpointStore 🤔

Thanks @rkhachatryan for looking into. May I know is there any risk for production jobs?

…g for a checkpoint

…ests

Efrat19 · 2026-02-12T08:44:53Z

...rc/main/java/org/apache/flink/runtime/jobgraph/tasks/CheckpointCoordinatorConfiguration.java

+    public long getInitialTriggeringDelay() {
+        return pauseSourcesUntilFirstCheckpoint
+                ? ThreadLocalRandom.current()
+                        .nextLong(minPauseBetweenCheckpoints, minPauseBetweenCheckpoints * 2 + 1)


nit: Maybe add comment / a variable to reason about this math?

Efrat19 · 2026-02-12T08:45:05Z

.../test/java/org/apache/flink/test/state/operator/restore/AbstractOperatorRestoreTestBase.java

    private JobGraph createJobGraph(ExecutionMode mode) {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.enableCheckpointing(500, CheckpointingMode.EXACTLY_ONCE);
+        // todo: review the test and timings


Efrat19

Thank you for these valuable contributions

Efrat19 · 2026-02-12T08:51:26Z

flink-runtime/src/test/java/org/apache/flink/streaming/api/operators/SourceOperatorTest.java

+        return Arrays.asList(true, false);
+    }
+
+    @Parameter public boolean pauseSourcesUntilCheckpoint;


nit property declaration to precede method

rkhachatryan · 2026-02-12T13:03:57Z

Thanks for the reviews!
I'm closing this PR in favor of #27589 - it includes this PR commits

rkhachatryan requested a review from 1996fanrui January 19, 2026 12:53

rkhachatryan marked this pull request as ready for review January 19, 2026 12:54

rkhachatryan requested a review from pnowojski January 19, 2026 12:54

davidradl reviewed Jan 20, 2026

View reviewed changes

github-actions bot added the community-reviewed PR has been reviewed by the community. label Jan 20, 2026

1996fanrui reviewed Jan 21, 2026

View reviewed changes

flink-core/src/main/java/org/apache/flink/configuration/CheckpointingOptions.java Outdated Show resolved Hide resolved

rkhachatryan requested a review from 1996fanrui January 23, 2026 00:04

1996fanrui self-assigned this Jan 23, 2026

1996fanrui approved these changes Jan 23, 2026

View reviewed changes

[hotfix] Close OutputWriter in SourceOperatorStreamTaskTest

845e5cb

rkhachatryan force-pushed the f38939 branch from cea1229 to 7f56978 Compare January 23, 2026 09:41

[FLINK-38939] Pause Sources until the first checkpoint barrier is rec…

168cff8

…eived This allows to prioritize processing of recovered records (when recovering from an unaligned checkpoint)

rkhachatryan force-pushed the f38939 branch from 7f56978 to 2bb4a3a Compare January 23, 2026 12:43

rkhachatryan added 2 commits January 24, 2026 17:18

[hotfix] Try to get last checkpoint on recovery regardless of checkpo…

16d7397

…inting interval The check doesn't make sense because checkpointing might be disabled before recovery; or there might be a manual checkpoint.

[hotfix] Move checkpointing configuration code to CheckpointCoordinat…

40a2ff5

…orConfiguration

rkhachatryan force-pushed the f38939 branch from 2bb4a3a to b4ff432 Compare January 24, 2026 17:18

rkhachatryan commented Jan 24, 2026

View reviewed changes

[FLINK-38939] Minimize checkpoint trigger delay if sources are waitin…

708a116

…g for a checkpoint

rkhachatryan force-pushed the f38939 branch from b4ff432 to 708a116 Compare January 25, 2026 10:17

[hotfix][tests] Increase min pause between checkpoints in migration t…

7729a99

…ests

pnowojski mentioned this pull request Feb 11, 2026

[FLINK-37399][runtime][source] Buffer watermarks for watermark alignment #27589

Open

Efrat19 reviewed Feb 12, 2026

View reviewed changes

rkhachatryan closed this Feb 12, 2026

[FLINK-38939][runtime] Pause sources until the 1st checkpoint to prioritize processing recovered records #27440

[FLINK-38939][runtime] Pause sources until the 1st checkpoint to prioritize processing recovered records #27440

Conversation

rkhachatryan commented Jan 19, 2026

Uh oh!

flinkbot commented Jan 19, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

CI report:

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

1996fanrui left a comment

Choose a reason for hiding this comment

Uh oh!

rkhachatryan commented Jan 21, 2026

Uh oh!

1996fanrui left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

1996fanrui left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Efrat19 left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

rkhachatryan commented Feb 12, 2026

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

5 participants

flinkbot commented Jan 19, 2026 •

edited

Loading