[
https://issues.apache.org/jira/browse/FLINK-40504?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=18118427#comment-18118427
]
yongfu.gao commented on FLINK-40504:
------------------------------------
Suggested Release Note for this fix (I don't have permission to set the Release
Note field myself):
{quote}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.{quote}
Could a committer please set it on this ticket when reviewing?
> 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)
