[
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 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.
was:
h2. Motivation
For Fluss-to-Fluss replication, a missing sink table is created automatically.
However, the source table's bucket keys are not carried in the CDC schema.
Unless the user repeats them in the sink's bucket.key option, the new target
can have a different bucket layout.
The example below uses an ordinary non-partitioned table.
h2. Minimal reproduction
# Create a non-partitioned Fluss source table cdc_repro.orders with the
following definition:
{code}
Columns: id INT NOT NULL, tenant INT NOT NULL, marker STRING
Primary key: (id, tenant)
Bucket keys: (tenant)
Bucket count: 2
{code}
# Leave cdc_repro.orders absent in a second Fluss cluster.
# Start a Fluss-to-Fluss CDC pipeline for that table, without a sink bucket.key
override.
# Inspect the bucket keys of the automatically created target table.
The current code derives target bucket keys (id, tenant), instead of preserving
the source bucket keys (tenant). The columns and primary key can therefore
match while the bucket layout differs.
h2. Current implementation
Reviewed Apache Flink CDC master at db45d06a1e35bc6be87f7b58913c768f8f299d3e:
* FlussMetadataAccessor.getTableSchema() passes only the Fluss schema and
partition keys to FlussConversions.toCdcSchema(); source bucket keys are not
propagated.
* FlussMetaDataApplier.applyCreateTable() gets bucket keys from the sink
configuration.
* FlussConversions.toFlussTable() defaults unspecified bucket keys to primary
keys minus partition keys.
Source references:
*
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/FlussMetadataAccessor.java#L98-L104
*
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/utils/FlussConversions.java#L169-L186
h2. Expected behavior / proposal
When automatically creating a Fluss sink table from a Fluss source, preserve
the source bucket keys if no explicit sink bucket.key override is provided.
* Keep explicit sink overrides authoritative.
* Keep the current default for sources that do not provide Fluss bucket
metadata.
* Preserve the existing target table's layout when the target already exists.
The current documentation describes the existing default, so this issue
proposes improving the Fluss-to-Fluss auto-creation behavior rather than
claiming the implementation violates that documented default.
h2. Related work and workaround
FLINK-40850 / PR #4564 fixes compatibility checks for pre-created primary-key
tables with custom bucket keys. It explicitly leaves new-table creation
unchanged, so it does not address this case.
The current workaround is to repeat the source bucket keys in the sink
configuration:
{code:yaml}
sink:
type: fluss
bootstrap.servers: <target-bootstrap>
bucket.key: cdc_repro.orders:tenant
{code}
h2. Validation scope
The metadata propagation and defaulting behavior above were confirmed by
source-code inspection. A new upstream end-to-end reproduction has not been run
for this report.
Issue Type: Bug (was: Improvement)
Summary: Fluss pipeline source does not retire dropped or expired
partitions while running (was: Preserve source bucket keys when auto-creating
Fluss-to-Fluss sink tables)
> 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
>
> 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.
--
This message was sent by Atlassian Jira
(v10.3.26#10030026)
