[FLINK-40475][runtime] Fix watermark loss and stall in StatusWatermarkValve when subpartitions realign after idleness - #29024
Merged
Conversation
…StatusWatermarkValve become idle Previously the flush introduced by FLINK-7728 was skipped unless the last subpartition to become idle held the current min watermark. When the last subpartition to become idle was unaligned - it went idle, resumed being active, and only partially caught up to the last output watermark - the flush was skipped and the final watermark depended on the order in which the inputs became idle: the exact defect FLINK-7728 was created to fix. The guard was not a valid optimization, because the assumption behind it (the min watermark was already advanced as subpartitions individually became idle) does not hold when unaligned subpartitions are involved. Removing it cannot regress the output watermark, since findAndOutputMaxWatermarkAcrossAllSubpartitions only emits when the max advances past the last output watermark. This also makes the valve consistent with CombinedWatermarkStatus, which flushes unconditionally since FLINK-38454. Generated-by: Claude Code (claude-fable-5)
…aligns after idleness The emission logic of StatusWatermarkValve relies on the invariant that, whenever the set of watermark-aligned subpartitions is non-empty, its min watermark equals the last output watermark. The idle->active branch broke this invariant: it re-added a caught-up subpartition to the aligned set without re-deriving the min. When no other aligned subpartition remained, the realigned subpartition's watermark was stalled indefinitely - it was only emitted once an even larger watermark arrived on that subpartition. Re-derive (and possibly emit) the min watermark after the reactivation, after the ACTIVE status was emitted so that downstream inputs do not drop the watermark while they still consider the input idle. Also drop the equivalent guard on the partial-idle path, which silently depended on the unbroken invariant, and document the invariants on the aligned set. Generated-by: Claude Code (claude-fable-5)
Collaborator
rkhachatryan
approved these changes
Aug 27, 2026
rkhachatryan
left a comment
Contributor
There was a problem hiding this comment.
LGTM, thanks for the fix!
davidradl
approved these changes
Aug 28, 2026
1 task
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
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.
What is the purpose of the change
When a subpartition of
StatusWatermarkValvereactivates after idleness and rejoins the aligned set, the min watermark across aligned subpartitions is not re-derived. This breaks the invariant that the aligned min equals the last output watermark, which two guards (subpartitionStatus.watermark == lastOutputWatermark) silently rely on, with two symptoms:CombinedWatermarkStatus, which flushes unconditionally since FLINK-38454.Brief change log
alignedSubpartitionStatusesVerifying this change
This change added tests and can be verified as follows:
testMultipleInputFlushMaxWatermarkOnceAllInputsBecomeIdleWithUnalignedLastChannel, reproducing the order-dependent all-idle flush (fails on master: emits onlyWatermarkStatus(IDLE)instead ofWatermark(100))testWatermarkAdvancesWhenReactivatedChannelBecomesOnlyAlignedChannelandtestRealignmentAfterResumeActiveEmitsNewMinWatermark, reproducing the stalled watermark on realignment (fail on master: no watermark emitted)StatusWatermarkValveTestare unaffected; also verified green:OneInputStreamTaskTest,TwoInputStreamTaskTest,WatermarkOutputMultiplexerTest,SourceOperatorSplitWatermarkAlignmentTestDoes this pull request potentially affect one of the following parts:
@Public(Evolving): noDocumentation
Note on behavior: jobs that previously ended with a lower watermark depending on idle-arrival order will now deterministically receive the flushed max watermark (timers/windows in that range fire); a release note is proposed on the JIRA ticket.
Was generative AI tooling used to co-author this PR?
Generated-by: Claude Code (claude-fable-5)