[
https://issues.apache.org/jira/browse/FLINK-40371?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Kushagra Tyagi updated FLINK-40371:
-----------------------------------
Description:
*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.
was:
*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.
*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.
> 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)
