[BP-1.20][FLINK-39481][task] Execute deferrable mails when finishing an operator#28650
Merged
MartijnVisser merged 2 commits intoJul 6, 2026
Merged
Conversation
Since FLINK-35528 the continuation mail of an interrupted timer-firing chain is deferrable and hidden from tryYield(). The finish drain in StreamOperatorWrapper therefore skipped it, so a task reaching end of input mid-firing forwarded EndOfData before the pending watermark, and downstream operators discarded the state of windows that only fire on that watermark. Restore the pre-FLINK-35528 behaviour by draining with a new yield variant that matches deferrable mails by their original priority, used only in that place. Generated-by: Claude Code (Fable 5) (cherry picked from commit 83055ae)
… of input Exercises the full SQL pipeline (two-phase split-distinct window aggregation on RocksDB) under backpressure with the configuration in which all CI failures of WindowDistinctAggregateITCase occurred, comparing repeated checkpoint/failure/restore/end-of-input rounds against a golden run within a fixed time budget. Without the FLINK-39481 fix this loses the trailing window family in roughly one of three executions. Generated-by: Claude Code (Fable 5) (cherry picked from commit bbc688c)
Collaborator
pnowojski
approved these changes
Jul 6, 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
3 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.
What is the purpose of the change
Backport of #28608 to release-1.20.
An interrupted timer-firing chain defers its continuation (and the downstream watermark emission) to a deferrable mail. The operator-finish drain skipped deferrable mails, so
EndOfDatacould overtake the deferred watermark and downstream operators never received it, losing the state of windows that only fire on it (FLINK-39481, observed asWindowDistinctAggregateITCaseexactly-once data loss after restore).The fix executes deferrable mails when finishing an operator.
Brief change log
Cherry-pick of 83055ae and bbc688c from master. The production fix applied cleanly (on release-1.20 the task/mailbox classes live under
flink-streaming-java, and the master-only urgent-mails context was dropped from theTaskMailboxImplhunks).The new test needed a port to the release-1.20 API: there is no
useInterruptibleTimershook and noexecution.checkpointing.unaligned.interruptible-timers.enabledoption on this branch; interruptible timers are operator-opt-in. The test operator therefore follows the branch-native pattern of the neighboring operator in the same test class (implementsYieldingOperator, wires aMailboxWatermarkProcessorinopen()/processWatermark), and the 2.x-only config lines are dropped. The ported test still exercises the fixed path: the release-1.20MailboxWatermarkProcessorsubmits its watermark continuation withMailOptions.deferrable(), which is exactly the mail the fix now drains at operator finish.Verifying this change
StreamOperatorWrapperTest,MailboxExecutorImplTestandUnalignedCheckpointsInterruptibleTimersTest(including the newtestDeferredWatermarkIsEmittedBeforeEndOfData, the red-green test for this fix) pass on this branch.WindowDistinctAggregateStressITCaseis included in the cherry-pick and is validated by CI.Does this pull request potentially affect one of the following parts:
Same as #28608: no dependencies, no public API, no serializers, no per-record code paths; touches the operator-finish path of the task lifecycle.
Documentation
Was generative AI tooling used to co-author this PR?
Generated-by: Claude Fable 5