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)
