This is an automated email from the ASF dual-hosted git repository.
MartijnVisser pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/master by this push:
new bb773b5fb8a [FLINK-40475][runtime] Fix watermark loss and stall in
StatusWatermarkValve when subpartitions realign after idleness (#29024)
bb773b5fb8a is described below
commit bb773b5fb8a4a87dd4efc3a8ae261b7e08e602a6
Author: Martijn Visser <[email protected]>
AuthorDate: Fri Aug 28 14:15:57 2026 +0200
[FLINK-40475][runtime] Fix watermark loss and stall in StatusWatermarkValve
when subpartitions realign after idleness (#29024)
[FLINK-40475][runtime] Fix watermark loss and stall in StatusWatermarkValve
when subpartitions realign after idleness
The emission logic of StatusWatermarkValve relies on two invariants over
the set of watermark-aligned subpartitions: it is only empty when no
subpartition is active, and whenever it is non-empty its min watermark
equals the last output watermark. Both were violated once an unaligned
subpartition - one that went idle, resumed, and only partially caught up
to the last output watermark - was involved:
- The FLINK-7728 all-idle flush was skipped unless the last subpartition
to become idle held the current min watermark. That assumption doesn't
hold when unaligned subpartitions are involved, since an unaligned
subpartition's watermark is invisible to the min-derive path. This made
the final watermark depend on the order in which inputs became idle -
exactly the defect FLINK-7728 was meant to fix.
- The idle->active branch 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 stalled indefinitely,
emitted only once an even larger watermark arrived on it.
Fix both by re-deriving from the aligned set unconditionally instead of
relying on the broken shortcut:
- Flush the max watermark unconditionally once all subpartitions go idle.
findAndOutputMaxWatermarkAcrossAllSubpartitions only emits when the max
advances past the last output watermark, so this cannot regress
monotonicity, and it keeps the valve consistent with
CombinedWatermarkStatus, which has flushed unconditionally since
FLINK-38454.
- Re-derive (and possibly emit) the min watermark after a subpartition
realigns, after the ACTIVE status is emitted so downstream inputs don't
drop the watermark while they still consider the input idle.
Documents the invariants on alignedSubpartitionStatuses.
Generated-by: Claude Code (claude-fable-5)
---
.../watermarkstatus/StatusWatermarkValve.java | 64 +++++-----
.../watermarkstatus/StatusWatermarkValveTest.java | 137 +++++++++++++++++++++
2 files changed, 173 insertions(+), 28 deletions(-)
diff --git
a/flink-runtime/src/main/java/org/apache/flink/streaming/runtime/watermarkstatus/StatusWatermarkValve.java
b/flink-runtime/src/main/java/org/apache/flink/streaming/runtime/watermarkstatus/StatusWatermarkValve.java
index 29f63c959d4..0109da532ec 100644
---
a/flink-runtime/src/main/java/org/apache/flink/streaming/runtime/watermarkstatus/StatusWatermarkValve.java
+++
b/flink-runtime/src/main/java/org/apache/flink/streaming/runtime/watermarkstatus/StatusWatermarkValve.java
@@ -69,7 +69,20 @@ public class StatusWatermarkValve {
/** The last watermark status emitted from the valve. */
private WatermarkStatus lastOutputWatermarkStatus;
- /** A heap-based priority queue to help find the minimum watermark. */
+ /**
+ * A heap-based priority queue to help find the minimum watermark. It
contains exactly the
+ * watermark-aligned subpartitions, for which the following invariants
hold:
+ *
+ * <ul>
+ * <li>only active subpartitions can be aligned
+ * <li>the watermark of an aligned subpartition is never smaller than
{@link
+ * #lastOutputWatermark}
+ * <li>whenever the queue is non-empty, its minimum watermark equals
{@link
+ * #lastOutputWatermark}; this is maintained by re-deriving the min
watermark via {@link
+ * #findAndOutputNewMinWatermarkAcrossAlignedSubpartitions} after
every change to the
+ * aligned set or to an aligned subpartition's watermark
+ * </ul>
+ */
private final HeapPriorityQueue<SubpartitionStatus>
alignedSubpartitionStatuses;
/** Whether there are multiple subpartitions transmitted through the same
input channel. */
@@ -221,30 +234,20 @@ public class StatusWatermarkValve {
// if all subpartitions of the valve are now idle, we need to
output an idle stream
// status from the valve (this also marks the valve as idle)
if
(!SubpartitionStatus.hasActiveSubpartitions(subpartitionStatuses)) {
-
- // now that all subpartitions are idle and no subpartitions
will continue to advance
- // its
- // watermark,
- // we should "flush" all watermarks across all subpartitions;
effectively, this
- // means
- // emitting
- // the max watermark across all subpartitions as the new
watermark. Also, since we
- // already try to advance
- // the min watermark as subpartitions individually become
IDLE, here we only need to
- // perform the flush
- // if the watermark of the last active subpartition that just
became idle is the
- // current
- // min watermark.
- if (subpartitionStatus.watermark == lastOutputWatermark) {
- findAndOutputMaxWatermarkAcrossAllSubpartitions(output);
- }
+ // now that all subpartitions are idle and no subpartition
will continue to advance
+ // its watermark, we should "flush" all watermarks across all
subpartitions;
+ // effectively, this means emitting the max watermark across
all subpartitions as
+ // the new watermark, so that the eventual watermark is
independent of the order in
+ // which the subpartitions became idle. This also keeps the
valve consistent with
+ // the unconditional flush in CombinedWatermarkStatus.
+ findAndOutputMaxWatermarkAcrossAllSubpartitions(output);
lastOutputWatermarkStatus = WatermarkStatus.IDLE;
output.emitWatermarkStatus(lastOutputWatermarkStatus);
- } else if (subpartitionStatus.watermark == lastOutputWatermark) {
- // if the watermark of the subpartition that just became idle
equals the last output
- // watermark (the previous overall min watermark), we may be
able to find a new
- // min watermark from the remaining aligned subpartitions
+ } else {
+ // the min watermark across the remaining aligned
subpartitions may be larger than
+ // the last output watermark, now that the subpartition that
just became idle is no
+ // longer part of the aligned set
findAndOutputNewMinWatermarkAcrossAlignedSubpartitions(output);
}
} else if (watermarkStatus.isActive() &&
subpartitionStatus.watermarkStatus.isIdle()) {
@@ -252,21 +255,26 @@ public class StatusWatermarkValve {
subpartitionStatus.watermarkStatus = WatermarkStatus.ACTIVE;
// if the last watermark of the subpartition, before it was marked
idle, is still
- // larger than
- // the overall last output watermark of the valve, then we can set
the subpartition to
- // be
- // aligned already.
+ // larger than the overall last output watermark of the valve,
then we can set the
+ // subpartition to be aligned already.
if (subpartitionStatus.watermark >= lastOutputWatermark) {
markWatermarkAligned(subpartitionStatus);
}
// if the valve was previously marked to be idle, mark it as
active and output an active
- // stream
- // status because at least one of the subpartitions is now active
+ // stream status because at least one of the subpartitions is now
active
if (lastOutputWatermarkStatus.isIdle()) {
lastOutputWatermarkStatus = WatermarkStatus.ACTIVE;
output.emitWatermarkStatus(lastOutputWatermarkStatus);
}
+
+ // the subpartition that just realigned may hold the new min
watermark of the aligned
+ // set (e.g. if it is the only aligned subpartition), in which
case its watermark must
+ // be emitted now; otherwise it would be stalled until an even
larger watermark arrives
+ // on the subpartition. This must happen after the ACTIVE status
was emitted, so that
+ // downstream inputs do not drop the watermark while they still
consider this input
+ // idle.
+ findAndOutputNewMinWatermarkAcrossAlignedSubpartitions(output);
}
}
diff --git
a/flink-runtime/src/test/java/org/apache/flink/streaming/runtime/watermarkstatus/StatusWatermarkValveTest.java
b/flink-runtime/src/test/java/org/apache/flink/streaming/runtime/watermarkstatus/StatusWatermarkValveTest.java
index e139f4c4866..e222d531def 100644
---
a/flink-runtime/src/test/java/org/apache/flink/streaming/runtime/watermarkstatus/StatusWatermarkValveTest.java
+++
b/flink-runtime/src/test/java/org/apache/flink/streaming/runtime/watermarkstatus/StatusWatermarkValveTest.java
@@ -336,6 +336,53 @@ class StatusWatermarkValveTest {
assertThat(valveOutput.popLastSeenOutput()).isNull();
}
+ /**
+ * Tests that the max watermark across all channels is flushed once all
inputs become idle, even
+ * when the last channel to become idle is unaligned, i.e. it went idle,
resumed being active,
+ * and only partially caught up to the last output watermark. The eventual
watermark advancement
+ * result must be independent of the order in which the inputs become idle
(FLINK-7728).
+ */
+ @Test
+ void
testMultipleInputFlushMaxWatermarkOnceAllInputsBecomeIdleWithUnalignedLastChannel()
+ throws Exception {
+ StatusWatermarkOutput valveOutput = new StatusWatermarkOutput();
+ StatusWatermarkValve valve = new StatusWatermarkValve(3);
+
+ valve.inputWatermark(new Watermark(100), 0, valveOutput);
+ valve.inputWatermark(new Watermark(50), 1, valveOutput);
+ valve.inputWatermark(new Watermark(70), 2, valveOutput);
+ assertThat(valveOutput.popLastSeenOutput()).isEqualTo(new
Watermark(50));
+ assertThat(valveOutput.popLastSeenOutput()).isNull();
+
+ // channel 1 becomes idle; the new min watermark is computed from
channels 0 and 2
+ valve.inputWatermarkStatus(WatermarkStatus.IDLE, 1, valveOutput);
+ assertThat(valveOutput.popLastSeenOutput()).isEqualTo(new
Watermark(70));
+ assertThat(valveOutput.popLastSeenOutput()).isNull();
+
+ // channel 1 resumes being active, but stays unaligned since its
watermark (50) has not
+ // caught up to the last output watermark (70)
+ valve.inputWatermarkStatus(WatermarkStatus.ACTIVE, 1, valveOutput);
+
assertThat(valve.getSubpartitionStatus(1).isWatermarkAligned).isFalse();
+ assertThat(valveOutput.popLastSeenOutput()).isNull();
+
+ // channel 1 partially catches up; its watermark is recorded but it
remains unaligned
+ valve.inputWatermark(new Watermark(60), 1, valveOutput);
+ assertThat(valve.getSubpartitionStatus(1).watermark).isEqualTo(60);
+
assertThat(valve.getSubpartitionStatus(1).isWatermarkAligned).isFalse();
+ assertThat(valveOutput.popLastSeenOutput()).isNull();
+
+ valve.inputWatermarkStatus(WatermarkStatus.IDLE, 0, valveOutput);
+ valve.inputWatermarkStatus(WatermarkStatus.IDLE, 2, valveOutput);
+ assertThat(valveOutput.popLastSeenOutput()).isNull();
+
+ // the unaligned channel 1 is the last channel to become idle; the max
watermark across
+ // all channels must still be flushed
+ valve.inputWatermarkStatus(WatermarkStatus.IDLE, 1, valveOutput);
+ assertThat(valveOutput.popLastSeenOutput()).isEqualTo(new
Watermark(100));
+
assertThat(valveOutput.popLastSeenOutput()).isEqualTo(WatermarkStatus.IDLE);
+ assertThat(valveOutput.popLastSeenOutput()).isNull();
+ }
+
/**
* Tests that when idle channels become active again, they need to "catch
up" with the latest
* watermark before they are considered for min watermark computation
again.
@@ -431,6 +478,96 @@ class StatusWatermarkValveTest {
assertThat(valveOutput.popLastSeenOutput()).isNull();
}
+ /**
+ * Tests that when a channel reactivates and realigns while no other
aligned channels remain
+ * (all remaining channels are idle or unaligned), the min watermark
across aligned channels is
+ * re-derived and emitted; otherwise the realigned channel's watermark
would be stalled until an
+ * even higher watermark arrives on it.
+ */
+ @Test
+ void
testWatermarkAdvancesWhenReactivatedChannelBecomesOnlyAlignedChannel() throws
Exception {
+ StatusWatermarkOutput valveOutput = new StatusWatermarkOutput();
+ StatusWatermarkValve valve = new StatusWatermarkValve(3);
+
+ valve.inputWatermark(new Watermark(10), 0, valveOutput);
+ valve.inputWatermark(new Watermark(7), 1, valveOutput);
+
+ // channel 2 becomes idle; the new min watermark is computed from
channels 0 and 1
+ valve.inputWatermarkStatus(WatermarkStatus.IDLE, 2, valveOutput);
+ assertThat(valveOutput.popLastSeenOutput()).isEqualTo(new
Watermark(7));
+ assertThat(valveOutput.popLastSeenOutput()).isNull();
+
+ // channel 2 resumes being active but is unaligned (its watermark has
never advanced)
+ valve.inputWatermarkStatus(WatermarkStatus.ACTIVE, 2, valveOutput);
+ assertThat(valveOutput.popLastSeenOutput()).isNull();
+
+ // channels 0 and 1 become idle; channel 2 is the only active channel
but it is unaligned,
+ // so no watermark can be emitted
+ valve.inputWatermarkStatus(WatermarkStatus.IDLE, 0, valveOutput);
+ valve.inputWatermarkStatus(WatermarkStatus.IDLE, 1, valveOutput);
+ assertThat(valveOutput.popLastSeenOutput()).isNull();
+
+ // channel 0 resumes being active; its watermark (10) has already
caught up to the last
+ // output watermark (7), so it realigns and the min watermark must
advance to 10
+ valve.inputWatermarkStatus(WatermarkStatus.ACTIVE, 0, valveOutput);
+ assertThat(valve.getSubpartitionStatus(0).isWatermarkAligned).isTrue();
+ assertThat(valveOutput.popLastSeenOutput()).isEqualTo(new
Watermark(10));
+ assertThat(valveOutput.popLastSeenOutput()).isNull();
+
+ // the unaligned channel 2 becoming idle again must not change anything
+ valve.inputWatermarkStatus(WatermarkStatus.IDLE, 2, valveOutput);
+ assertThat(valveOutput.popLastSeenOutput()).isNull();
+ }
+
+ /**
+ * Tests that when a channel reactivates and realigns with a watermark
larger than the last
+ * output watermark, the min watermark across aligned channels is
re-derived and emitted, and
+ * that the flush once all inputs become idle afterwards does not emit a
duplicate watermark.
+ */
+ @Test
+ void testRealignmentAfterResumeActiveEmitsNewMinWatermark() throws
Exception {
+ StatusWatermarkOutput valveOutput = new StatusWatermarkOutput();
+ StatusWatermarkValve valve = new StatusWatermarkValve(3);
+
+ valve.inputWatermark(new Watermark(10), 0, valveOutput);
+ valve.inputWatermark(new Watermark(20), 1, valveOutput);
+ valve.inputWatermark(new Watermark(30), 2, valveOutput);
+ assertThat(valveOutput.popLastSeenOutput()).isEqualTo(new
Watermark(10));
+ assertThat(valveOutput.popLastSeenOutput()).isNull();
+
+ // channel 2 becomes idle; its watermark (30) is above the min, so
nothing changes
+ valve.inputWatermarkStatus(WatermarkStatus.IDLE, 2, valveOutput);
+ assertThat(valveOutput.popLastSeenOutput()).isNull();
+
+ // channel 0 becomes idle; the new min watermark is computed from
channel 1
+ valve.inputWatermarkStatus(WatermarkStatus.IDLE, 0, valveOutput);
+ assertThat(valveOutput.popLastSeenOutput()).isEqualTo(new
Watermark(20));
+ assertThat(valveOutput.popLastSeenOutput()).isNull();
+
+ // channel 0 resumes being active, but stays unaligned (10 < 20)
+ valve.inputWatermarkStatus(WatermarkStatus.ACTIVE, 0, valveOutput);
+
assertThat(valve.getSubpartitionStatus(0).isWatermarkAligned).isFalse();
+ assertThat(valveOutput.popLastSeenOutput()).isNull();
+
+ // channel 1 becomes idle; no aligned channels remain, so nothing can
be emitted
+ valve.inputWatermarkStatus(WatermarkStatus.IDLE, 1, valveOutput);
+ assertThat(valveOutput.popLastSeenOutput()).isNull();
+
+ // channel 2 resumes being active; its watermark (30) is above the
last output watermark
+ // (20), so it realigns and the min watermark must advance to 30
+ valve.inputWatermarkStatus(WatermarkStatus.ACTIVE, 2, valveOutput);
+ assertThat(valve.getSubpartitionStatus(2).isWatermarkAligned).isTrue();
+ assertThat(valveOutput.popLastSeenOutput()).isEqualTo(new
Watermark(30));
+ assertThat(valveOutput.popLastSeenOutput()).isNull();
+
+ // all channels become idle; the flush must not emit another
watermark, since the max
+ // watermark across all channels (30) has already been emitted
+ valve.inputWatermarkStatus(WatermarkStatus.IDLE, 0, valveOutput);
+ valve.inputWatermarkStatus(WatermarkStatus.IDLE, 2, valveOutput);
+
assertThat(valveOutput.popLastSeenOutput()).isEqualTo(WatermarkStatus.IDLE);
+ assertThat(valveOutput.popLastSeenOutput()).isNull();
+ }
+
private static class StatusWatermarkOutput implements
PushingAsyncDataInput.DataOutput {
private BlockingQueue<StreamElement> allOutputs = new
LinkedBlockingQueue<>();
