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

["Martijn Visser (Jira)" <ji...@apache.org>]

Martijn Visser created FLINK-40504:
--------------------------------------

             Summary: 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


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



--
This message was sent by Atlassian Jira
(v8.20.10#820010)