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 25892ef1a49ce035fbeea474071b24bb307fa1f7 Author: yongfu.gao <[email protected]> AuthorDate: Sun Sep 27 01:36:45 2026 +0800 [FLINK-40504][core] Report activity from WatermarksWithIdleness on events onEvent() reset the idleness timer without undoing the markIdle() it had emitted earlier, so the output stayed announced as idle until the wrapped generator produced an advancing watermark. It now reports activity on the first record and after each idle period, so an output whose generator never emits a watermark is still known to be active. WatermarksWithIdleness is @Public, so this is a separate commit to keep its release note easy to trace. Generated-by: DeepSeek Harness (deepseek-v4.1-flash) --- .../api/common/eventtime/WatermarksWithIdleness.java | 20 ++++++++++++++++++-- .../common/eventtime/WatermarksWithIdlenessTest.java | 20 ++++++++++++++++++++ 2 files changed, 38 insertions(+), 2 deletions(-) diff --git a/flink-core/src/main/java/org/apache/flink/api/common/eventtime/WatermarksWithIdleness.java b/flink-core/src/main/java/org/apache/flink/api/common/eventtime/WatermarksWithIdleness.java index 4cc651cf6e9..586190f14d7 100644 --- a/flink-core/src/main/java/org/apache/flink/api/common/eventtime/WatermarksWithIdleness.java +++ b/flink-core/src/main/java/org/apache/flink/api/common/eventtime/WatermarksWithIdleness.java @@ -32,7 +32,7 @@ import static org.apache.flink.util.Preconditions.checkNotNull; /** * A WatermarkGenerator that adds idleness detection to another WatermarkGenerator. If no events * come within a certain time (timeout duration) then this generator marks the stream as idle, until - * the next watermark is generated. + * an event arrives or the wrapped generator emits a watermark. */ @Public public class WatermarksWithIdleness<T> implements WatermarkGenerator<T> { @@ -43,6 +43,13 @@ public class WatermarksWithIdleness<T> implements WatermarkGenerator<T> { private boolean isIdleNow = false; + /** + * Whether the wrapped generator has ever seen an event. The first event has to report activity + * even though nothing was marked idle before, otherwise an output whose generator never emits a + * watermark would never be known to be active. + */ + private boolean isActivityReported = false; + /** * Creates a new WatermarksWithIdleness generator to the given generator idleness detection with * the given timeout. @@ -67,7 +74,16 @@ public class WatermarksWithIdleness<T> implements WatermarkGenerator<T> { public void onEvent(T event, long eventTimestamp, WatermarkOutput output) { watermarks.onEvent(event, eventTimestamp, output); idlenessTimer.activity(); - isIdleNow = false; + + if (isIdleNow || !isActivityReported) { + // A record is evidence of activity in its own right. Waiting for the wrapped generator + // to produce an advancing watermark instead would leave the output announced as idle + // for as long as the resumed input stays behind the watermark it reached before, and an + // output whose generator never emits a watermark would never be known to be active. + output.markActive(); + isIdleNow = false; + isActivityReported = true; + } } @Override diff --git a/flink-core/src/test/java/org/apache/flink/api/common/eventtime/WatermarksWithIdlenessTest.java b/flink-core/src/test/java/org/apache/flink/api/common/eventtime/WatermarksWithIdlenessTest.java index f2af73de2b0..1a89a841df0 100644 --- a/flink-core/src/test/java/org/apache/flink/api/common/eventtime/WatermarksWithIdlenessTest.java +++ b/flink-core/src/test/java/org/apache/flink/api/common/eventtime/WatermarksWithIdlenessTest.java @@ -107,6 +107,26 @@ class WatermarksWithIdlenessTest { assertThat(timer.checkIfIdle()).isTrue(); } + @Test + void testMarksActiveOnFirstEventAfterIdleness() { + final Duration idleTimeout = Duration.ofMillis(10); + final ManualClock clock = new ManualClock(System.nanoTime()); + final WatermarksWithIdleness<Long> watermarks = + new WatermarksWithIdleness<>(new NoWatermarksGenerator<>(), idleTimeout, clock); + final TestingWatermarkOutput output = new TestingWatermarkOutput(); + + watermarks.onPeriodicEmit(output); // start the idleness timer + clock.advanceTime(idleTimeout.plusMillis(1)); + watermarks.onPeriodicEmit(output); + assertThat(output.isIdle()).isTrue(); + + watermarks.onEvent(1L, 1L, output); + + assertThat(output.isIdle()) + .as("a record is activity in its own right, no watermark has to advance") + .isFalse(); + } + private static IdlenessTimer createTimerAndMakeIdle(ManualClock clock, Duration idleTimeout) { final IdlenessTimer timer = new IdlenessTimer(clock, idleTimeout);
