[
https://issues.apache.org/jira/browse/FLINK-41313?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
liting liu updated FLINK-41313:
-------------------------------
Description:
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.
was:
h3. Problem
The Fluss pipeline source may fail to resume data consumption when restoring a
checkpoint that references partitions which have been dropped or expired. The
logical table and other partitions can still exist, but the source retains
obsolete partition splits and keeps requesting metadata for their old partition
IDs. The job can remain RUNNING while valid partitions stop making progress.
The same missing partition-retirement handling also affects a running source
when a partition disappears while its logical table remains subscribed.
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
>
> 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.
--
This message was sent by Atlassian Jira
(v10.3.26#10030026)
