[
https://issues.apache.org/jira/browse/FLINK-41312?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
ASF GitHub Bot updated FLINK-41312:
-----------------------------------
Labels: pull-request-available (was: )
> Fluss pipeline source does not retire dropped or expired partitions while
> running
> ---------------------------------------------------------------------------------
>
> Key: FLINK-41312
> URL: https://issues.apache.org/jira/browse/FLINK-41312
> Project: Flink
> Issue Type: Bug
> Components: Flink CDC
> Reporter: liting liu
> Priority: Major
> Labels: pull-request-available
>
> h2. Problem
> When a partition is dropped or expires while a Fluss pipeline source is
> running, the logical table and other partitions can remain subscribed. The
> source can retain splits and SDK subscriptions for the absent partition,
> repeatedly request metadata for its old partition ID, and delay or stall
> consumption of surviving partitions while the job remains RUNNING.
> This issue covers retirement of partitions that are already active in a
> running source. Filtering obsolete splits during whole-job checkpoint
> recovery is tracked separately by FLINK-41313.
> h2. Reproduction
> # Create a partitioned Fluss primary-key table with partitions P_old and
> P_live.
> # Start a CDC pipeline for the table with periodic partition discovery
> enabled, write records into both partitions, and wait until both are consumed.
> # Keep the pipeline running and drop P_old through the Fluss Admin API,
> leaving the logical table and P_live intact. Partition expiration can cause
> the same condition.
> # Continue writing records into P_live.
> # Observe the source's partition subscriptions, metadata requests, and
> consumption progress. The absent partition should be retired without
> interrupting the surviving partition.
> h2. Root cause
> * Dynamic subscription removal currently operates at logical-table
> granularity. A partition can disappear without removing its table from the
> subscribed table set.
> * The reader therefore retains already-active partition splits, and the split
> reader's existing whole-table removal path does not unsubscribe those
> partition/bucket subscriptions.
> * Missing-partition metadata retries in the shared fetching path can delay
> valid partitions. Refreshing metadata alone does not retire an obsolete
> subscription.
> h2. Proposed fix
> * Build on the complete partition inventory introduced by FLINK-41313 and
> compare it with the reader's active splits after a successful discovery.
> * Route absent-partition cleanup through the fetcher thread. Unsubscribe the
> exact partition ID and buckets, and cancel their active or queued snapshot
> scans.
> * Return removed split IDs through the normal finished-split path so reader
> checkpoint state is cleaned consistently.
> * Preserve the logical table, schema state, and resources required by
> surviving partitions.
> * Keep existing checkpoint serialization formats and split IDs. Same-name
> partition recreation with a different physical ID remains a separate concern.
--
This message was sent by Atlassian Jira
(v10.3.26#10030026)
