[
https://issues.apache.org/jira/browse/FLINK-41312?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
liting liu updated FLINK-41312:
-------------------------------
Description:
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 restore-time partition inventory introduced by FLINK-41313.
Refresh reader inventories when the discovered bucket set changes, including
partition changes within the same subscribed table.
* Reconcile runtime pending assignments and snapshot bookkeeping, and reject
obsolete asynchronous initialization results or returned splits using current
bucket identities.
* Validate incoming and already-active reader splits against each successful
complete inventory and request cleanup for absent partitions.
* Route cleanup through the fetcher thread. Unsubscribe the exact partition ID
and buckets, cancel active or queued snapshot scans, and return removed IDs
through the normal finished-split path.
* 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.
was:
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.
> 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 restore-time partition inventory introduced by FLINK-41313.
> Refresh reader inventories when the discovered bucket set changes, including
> partition changes within the same subscribed table.
> * Reconcile runtime pending assignments and snapshot bookkeeping, and reject
> obsolete asynchronous initialization results or returned splits using current
> bucket identities.
> * Validate incoming and already-active reader splits against each successful
> complete inventory and request cleanup for absent partitions.
> * Route cleanup through the fetcher thread. Unsubscribe the exact partition
> ID and buckets, cancel active or queued snapshot scans, and return removed
> IDs through the normal finished-split path.
> * 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)
