[
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
* 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.
was:
h3. 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.
h3. Reproduction
# Create a partitioned Fluss PK 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; obsolete partition IDs must not
block consumption.
h3. Root cause
* Discovery reconciles logical-table subscriptions and initializes new physical
partitions, but does not retire disappeared partitions from an otherwise
subscribed table.
* Enumerator assigned/pending state can therefore retain obsolete physical
paths. A table-level subscription snapshot does not distinguish a valid
restored split from a split belonging to a deleted partition.
* The obsolete split can reach the SDK scanner, which keeps attempting metadata
refreshes for the missing numeric partition ID. Synchronous metadata waits
consume fetch time and delay valid buckets handled by the same reader.
h3. Proposed fix
* Reconcile physical partitions only after a successful complete discovery. A
failed or incomplete metadata query must not imply partition deletion.
* Clear obsolete coordinator assignment/pending state before publishing the
complete physical-partition inventory to readers, including newly registered or
restored readers.
* Keep restored splits staged until the inventory arrives, filter absent
partitions before SDK subscription, and reject obsolete initialization results
or returned splits using current bucket identities.
* Execute precise SDK partition/bucket unsubscribe and snapshot cancellation on
the fetcher thread. Finish the affected splits through normal reader state
cleanup while preserving surviving partitions and logical-table
schema/resources.
* Preserve existing checkpoint serializer versions and split IDs. Same-name
partition recreation with a new physical ID requires separate generation-aware
handling.
> 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
> * 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.
--
This message was sent by Atlassian Jira
(v10.3.26#10030026)
