[
https://issues.apache.org/jira/browse/FLINK-41312?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
liting liu closed FLINK-41312.
------------------------------
Resolution: Won't Do
Closing as Won't Do after design review. Flink CDC supports heterogeneous
sources and many-to-one routing; source Fluss tables can have different bucket
keys, so there is no unique layout to inherit for a shared sink table. Target
bucket keys should continue to be determined by explicit sink bucket.key or the
existing defaults. The local automatic-inheritance draft has been reverted;
this proposal will not be pursued.
> Preserve source bucket keys when auto-creating Fluss-to-Fluss sink tables
> -------------------------------------------------------------------------
>
> Key: FLINK-41312
> URL: https://issues.apache.org/jira/browse/FLINK-41312
> Project: Flink
> Issue Type: Improvement
> Components: Flink CDC
> Reporter: liting liu
> Priority: Major
>
> 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.
--
This message was sent by Atlassian Jira
(v10.3.26#10030026)
