[
https://issues.apache.org/jira/browse/FLINK-41313?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
ASF GitHub Bot updated FLINK-41313:
-----------------------------------
Labels: pull-request-available (was: )
> 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
>
> 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)
