[
https://issues.apache.org/jira/browse/FLINK-40499?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Martijn Visser reassigned FLINK-40499:
--------------------------------------
Assignee: Martijn Visser
> 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
> Assignee: Martijn Visser
> Priority: Major
> Labels: pull-request-available
>
> 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)
