[
https://issues.apache.org/jira/browse/FLINK-40504?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Martijn Visser updated FLINK-40504:
-----------------------------------
Description:
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.
was:
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
> 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
> Priority: Major
>
> 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)
