Martijn Visser created FLINK-40499:
--------------------------------------
Summary: Final MAX_WATERMARK is dropped at drain when the source's
last announced status is IDLE
Key: FLINK-40499
URL: https://issues.apache.org/jira/browse/FLINK-40499
Project: Flink
Issue Type: Bug
Components: Runtime / Task
Reporter: Martijn Visser
When a job drains (bounded input finishing, or stop-with-savepoint --drain),
{{StreamTask.endData}} calls {{advanceToEndOfEventTime()}}, which for source
tasks emits
{{Watermark.MAX_WATERMARK}} directly into the operator chain output
({{SourceOperatorStreamTask.java:217-219}}; same in {{SourceStreamTask}} and
{{MultipleInputStreamTask}}). However, both
{{RecordWriterOutput.emitWatermark}} and
{{ChainingOutput.emitWatermark}} silently drop any watermark while the last
announced
{{WatermarkStatus}} is IDLE. A source subtask that is idle at drain time
therefore never
delivers its final MAX watermark.
If at least one sibling channel is active at drain, its MAX_WATERMARK rescues
downstream
(idle channels are excluded from the min). If *all* channels are idle — e.g. a
bounded
Kafka topic whose partitions ran dry before the idle timeout, then the job
finishes —
downstream completes with the watermark stuck at the last flushed value:
event-time
timers and windows waiting for {{Long.MAX_VALUE}} never fire, and the final
output is
silently missing.
Reproduction (both red on master):
* Task level: a {{SourceOperatorStreamTask}} (mock source whose reader calls
{{output.markIdle()}} and then finishes) drains with task output
{{[WatermarkStatus(IDLE), EndOfData(DRAIN)]}} — no MAX watermark — versus
{{[MAX_WATERMARK, EndOfData(DRAIN)]}} in the existing green test
{{testEmittingMaxWatermarkAfterReadingAllRecords}} for a non-idle source.
* Unit level: {{ChainingOutput}}: {{emitWatermarkStatus(IDLE)}} followed by
{{emitWatermark(MAX_WATERMARK)}} — the MAX watermark is dropped at the idle
guard
({{ChainingOutput.java:117-120}}; identical guard in
{{RecordWriterOutput.java:149-152}}).
No rescue mechanism exists: {{advanceToEndOfEventTime}} emits MAX without a
preceding
ACTIVE status; {{EndOfData}} / {{EndOfPartitionEvent}} do nothing
watermark-related
downstream; and non-source tasks' {{advanceToEndOfEventTime}} is an explicit
no-op.
Proposed fix: emit {{WatermarkStatus.ACTIVE}} before the MAX watermark in the
{{advanceToEndOfEventTime}} paths — the pattern already used by
{{TimestampsAndWatermarksOperator.WatermarkEmitter}} and the table
{{WatermarkAssignerOperator.processWatermark}}, and the same ordering rule
FLINK-40475
applies inside {{StatusWatermarkValve}}. Longer term, a FINISHED watermark
status
(FLINK-38477) would subsume this.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
