Hongshun Wang created FLINK-40331:
-------------------------------------
Summary: [FLINK CDC][fluss] Add an opt-in forward routing mode for
fluss sink
Key: FLINK-40331
URL: https://issues.apache.org/jira/browse/FLINK-40331
Project: Flink
Issue Type: Bug
Components: Flink CDC
Affects Versions: cdc-3.5.0
Reporter: Hongshun Wang
Fix For: cdc-3.6.0
h3. Description
When synchronizing data from Fluss source to Fluss sink, the upstream and
downstream tables may use the same bucket layout, but the current pipeline
routing still repartitions data before the sink.
For Fluss source, splits are generated per {{TableBucket}} and assigned to
source readers based on the bucket identity. However, after the source emits
CDC events, the original bucket identity is not preserved in the event. Before
writing to the sink, the pipeline currently routes records through the generic
sink partitioning logic. For Fluss sink, this means:
* Primary-key tables are routed by hash.
* Log tables without primary keys are distributed with a random component to
avoid sending all records to one subtask.
As a result, data read from one upstream bucket can be scattered across
multiple sink subtasks. In a mirror synchronization scenario, each sink writer
may need to write data for all subscribed tables and buckets, making each
writer-side batch much smaller and less efficient.
For example, assume a Fluss-to-Fluss mirror job subscribes to 50 tables, each
with 16 buckets, and the pipeline parallelism is 8. Even when the upstream and
downstream bucket layouts are identical, the current repartitioning makes
records from each source bucket fan out to sink writers instead of keeping a
one-to-one source-reader-to-sink-writer path. This reduces batch aggregation
efficiency on the sink side.
h3. Proposal
Introduce an explicit opt-in configuration sink to force forward routing .
In the Fluss source, each split corresponds to a Fluss {{{}TableBucket{}}}, and
split assignment is based on {{{}TableBucket{}}}. If the pipeline uses forward
routing and the source/sink parallelism is identical, one source subtask can
forward its bucket data directly to the corresponding sink subtask.
This avoids scattering records from the same upstream bucket to all sink
writers and can improve sink-side batch size and write efficiency.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
