[
https://issues.apache.org/jira/browse/FLINK-40499?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Martijn Visser resolved FLINK-40499.
------------------------------------
Fix Version/s: 2.2.2
2.3.1
2.4.0
Resolution: Fixed
Fixed in apache/flink
master: ebfb1b0c945d3a769e5c17184f198854b75df0f3
release-2.4: d61c59c83095506d3e842f2e3c9045368b46fb0f
release-2.3: 32749e25c7106a310e9878a0af40fedd05ad25ab
release-2.2: 73509baed7131c446e18e8285662fdccc272abe8
> 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
> Affects Versions: 1.20.5, 2.2.2, 2.3.1, 2.4.0
> Reporter: Martijn Visser
> Assignee: Martijn Visser
> Priority: Major
> Labels: pull-request-available
> Fix For: 2.2.2, 2.3.1, 2.4.0
>
>
> 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
(v10.3.26#10030026)
