[
https://issues.apache.org/jira/browse/FLINK-41313?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
liting liu updated FLINK-41313:
-------------------------------
Description:
h2. Problem
Restoring a Fluss pipeline source from a checkpoint may fail to resume data
consumption if a referenced partition was dropped or expired after the
checkpoint was taken, even though the table and other partitions still exist.
Recovery should skip absent partitions and resume the surviving partitions from
their checkpointed positions.
h2. Reproduction
# Create a partitioned Fluss primary-key table with at least two partitions and
multiple buckets. Enable periodic discovery in the CDC pipeline.
# Start the pipeline, consume data from both partitions, and complete a
retained externalized checkpoint.
# Cancel the whole job and wait for termination.
# Drop one partition through the Fluss Admin API, or allow it to expire. Keep
the logical table and another partition available.
# Write new rows to the surviving partition.
# Restore the whole job from the earlier checkpoint with the same operator
UIDs. The source should skip the absent partition and continue from the
surviving partition's checkpoint position.
h2. Root cause
* Discovery reconciles logical-table subscriptions and initializes new physical
partitions, but does not filter absent partitions from restored coordinator
state.
* Assigned reader splits and their consumption positions are restored
independently from reader operator state. Cleaning the enumerator's pending
assignments alone cannot remove these obsolete reader splits.
* A table-level subscription snapshot cannot distinguish a live restored split
from one belonging to an absent partition. An obsolete split can therefore
reach the SDK scanner and trigger repeated metadata requests for the missing
partition ID, delaying valid splits handled by the same reader.
h2. Proposed fix
* Obtain a complete physical-partition inventory from the first successful
discovery. A failed metadata query must not imply partition deletion.
* Remove absent partitions from restored coordinator assignments and pending
splits before publishing the first subscription snapshot.
* Keep independently restored reader splits staged until that first snapshot
arrives, then filter absent partitions before SDK activation.
* Preserve normal runtime split activation, initialization callbacks,
checkpoint serializer versions, and split IDs. Same-name partition recreation
with a new physical ID remains a separate concern.
* Scope this issue to whole-job checkpoint recovery. Continuous inventory
refresh, stale-split handling, and retirement of already-active partitions
belong to FLINK-41312.
was:
h2. Problem
Restoring a Fluss pipeline source from a checkpoint may fail to resume data
consumption if a referenced partition was dropped or expired after the
checkpoint was taken, even though the table and other partitions still exist.
Recovery should skip absent partitions and resume the surviving partitions from
their checkpointed positions.
h2. Reproduction
# Create a partitioned Fluss primary-key table with at least two partitions and
multiple buckets. Enable periodic discovery in the CDC pipeline.
# Start the pipeline, consume data from both partitions, and complete a
retained externalized checkpoint.
# Cancel the whole job and wait for termination.
# Drop one partition through the Fluss Admin API, or allow it to expire. Keep
the logical table and another partition available.
# Write new rows to the surviving partition.
# Restore the whole job from the earlier checkpoint with the same operator
UIDs. The source should skip the absent partition and continue from the
surviving partition's checkpoint position.
h2. Root cause
* Discovery reconciles logical-table subscriptions and initializes new physical
partitions, but does not filter absent partitions from restored coordinator
state.
* Assigned reader splits and their consumption positions are restored
independently from reader operator state. Cleaning the enumerator's pending
assignments alone cannot remove these obsolete reader splits.
* A table-level subscription snapshot cannot distinguish a live restored split
from one belonging to an absent partition. An obsolete split can therefore
reach the SDK scanner and trigger repeated metadata requests for the missing
partition ID, delaying valid splits handled by the same reader.
h2. Proposed fix
* Reconcile restored physical-partition state only after a successful complete
discovery. A failed metadata query must not imply partition deletion.
* Clear obsolete coordinator assignment and pending state before publishing the
complete physical-partition inventory to readers.
* Keep restored reader splits staged until the inventory arrives, filter absent
partitions before SDK subscription, and reject obsolete initialization results
or returned splits using current bucket identities.
* Preserve existing checkpoint serializer versions and split IDs. Same-name
partition recreation with a new physical ID remains a separate concern.
* Scope this issue to whole-job checkpoint recovery. Retirement of
already-active partitions during continuous operation is tracked separately by
FLINK-41312.
> Fluss pipeline source fails to recover from checkpoints referencing dropped
> or expired partitions
> -------------------------------------------------------------------------------------------------
>
> Key: FLINK-41313
> URL: https://issues.apache.org/jira/browse/FLINK-41313
> Project: Flink
> Issue Type: Bug
> Components: Flink CDC
> Reporter: liting liu
> Priority: Major
> Labels: pull-request-available
>
> h2. Problem
> Restoring a Fluss pipeline source from a checkpoint may fail to resume data
> consumption if a referenced partition was dropped or expired after the
> checkpoint was taken, even though the table and other partitions still exist.
> Recovery should skip absent partitions and resume the surviving partitions
> from their checkpointed positions.
> h2. Reproduction
> # Create a partitioned Fluss primary-key table with at least two partitions
> and multiple buckets. Enable periodic discovery in the CDC pipeline.
> # Start the pipeline, consume data from both partitions, and complete a
> retained externalized checkpoint.
> # Cancel the whole job and wait for termination.
> # Drop one partition through the Fluss Admin API, or allow it to expire. Keep
> the logical table and another partition available.
> # Write new rows to the surviving partition.
> # Restore the whole job from the earlier checkpoint with the same operator
> UIDs. The source should skip the absent partition and continue from the
> surviving partition's checkpoint position.
> h2. Root cause
> * Discovery reconciles logical-table subscriptions and initializes new
> physical partitions, but does not filter absent partitions from restored
> coordinator state.
> * Assigned reader splits and their consumption positions are restored
> independently from reader operator state. Cleaning the enumerator's pending
> assignments alone cannot remove these obsolete reader splits.
> * A table-level subscription snapshot cannot distinguish a live restored
> split from one belonging to an absent partition. An obsolete split can
> therefore reach the SDK scanner and trigger repeated metadata requests for
> the missing partition ID, delaying valid splits handled by the same reader.
> h2. Proposed fix
> * Obtain a complete physical-partition inventory from the first successful
> discovery. A failed metadata query must not imply partition deletion.
> * Remove absent partitions from restored coordinator assignments and pending
> splits before publishing the first subscription snapshot.
> * Keep independently restored reader splits staged until that first snapshot
> arrives, then filter absent partitions before SDK activation.
> * Preserve normal runtime split activation, initialization callbacks,
> checkpoint serializer versions, and split IDs. Same-name partition recreation
> with a new physical ID remains a separate concern.
> * Scope this issue to whole-job checkpoint recovery. Continuous inventory
> refresh, stale-split handling, and retirement of already-active partitions
> belong to FLINK-41312.
--
This message was sent by Atlassian Jira
(v10.3.26#10030026)
