[
https://issues.apache.org/jira/browse/FLINK-40499?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Martijn Visser updated FLINK-40499:
-----------------------------------
Release Note: When a source was idle at the moment it finished (end of a
bounded input or stop-with-savepoint --drain), the final watermark was dropped
and downstream event-time windows and timers did not fire. The source now marks
itself active before the final watermark, so operators that implement
processWatermarkStatus can see one extra ACTIVE status just before end of input.
> 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
>
> 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)
