[jira] [Commented] (FLINK-40371) Watermark alignment does not consistently bound cross-subtask event-time drift after restoring from a savepoint

["Kushagra Tyagi (Jira)" <ji...@apache.org>]

    [ https://issues.apache.org/jira/browse/FLINK-40371?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=18103917#comment-18103917 ] 

Kushagra Tyagi commented on FLINK-40371:
----------------------------------------

*Additional details and reproduction evidence*
Environment:
- Flink: 1.19.1
- flink-connector-kafka: 3.2.0-1.19

Config:
WatermarkStrategy<AccessLog> watermarkStrategy = WatermarkStrategy
        .<AccessLog>forBoundedOutOfOrderness(Duration.ofMillis(100))
        .withTimestampAssigner(new AccessLogTimestampAssigner())
        .withWatermarkAlignment("my-alignment-group", Duration.ofSeconds(alignmentDuration));
Tested with alignmentDuration set to both 30 seconds and 3 seconds. Same divergence in both cases.

We reproduced the difference directly, same job, same size backlog, only the restart method changed:

- Restart from a savepoint after stopping the job: subtask level currentEmitEventTimeLag diverges by multiple minutes across subtasks, growing for roughly the first 15 or more minutes after restart before beginning to close. watermarkAlignmentDrift stays within or near the configured bound the entire time, so the coordinator level metric does not show the problem, only the actual per record event time lag does.
- Fresh start from a timestamp with a comparable or larger backlog: all subtasks track together almost exactly, same average, same max, no meaningful divergence, even at nearly a full day of lag.

We also 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.

> 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
>    Affects Versions: kafka-3.2.0
>            Reporter: Kushagra Tyagi
>            Priority: Major
>         Attachments: catchup_using_savepoint.png, catchup_using_timestamp.png
>
>
> We configured a KafkaSource with forBoundedOutOfOrderness(100ms) and withWatermarkAlignment(groupName, maxDrift), across roughly 48 parallel source subtasks reading a partitioned Kafka topic.
> We observed a consistent difference in behavior depending on how the job was started when it had a large backlog to catch up on:
>  - Fresh start from a timestamp (OffsetsInitializer.timestamp(...), no prior state): subtasks' per-record event-time lag (currentEmitEventTimeLag) stayed tightly bounded across all subtasks, within roughly the configured max drift plus a small margin, even when the backlog represented many hours of lag (observed up to ~22 hours of lag with drift staying within a few minutes across subtasks). (catchup_using_timestamp.png)
>  - Restart from a savepoint/checkpoint after the job was stopped, with a comparable backlog to catch up: subtasks' currentEmitEventTimeLag diverged from each other by multiple minutes, growing over roughly the first 15+ minutes after restart, well beyond the configured max drift. (catchup_using_savepoint.png)
> Notably, the watermarkAlignmentDrift metric stayed within/near the configured bound in both cases, while currentEmitEventTimeLag diverged significantly in the savepoint-restore case. This suggests the coordinator's reported/enforced alignment drift may become decoupled from the actual per-subtask processing pace specifically during the window right after restoring from a savepoint, while it holds correctly for a subtask population that starts from a clean/fresh state.
> *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 an extended period before (if ever — not yet confirmed) converging.
> *Affects:* Flink 1.19.x, flink-connector-kafka 3.2.0-1.19 (observed; 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 triage.



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