[
https://issues.apache.org/jira/browse/FLINK-41313?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
liting liu updated FLINK-41313:
-------------------------------
Description:
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 and local verification
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.
was:
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.
Summary: Fluss pipeline source fails to recover from checkpoints
referencing dropped or expired partitions (was: Fluss pipeline source does not
unsubscribe dropped or expired partitions)
> 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. 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 and local verification
> 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.
--
This message was sent by Atlassian Jira
(v10.3.26#10030026)
