liting liu created FLINK-41313:
----------------------------------
Summary: 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
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)
