Admaing opened a new pull request, #29239:
URL: https://github.com/apache/flink/pull/29239
## What is the purpose of the change
A FLIP-27 source that has been announced as IDLE downstream stays IDLE after
it resumes, unless its
next watermark strictly advances the combined watermark. Two mechanisms
compound:
1. `WatermarkOutputMultiplexer.updateCombinedWatermark()` reports the
combined idle state via
`underlyingOutput.markIdle()`, but has no counterpart for the transition
back to active; the
per-output `markActive()` implementations only clear internal partial
state.
2. `WatermarkToDataOutput.emitWatermark()` returns at the monotonicity guard
(`newWatermark <= maxWatermarkSoFar`) before reaching
`markActiveInternally()`, although the
`WatermarkOutput` contract states that emitting a watermark implicitly
marks the stream active.
A resumed split with backlog whose watermarks are `<=` the maximum flushed
while all splits were
idle therefore reaches the underlying output with neither a watermark nor an
ACTIVE status, and its
records are dropped as late downstream. Every sibling path re-activates
eagerly
(`StatusWatermarkValve`, legacy `StreamSourceContexts`, the table
`WatermarkAssignerOperator`, and
DataStream V2); the FLIP-27 multiplexer path was the only one without a
re-activation signal.
This also covers the new-split special case of the same defect
(FLINK-22926): registering a new
output while the combined status is idle now re-activates it right away,
instead of only on the
next periodic emit.
## Brief change log
- `WatermarkOutputMultiplexer` keeps track of the idle state last reported
to the underlying output
and reports the idle -> active transition via `markActive()`.
- A watermark on a previously idle multiplexed output triggers the combined
update, so the
re-activation is propagated eagerly instead of only on the next periodic
emit.
- `registerNewOutput()` updates and reports the combined status right away,
so registering a new
split while the source is idle marks the underlying output active
immediately (FLINK-22926).
- `WatermarkToDataOutput.emitWatermark()` marks the output active before the
monotonicity guard.
- Tests: 4 new `WatermarkOutputMultiplexerTest` cases, 1 new
`WatermarkToDataOutputTest` case and
2 new end-to-end `SourceOperatorEventTimeTest` cases.
## Verifying this change
This change added tests and can be verified as follows:
- `mvn -pl flink-core -am -Dtest=WatermarkOutputMultiplexerTest test`
- `mvn -pl flink-runtime -am
-Dtest='WatermarkToDataOutputTest,SourceOperatorEventTimeTest' test`
- All 7 new tests were verified to fail without the corresponding production
change and to pass
with it (red/green verification by reverting the three production files,
and separately by
reverting the `registerNewOutput()` update).
- Broader regression over all watermark-related test classes in
`flink-core-api`, `flink-core` and
`flink-runtime` (20 classes, 137 tests) passes on a clean build.
-
`SourceOperatorEventTimeTest#testResumingSplitWithoutAdvancingWatermarkEmitsActive`
reproduces the
scenario from the ticket: all splits go idle, the combined watermark is
flushed to the maximum, a
split resumes with backlog whose watermarks do not advance the combined
watermark, and the edge
must emit ACTIVE.
- `SourceOperatorEventTimeTest#testRegisteringNewSplitWhileIdleEmitsActive`
covers FLINK-22926. It
deliberately does not advance the processing time service, because a
periodic watermark emit
would re-activate the output lazily as well and would mask the immediate
re-activation under test.
Notes for reviewers:
- The only `TimestampsAndWatermarks.WatermarkUpdateListener` implementation
is `SourceOperator`.
`emitWatermark()` now calls `markActiveInternally()` before
`updateCurrentEffectiveWatermark()`
for advancing watermarks (the downstream event order ACTIVE -> watermark
is unchanged).
`updateIdle()` only writes a field that `checkWatermarkAlignment()` does
not read, so the swapped
listener callback order has no observable effect.
- `unregisterOutput()` still does not update the combined status, so
removing the last active output
while others are idle is only reflected on the next periodic emit. This is
a pre-existing,
bounded-delay issue of the same family and is deliberately left out of
this change.
- This is complementary to #29037 (FLINK-40499), which re-activates at drain
time in the stream
tasks; this PR fixes the root cause and does not touch those files. It
also covers the new-split
case described in FLINK-22926, which can be closed together with this
change.
## Does this pull request potentially affect one of the following parts:
- Dependencies (does it add or upgrade a dependency): no
- The public API: no (`WatermarkOutputMultiplexer` and
`WatermarkToDataOutput` are `@Internal`)
- The serializers: no
- The runtime per-record code paths (performance sensitive): yes, minimally
â
`ImmediateOutput.emitWatermark()` gains one field read and a branch; the
combined update (which
iterates all splits) still only runs on a watermark advance, an idle ->
active transition, or a
registration
- Anything that affects deployment or recovery: no
- The S3 file system connector: no
## Documentation
- Does this pull request introduce a new feature? no
- If yes, how is the feature documented? not applicable
---
- [X] Yes (please specify the tool below)
Generated-by: DeepSeek Harness (deepseek-v4.1-flash)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
