[
https://issues.apache.org/jira/browse/FLINK-41313?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=18136958#comment-18136958
]
liting liu commented on FLINK-41313:
------------------------------------
please assign to me
> Fluss pipeline source does not unsubscribe 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
> The Fluss pipeline source does not retire a dropped or expired physical
> partition while its logical table remains in the TableDiscoverer subscription.
> After the partition disappears from Fluss metadata, readers can retain its
> splits and keep requesting metadata for the old partition ID. LogFetcher
> repeatedly reports:
> {code}
> Receive PartitionNotExistException when update metadata, ignore it
> org.apache.fluss.exception.PartitionNotExistException: The partition id
> '<expired-id>' does not exist or you don't have permission to access it.
> {code}
> The job can remain RUNNING and continue checkpointing while source fetcher
> threads repeatedly wait in MetadataUtils.sendMetadataRequestAndRebuildCluster
> / LogFetcher.checkAndUpdateMetadata. This consumes fetch time and delays
> other valid buckets handled by the same reader.
> h3. Scope and current-code check
> Checked Apache Flink CDC master at db45d06a1e35bc6be87f7b58913c768f8f299d3e.
> FLINK-40575 / PR #4527 implements logical-table unsubscription. Its
> table-removal detection compares TablePath sets. If a table stays subscribed
> and only one partition disappears, no table-removal event is generated. The
> existing removeTables cleanup therefore does not retire that partition.
> The current bucket-discovery path initializes new buckets but does not
> reconcile removed physical partitions. The reader events and cleanup entry
> points still operate at logical-table granularity.
> * [Enumerator: subscription update and table-level
> removal|https://github.com/apache/flink-cdc/blob/db45d06a1e35bc6be87f7b58913c768f8f299d3e/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-fluss/src/main/java/org/apache/flink/cdc/connectors/fluss/source/enumerator/FlussSourceEnumerator.java#L294-L367]
> * [Reader: table subscription
> events|https://github.com/apache/flink-cdc/blob/db45d06a1e35bc6be87f7b58913c768f8f299d3e/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-fluss/src/main/java/org/apache/flink/cdc/connectors/fluss/source/reader/FlussSourceReader.java#L179-L208]
> * [Split reader: existing whole-table
> cleanup|https://github.com/apache/flink-cdc/blob/db45d06a1e35bc6be87f7b58913c768f8f299d3e/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-fluss/src/main/java/org/apache/flink/cdc/connectors/fluss/source/reader/FlussSplitReader.java#L410-L445]
> h3. Reproduction / regression-test scenario
> # Create a partitioned Fluss table with multiple buckets and at least two
> live partitions.
> # Start a streaming Fluss pipeline source and wait until readers subscribe to
> both partitions. Keep the table in the discoverer's subscription throughout
> the test.
> # Drop one existing partition, or allow it to expire.
> # Allow a successful partition discovery cycle to observe that the partition
> is absent.
> # Verify whether the old partition's buckets are unsubscribed and its split
> state is retired while the remaining partitions continue to be read.
> # Take a checkpoint and restore. Verify that a still-absent partition is not
> assigned or subscribed again.
> The repeated metadata-request behavior was observed in a deployed CDC
> 3.7-SNAPSHOT / Fluss 1.1-SNAPSHOT job without a job restart during partition
> rotation. The latest upstream master was checked statically as above; a
> standalone automated reproduction on that exact upstream commit remains to be
> added.
> h3. Expected behavior
> * Detect removed physical partitions after a successful authoritative
> metadata discovery, independently of logical-table subscription changes.
> * Notify the affected readers and invoke the SDK's precise partition/bucket
> unsubscribe operation.
> * Mark removed splits finished through the normal reader cleanup path and
> reconcile enumerator assignment/pending state so checkpoint recovery does not
> revive absent partitions.
> * Continue consuming remaining partitions and other tables. Do not emit a
> DropTableEvent or remove the sink table merely because one source partition
> expired.
> * Do not treat a failed or incomplete metadata listing as an authoritative
> partition deletion.
> * Handle later partition re-creation using its physical partition identity,
> without reviving stale split offsets.
> h3. Related work
> * FLINK-40575 / https://github.com/apache/flink-cdc/pull/4527 covers
> whole-table unsubscription, not this single-partition lifecycle.
> * https://github.com/apache/fluss/issues/3466 discusses repeated metadata
> requests for expired partitions and SDK/server mitigations. This issue tracks
> the CDC connector's partition-removal and checkpoint-cleanup lifecycle.
> * https://github.com/apache/fluss/issues/2202 proposes per-partition metadata
> failure reporting; it does not provide the missing CDC cleanup flow.
--
This message was sent by Atlassian Jira
(v10.3.26#10030026)
