[
https://issues.apache.org/jira/browse/FLINK-41313?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
liting liu updated FLINK-41313:
-------------------------------
Description:
h3. Problem
The Fluss pipeline source may fail to resume data consumption when restoring a
checkpoint that references partitions which have been dropped or expired. The
logical table and other partitions can still exist, but the source retains
obsolete partition splits and keeps requesting metadata for their old partition
IDs. The job can remain RUNNING while valid partitions stop making progress.
The same missing partition-retirement handling also affects a running source
when a partition disappears while its logical table remains subscribed.
h3. Reproduction
# Create a partitioned Fluss PK 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; obsolete partition IDs must not
block consumption.
h3. Root cause
* Discovery reconciles logical-table subscriptions and initializes new physical
partitions, but does not retire disappeared partitions from an otherwise
subscribed table.
* Enumerator assigned/pending state can therefore retain obsolete physical
paths. A table-level subscription snapshot does not distinguish a valid
restored split from a split belonging to a deleted partition.
* The obsolete split can reach the SDK scanner, which keeps attempting metadata
refreshes for the missing numeric partition ID. Synchronous metadata waits
consume fetch time and delay valid buckets handled by the same reader.
h3. Proposed fix
* Reconcile physical partitions only after a successful complete discovery. A
failed or incomplete metadata query must not imply partition deletion.
* Clear obsolete coordinator assignment/pending state before publishing the
complete physical-partition inventory to readers, including newly registered or
restored readers.
* Keep restored splits staged until the inventory arrives, filter absent
partitions before SDK subscription, and reject obsolete initialization results
or returned splits using current bucket identities.
* Execute precise SDK partition/bucket unsubscribe and snapshot cancellation on
the fetcher thread. Finish the affected splits through normal reader state
cleanup while preserving surviving partitions and logical-table
schema/resources.
* Preserve existing checkpoint serializer versions and split IDs. Same-name
partition recreation with a new physical ID requires separate generation-aware
handling.
was:
h3. Checkpoint recovery failure
A completed checkpoint can still reference a Fluss partition that is dropped or
expires before the entire CDC job is restored. The source must discard that
absent partition before SDK subscription and resume the surviving partitions
from their checkpoint positions. Otherwise recovery can remain blocked on
repeated metadata requests for obsolete partition IDs, even when the job status
is RUNNING.
Reproduction scenario:
# Create a partitioned Fluss PK table with two partitions and multiple buckets;
start the CDC pipeline and consume both partitions.
# Complete an externalized checkpoint, then cancel the entire job and wait for
termination.
# Physically drop one partition with the Admin API and verify the partition is
absent.
# Write new rows to the surviving partition while the job is stopped.
# Start a new job from the exact earlier checkpoint. It should consume the new
survivor rows without replaying the initial rows and continue completing
checkpoints.
h3. Proposed fix
A proposed fix reconciles coordinator assignments with successful complete
partition discovery, sends complete physical inventories to readers, and
filters restored splits before SDK subscription. Active obsolete splits are
unsubscribed/cancelled on the fetcher thread and finished through the normal
reader state-cleanup path. Surviving partition and logical-table resources are
preserved; existing checkpoint serializer versions are unchanged.
The full checkpoint -> whole-job cancellation -> actual dropPartition ->
restore-from-the-same-checkpoint scenario passed in earliest and full startup
modes on both Flink 1.20.3 / Java 11 and Flink 2.2.0 / Java 17 (4/4
executions). The test verifies the actual restored checkpoint path, exact
survivor rows without baseline replay, two further successful checkpoints, and
RUNNING status with automatic failure restart disabled. This is local embedded
Flink/Fluss validation, not deployment validation.
* [Checkpoint recovery
regression|https://github.com/litiliu/flink-cdc/commit/4cd5896485522d64af32730cb33cba783f9bbaeb]
* [Partition retirement
fix|https://github.com/litiliu/flink-cdc/commit/18499592a0e2b2f508c3e7aeba8168ebf21651b0]
Same-name partition recreation with a new physical ID, recovery during an
in-progress KV snapshot, and deletion racing after successful discovery remain
follow-up boundaries of this proposed fix.
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 upstream code was checked at the commit above; automated
regression coverage and local validation of the proposed fix have now been
added as described above.
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.
> 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
> Priority: Major
>
> h3. Problem
> The Fluss pipeline source may fail to resume data consumption when restoring
> a checkpoint that references partitions which have been dropped or expired.
> The logical table and other partitions can still exist, but the source
> retains obsolete partition splits and keeps requesting metadata for their old
> partition IDs. The job can remain RUNNING while valid partitions stop making
> progress.
> The same missing partition-retirement handling also affects a running source
> when a partition disappears while its logical table remains subscribed.
> h3. Reproduction
> # Create a partitioned Fluss PK 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; obsolete partition IDs must not
> block consumption.
> h3. Root cause
> * Discovery reconciles logical-table subscriptions and initializes new
> physical partitions, but does not retire disappeared partitions from an
> otherwise subscribed table.
> * Enumerator assigned/pending state can therefore retain obsolete physical
> paths. A table-level subscription snapshot does not distinguish a valid
> restored split from a split belonging to a deleted partition.
> * The obsolete split can reach the SDK scanner, which keeps attempting
> metadata refreshes for the missing numeric partition ID. Synchronous metadata
> waits consume fetch time and delay valid buckets handled by the same reader.
> h3. Proposed fix
> * Reconcile physical partitions only after a successful complete discovery. A
> failed or incomplete metadata query must not imply partition deletion.
> * Clear obsolete coordinator assignment/pending state before publishing the
> complete physical-partition inventory to readers, including newly registered
> or restored readers.
> * Keep restored splits staged until the inventory arrives, filter absent
> partitions before SDK subscription, and reject obsolete initialization
> results or returned splits using current bucket identities.
> * Execute precise SDK partition/bucket unsubscribe and snapshot cancellation
> on the fetcher thread. Finish the affected splits through normal reader state
> cleanup while preserving surviving partitions and logical-table
> schema/resources.
> * Preserve existing checkpoint serializer versions and split IDs. Same-name
> partition recreation with a new physical ID requires separate
> generation-aware handling.
--
This message was sent by Atlassian Jira
(v10.3.26#10030026)
