This is an automated email from the ASF dual-hosted git repository. MartijnVisser pushed a commit to branch release-2.3 in repository https://gitbox.apache.org/repos/asf/flink.git
commit ec0bfc6baf2a594e97b9a18738b0a79c602ddd45 Author: yongfu.gao <[email protected]> AuthorDate: Sun Sep 27 01:36:33 2026 +0800 [FLINK-40504][runtime] Mark WatermarkToDataOutput active on non-advancing watermarks emitWatermark() returned at the monotonicity guard before marking the output active, although the WatermarkOutput contract states that emitting a watermark implicitly marks the stream active. A source that resumes with watermarks that are not larger than the max watermark so far therefore stayed announced as idle downstream, and its records could be dropped as late. The activation is now reported before the guard. Source implementers can observe this change, so it needs a release note. Generated-by: DeepSeek Harness (deepseek-v4.1-flash) --- .../api/operators/source/WatermarkToDataOutput.java | 18 +++++++++++++----- .../operators/source/WatermarkToDataOutputTest.java | 15 +++++++++++++++ 2 files changed, 28 insertions(+), 5 deletions(-) diff --git a/flink-runtime/src/main/java/org/apache/flink/streaming/api/operators/source/WatermarkToDataOutput.java b/flink-runtime/src/main/java/org/apache/flink/streaming/api/operators/source/WatermarkToDataOutput.java index a283443634a..2aced695b0b 100644 --- a/flink-runtime/src/main/java/org/apache/flink/streaming/api/operators/source/WatermarkToDataOutput.java +++ b/flink-runtime/src/main/java/org/apache/flink/streaming/api/operators/source/WatermarkToDataOutput.java @@ -74,16 +74,24 @@ public final class WatermarkToDataOutput implements WatermarkOutput { @Override public void emitWatermark(Watermark watermark) { final long newWatermark = watermark.getTimestamp(); - if (newWatermark <= maxWatermarkSoFar) { - return; + final boolean watermarkAdvanced = newWatermark > maxWatermarkSoFar; + if (watermarkAdvanced) { + maxWatermarkSoFar = newWatermark; + watermarkEmitted.updateCurrentEffectiveWatermark(maxWatermarkSoFar); } - maxWatermarkSoFar = newWatermark; - watermarkEmitted.updateCurrentEffectiveWatermark(maxWatermarkSoFar); - try { + // Emitting a watermark implicitly marks this output as active, independently of whether + // the watermark advances the effective watermark (see WatermarkOutput). Otherwise a + // source that resumes with watermarks that are not larger than the max watermark so far + // would stay idle downstream. The activation must therefore not be gated on the + // monotonicity check below. markActiveInternally(); + if (!watermarkAdvanced) { + return; + } + output.emitWatermark( new org.apache.flink.streaming.api.watermark.Watermark(newWatermark)); } catch (ExceptionInChainedOperatorException e) { diff --git a/flink-runtime/src/test/java/org/apache/flink/streaming/api/operators/source/WatermarkToDataOutputTest.java b/flink-runtime/src/test/java/org/apache/flink/streaming/api/operators/source/WatermarkToDataOutputTest.java index 2e261dfa08d..4b534f99c98 100644 --- a/flink-runtime/src/test/java/org/apache/flink/streaming/api/operators/source/WatermarkToDataOutputTest.java +++ b/flink-runtime/src/test/java/org/apache/flink/streaming/api/operators/source/WatermarkToDataOutputTest.java @@ -65,4 +65,19 @@ class WatermarkToDataOutputTest { assertThat(testingOutput.events) .contains(WatermarkStatus.IDLE, WatermarkStatus.ACTIVE, new Watermark(100L)); } + + @Test + void becomingActiveEmitsStatusOnNonAdvancingWatermark() { + final CollectingDataOutput<Object> testingOutput = new CollectingDataOutput<>(); + final WatermarkToDataOutput wmOutput = new WatermarkToDataOutput(testingOutput); + + wmOutput.emitWatermark(new org.apache.flink.api.common.eventtime.Watermark(100L)); + wmOutput.markIdle(); + + // the watermark does not advance, but the source is active again + wmOutput.emitWatermark(new org.apache.flink.api.common.eventtime.Watermark(100L)); + + assertThat(testingOutput.events) + .containsExactly(new Watermark(100L), WatermarkStatus.IDLE, WatermarkStatus.ACTIVE); + } }
