[
https://issues.apache.org/jira/browse/FLINK-41313?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Yanquan Lv resolved FLINK-41313.
--------------------------------
Fix Version/s: cdc-3.8.0
Assignee: liting liu
Resolution: Fixed
Merged in master via 49fec143631bd045b69882e132b5115032baea7c.
> 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
> Assignee: liting liu
> Priority: Major
> Labels: pull-request-available
> Fix For: cdc-3.8.0
>
>
> 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)
