[FLINK-38939][runtime] Pause sources until the 1st checkpoint to prioritize processing recovered records by rkhachatryan · Pull Request #27440 · apache/flink

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

No description provided.

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

CI report:

533aff7 Azure: FAILURE

Bot commands

The @flinkbot bot supports the following commands:

@flinkbot run azure re-run the last Azure build

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

1996fanrui

LGTM assuming CI is green

Efrat19

Thank you for these valuable contributions

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 · 2026-02-13T13:07:46Z

Reopening to address test failures

rkhachatryan · 2026-02-18T09:35:16Z

@flinkbot run azure

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

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

…CoordinatorConfiguration

…g for a checkpoint

…ests

With more frequent checkpoints, ids can be duplicated in RestoreUpgradedJobITCase. This change adds a sipmle deduplication before the assertion.

rkhachatryan · 2026-02-18T23:13:27Z

Rebased.

Previous test failures were unrelated:

I believe I've addressed all the feedback, thanks again for the reviews.

Will be merging this PR.

… watermark After FLINK-38939 / apache#27440, if the source operator was stopped while waiting for the first checkpoint then the output needs to be initialized so final watermark can be emitted; otherwise, final checkpoint might fail with java.lang.IllegalStateException This commit fixes the issue by calling initializeMainOutput if necessary.

… watermark After FLINK-38939 / #27440, if the source operator was stopped while waiting for the first checkpoint then the output needs to be initialized so final watermark can be emitted; otherwise, final checkpoint might fail with java.lang.IllegalStateException This commit fixes the issue by calling initializeMainOutput if necessary.

… watermark After FLINK-38939 / apache#27440, if the source operator was stopped while waiting for the first checkpoint then the output needs to be initialized so final watermark can be emitted; otherwise, final checkpoint might fail with java.lang.IllegalStateException This commit fixes the issue by calling initializeMainOutput if necessary.

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

Comment thread docs/layouts/shortcodes/generated/pipeline_configuration.html Outdated

davidradl reviewed Jan 20, 2026

View reviewed changes

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

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

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

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

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

rkhachatryan commented Jan 24, 2026

View reviewed changes

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

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

pnowojski mentioned this pull request Feb 11, 2026

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

Merged

Efrat19 reviewed Feb 12, 2026

View reviewed changes

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

Efrat19 reviewed Feb 12, 2026

View reviewed changes

Comment thread .../test/java/org/apache/flink/test/state/operator/restore/AbstractOperatorRestoreTestBase.java Outdated

Efrat19 reviewed Feb 12, 2026

View reviewed changes

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

rkhachatryan closed this Feb 12, 2026

rkhachatryan reopened this Feb 13, 2026

rkhachatryan force-pushed the f38939 branch 3 times, most recently from d876c70 to 533aff7 Compare February 17, 2026 23:36

rkhachatryan added 7 commits February 18, 2026 23:04

[hotfix][tests] Close OutputWriter in SourceOperatorStreamTaskTest

71a11a2

[FLINK-38939][runtime] Pause Sources until the first checkpoint barri…

2bbb6bf

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

[hotfix][runtime] Try to get last checkpoint on recovery regardless o…

219a600

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

[hotfix][runtime] Move checkpointing configuration code to Checkpoint…

0b1a96a

…CoordinatorConfiguration

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

70fdc66

…g for a checkpoint

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

403b48e

…ests

[hotfix][tests] De-duplicate operator IDs in RestoreUpgradedJobITCase

75c8aea

With more frequent checkpoints, ids can be duplicated in RestoreUpgradedJobITCase. This change adds a sipmle deduplication before the assertion.

rkhachatryan force-pushed the f38939 branch from 533aff7 to 75c8aea Compare February 18, 2026 23:07

rkhachatryan merged commit ae0c1e6 into apache:master Feb 18, 2026

rkhachatryan deleted the f38939 branch February 18, 2026 23:14

rkhachatryan mentioned this pull request Feb 26, 2026

[FLINK-39167][runtime] Initialize source output before emitting final watermark #27684

Merged

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!

Uh oh!

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!

Uh oh!

Uh oh!

Uh oh!

Efrat19 left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

rkhachatryan commented Feb 12, 2026

Uh oh!

rkhachatryan commented Feb 13, 2026

Uh oh!

rkhachatryan commented Feb 18, 2026

Uh oh!

rkhachatryan commented Feb 18, 2026

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

5 participants

flinkbot commented Jan 19, 2026 •

edited

Loading