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