[
https://issues.apache.org/jira/browse/FLINK-40504?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Roman Khachatryan updated FLINK-40504:
--------------------------------------
Release Note: Idle source outputs are now re-activated on any sign of
activity, not only on an advancing watermark. A FLIP-27 source that had been
announced as IDLE downstream stayed IDLE after it resumed while its watermark
stayed behind the maximum already reached, so its records could be dropped as
late events; the same applied when a new split was registered while the source
was idle. WatermarksWithIdleness now also marks the output active again on the
first record after idleness, instead of waiting for the wrapped generator to
emit an advancing watermark.
> FLIP-27 source never re-emits WatermarkStatus.ACTIVE after idleness unless
> its watermark strictly advances
> ----------------------------------------------------------------------------------------------------------
>
> Key: FLINK-40504
> URL: https://issues.apache.org/jira/browse/FLINK-40504
> Project: Flink
> Issue Type: Bug
> Components: API / Core, Runtime / Task
> Reporter: Martijn Visser
> Assignee: yongfu.gao
> Priority: Major
> Labels: pull-request-available
>
> Two compounding mechanisms leave a resumed FLIP-27 source announced-IDLE
> downstream:
> # {{WatermarkOutputMultiplexer.updateCombinedWatermark()}} calls
> {{underlyingOutput.markIdle()}} when combined-idle but has no
> {{underlyingOutput.markActive()}} counterpart; the per-output {{markActive}}
> implementations only mutate internal partial state.
> # {{WatermarkToDataOutput.emitWatermark}} returns at the monotonicity guard
> ({{newWatermark <= maxWatermarkSoFar}}) *before* {{markActiveInternally()}}
> ({{WatermarkToDataOutput.java:77-85}}).
> The only path back to ACTIVE is therefore a strictly larger watermark.
> Scenario: all
> splits of a subtask go idle; the combined watermark is flushed to the max M
> across
> splits (FLINK-38454). A split then resumes with backlog whose watermarks are
> <= M
> (e.g. a recovered Kafka partition catching up). Result — proven by red unit
> tests on
> both layers: *zero* calls reach the underlying output (no watermark, no
> ACTIVE), even
> via {{onPeriodicEmit}}; the edge stays IDLE downstream indefinitely while
> records flow.
> Downstream min-watermark is then driven by the other inputs, past the resumed
> backlog,
> and its records are dropped as late.
> Every sibling path re-activates eagerly on resume: {{StatusWatermarkValve}}
> (explicitly,
> since FLINK-40475), the legacy {{StreamSourceContexts}} (ACTIVE on every
> collect), the
> table {{WatermarkAssignerOperator}} (ACTIVE on first record), and DataStream
> V2
> (explicit idle=false watermark on next record). The FLIP-27 multiplexer path
> is the only
> one without a re-activation signal.
> FLINK-22926 (open since 2021) describes the new-split registration special
> case of this
> same defect; it should be closed into this ticket.
> Note: the fix is a semantics decision (emit ACTIVE eagerly on
> record/markActive, aligning
> with all other paths) rather than a local patch - it also composes with
> FLINK-40499: a
> resumed-but-not-advanced source is exactly the still-IDLE state in which its
> final
> MAX_WATERMARK is then dropped at drain.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
