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 ba6bf2bf1dd41b0fe7ae67aff5ec6ad60ef83908 Author: yongfu.gao <[email protected]> AuthorDate: Sun Sep 27 01:36:33 2026 +0800 [FLINK-40504][core] Make PartialWatermark idleness tri-state An output's idleness is now UNKNOWN, ACTIVE or IDLE instead of a boolean that defaults to "not idle", and CombinedWatermarkStatus exposes hasActiveOutput() for callers that need to know whether an output actually reported being active. This is behaviour-neutral on its own: UNKNOWN combines exactly like the previous idle = false did, so it still holds the combined watermark back, and no caller uses hasActiveOutput() yet. Generated-by: DeepSeek Harness (deepseek-v4.1-flash) --- .../common/eventtime/CombinedWatermarkStatus.java | 41 ++++++++++++++++++++-- 1 file changed, 38 insertions(+), 3 deletions(-) diff --git a/flink-core/src/main/java/org/apache/flink/api/common/eventtime/CombinedWatermarkStatus.java b/flink-core/src/main/java/org/apache/flink/api/common/eventtime/CombinedWatermarkStatus.java index fb290ef957b..c94211305a1 100644 --- a/flink-core/src/main/java/org/apache/flink/api/common/eventtime/CombinedWatermarkStatus.java +++ b/flink-core/src/main/java/org/apache/flink/api/common/eventtime/CombinedWatermarkStatus.java @@ -46,6 +46,20 @@ final class CombinedWatermarkStatus { return idle; } + /** + * Returns true if any output has reported that it is active, which is what makes it safe to + * announce activity downstream. An output that has not said anything yet, or that only reported + * idleness, is not evidence that anything is producing. + */ + public boolean hasActiveOutput() { + for (PartialWatermark partialWatermark : partialWatermarks) { + if (partialWatermark.isActive()) { + return true; + } + } + return false; + } + public boolean remove(PartialWatermark o) { return partialWatermarks.remove(o); } @@ -108,8 +122,24 @@ final class CombinedWatermarkStatus { /** Per-output watermark state. */ static class PartialWatermark { + + /** + * The idleness the output has reported, if any. + * + * <p>A newly registered output is {@link #UNKNOWN}. It holds the combined watermark back + * the same way an active output does, but it is not evidence that anything is producing + * yet: outputs are registered for every assigned split, even when the reader never reports + * anything through them (for example when it emits everything through the main output, or + * when the watermark strategy does not use idleness at all). + */ + enum IdlenessState { + UNKNOWN, + ACTIVE, + IDLE + } + private long watermark = Long.MIN_VALUE; - private boolean idle = false; + private IdlenessState idlenessState = IdlenessState.UNKNOWN; private final WatermarkOutputMultiplexer.WatermarkUpdateListener onWatermarkUpdate; public PartialWatermark( @@ -139,11 +169,16 @@ final class CombinedWatermarkStatus { } private boolean isIdle() { - return idle; + return idlenessState == IdlenessState.IDLE; + } + + /** Returns true if this output has reported that it is active. */ + private boolean isActive() { + return idlenessState == IdlenessState.ACTIVE; } public void setIdle(boolean idle) { - this.idle = idle; + this.idlenessState = idle ? IdlenessState.IDLE : IdlenessState.ACTIVE; this.onWatermarkUpdate.onIdleUpdate(idle); } }
