[jira] (FLINK-40504) FLIP-27 source never re-emits WatermarkStatus.ACTIVE after idleness unless its watermark strictly advances

["yongfu.gao (Jira)" <ji...@apache.org>]

    [ https://issues.apache.org/jira/browse/FLINK-40504 ]


    yongfu.gao deleted comment on FLINK-40504:
    ------------------------------------

was (Author: JIRAUSER314604):
[~martijnvisser] The PR for this issue is up: https://github.com/apache/flink/pull/29239

It addresses the root cause of the two mechanisms described in the ticket:

* WatermarkOutputMultiplexer now reports the idle -> active transition of the combined status to the underlying output (previously it only reported markIdle()), and a watermark on a previously idle multiplexed output triggers the combined update, so the re-activation is no longer delayed until the next periodic emit.
* registerNewOutput() reports the combined status immediately, which covers the new-split case of the same defect described in FLINK-22926.
* WatermarkToDataOutput.emitWatermark() marks the output active before the monotonicity guard, as required by the WatermarkOutput contract.

Verification: 7 new tests (4 in WatermarkOutputMultiplexerTest, 1 in WatermarkToDataOutputTest and 2 end-to-end in SourceOperatorEventTimeTest). All 7 were verified to fail without the corresponding production change and to pass with it; a broader regression over 20 watermark-related test classes (137 tests) passes on a clean build. Azure CI on ca8d2501 is green.

This is complementary to #29037 (FLINK-40499) and does not touch the files changed there.

Could a committer please take a look, or point me to someone who can review it?

> 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)