[
https://issues.apache.org/jira/browse/FLINK-40371?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=18113947#comment-18113947
]
Sylwester Lachiewicz commented on FLINK-40371:
----------------------------------------------
Triage finding (2026-09-10): I traced the connector side first, since this is
filed against kafka-3.2.0. {{KafkaSourceReader}},
{{KafkaPartitionSplitReader#pauseOrResumeSplits}} and
{{KafkaSourceFetcherManager}} treat restored splits exactly like freshly
discovered ones, and per-split watermark state is re-created generically in
{{SourceOperator#open()}} before the reader registers. Nothing there is
Kafka-specific or restore-specific.
The mechanism that matches the reported shape is in flink-runtime:
[SourceCoordinator#announceCombinedWatermark|https://github.com/apache/flink/blob/master/flink-runtime/src/main/java/org/apache/flink/runtime/source/coordinator/SourceCoordinator.java#L184-L230]
only considers subtasks that have already sent a {{ReportedWatermarkEvent}}. A
subtask that has not reported yet is excluded from the minimum and receives no
{{WatermarkAlignmentEvent}}, so its max desired watermark stays unconstrained
until its own first report. That is the gap tracked by
[FLINK-36914|https://issues.apache.org/jira/browse/FLINK-36914], which names
restart as the trigger. It also explains why {{watermarkAlignmentDrift}} reads
near zero: the metric is computed only for subtasks already in the reporting
cohort. FLINK-40505 does not fit, because this scenario has no failover or
rescale.
Two things would confirm or kill the hypothesis from your job: the coordinator
log line {{Distributing maxAllowedWatermark=... to subTaskIds=...}} during the
divergence window (does it list fewer than all subtasks?), and the per-subtask
{{maxDesiredWatermark}} gauge over the same window. If the subtask list is
short right after restore, I suggest linking this ticket to FLINK-36914 and
moving the component to Runtime / Coordination.
_This comment was created with AI assistance._
> Watermark alignment does not consistently bound cross-subtask event-time
> drift after restoring from a savepoint
> ---------------------------------------------------------------------------------------------------------------
>
> Key: FLINK-40371
> URL: https://issues.apache.org/jira/browse/FLINK-40371
> Project: Flink
> Issue Type: Bug
> Components: Connectors / Kafka, Runtime / Coordination
> Affects Versions: kafka-3.2.0
> Reporter: Kushagra Tyagi
> Priority: Major
> Attachments: catchup_using_savepoint.png, catchup_using_timestamp.png
>
>
> *Setup*
> We configured a KafkaSource with forBoundedOutOfOrderness(100ms) and
> withWatermarkAlignment(groupName, maxDrift), across roughly 48 parallel
> source subtasks reading a partitioned Kafka topic.
> *What we observed*
> The behavior differs depending on how the job is started when it has a large
> backlog to catch up on.
> * Fresh start from a timestamp (OffsetsInitializer.timestamp(...), no prior
> state):
> Subtasks stayed tightly aligned. Per record event time lag
> (currentEmitEventTimeLag) stayed within roughly the configured max drift plus
> a small margin across all subtasks, even with a backlog representing many
> hours of lag. We observed this holding at up to about 22 hours of lag, with
> drift staying within a few minutes across subtasks. See attachment:
> catchup_using_timestamp.png.
> * Restart from a savepoint or checkpoint after the job was stopped, with a
> comparable backlog:
> Subtasks diverged from each other by multiple minutes.
> currentEmitEventTimeLag grew apart over roughly the first 15 minutes after
> restart, well beyond the configured max drift, before beginning to converge.
> See attachment: catchup_using_savepoint.png.
> *Key finding*
> watermarkAlignmentDrift stayed within or near the configured bound in both
> cases. currentEmitEventTimeLag diverged significantly, but only in the
> savepoint restore case.
> This suggests the coordinator's reported alignment drift can become decoupled
> from the actual per subtask processing pace, specifically in the window right
> after restoring from a savepoint. It holds correctly when subtasks start from
> a clean state.
> *Metrics referenced*
> - flink.operator.currentEmitEventTimeLag: per subtask, per record event time
> lag at emission. This is the metric that showed the real divergence between
> subtasks after a savepoint resume.
> - flink.operator.watermarkAlignmentDrift: the coordinator's own reported
> alignment drift. This stayed within bound in both cases, including the case
> where real lag diverged.
> *Ruled out*
> We checked whether the savepoint itself held stale or misaligned state. It
> did not. The job was well aligned at the moment it was stopped, so the
> savepoint captured an already aligned position for every subtask. The
> divergence appears after resume despite that, which suggests the difference
> is in how the alignment coordinator initializes or reconnects with subtasks
> on restore, not in the correctness of the saved state itself.
> *Expected behavior*
> Cross subtask watermark alignment should bound actual per record event time
> drift within roughly the configured max drift, regardless of whether the job
> started fresh or resumed from a savepoint.
> *Actual behavior*
> Alignment holds correctly on a fresh start. After a savepoint restore,
> subtasks can diverge by multiple minutes for roughly 10 to 15 minutes before
> converging.
> *Affects*
> Flink 1.19.x, flink connector kafka 3.2.0 for 1.19. Not yet tested on other
> versions.
> *Note*
> This is based on production metrics observed on a live pipeline, not an
> isolated minimal reproducer. Happy to help build one if that would help with
> triage.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
