[jira] [Created] (FLINK-40501) WatermarkStatus overtakes queued watermarks in AsyncWaitOperator, causing watermark loss downstream

["Martijn Visser (Jira)" <ji...@apache.org>]

Martijn Visser created FLINK-40501:
--------------------------------------

             Summary: WatermarkStatus overtakes queued watermarks in AsyncWaitOperator, causing watermark loss downstream
                 Key: FLINK-40501
                 URL: https://issues.apache.org/jira/browse/FLINK-40501
             Project: Flink
          Issue Type: Bug
          Components: API / DataStream, Runtime / Task
            Reporter: Martijn Visser


{{AsyncWaitOperator.processWatermark}} enqueues watermarks into the ordered async work
queue ({{AsyncWaitOperator.java:286-293}}), but the operator has no
{{processWatermarkStatus}} override, so {{AbstractStreamOperator.processWatermarkStatus}}
forwards IDLE/ACTIVE immediately. A status element that arrives *after* a watermark can
therefore be emitted *before* it.

Demonstration (ORDERED mode, async completion withheld) — red on master:
arrival order {{[record, W100, IDLE]}} produces emission order
{{[WatermarkStatus(IDLE), record, W100]}}.

The reordering is not benign: feeding the emitted sequence {{[IDLE, W100]}} into a
downstream {{StatusWatermarkValve}} drops W100 entirely — watermarks arriving on an idle
channel are ignored. The watermark is lost, not delayed; downstream stalls until a later
watermark arrives on that edge, permanently if it was the final one (e.g. MAX_WATERMARK
at drain).

{{RowTimeMiniBatchAssignerOperator}} has the same shape by inspection: it buffers
watermarks until the mini-batch boundary but forwards status immediately (no
{{processWatermarkStatus}} override).

Proposed fix: route {{WatermarkStatus}} through the async queue with the same ordering
guarantees as watermarks (the recovery replay in {{open()}} currently throws on
non-record/watermark/latency elements and would need extension), or hold back the status
until previously queued watermarks have been flushed.



--
This message was sent by Atlassian Jira
(v8.20.10#820010)