[
https://issues.apache.org/jira/browse/FLINK-40331?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
ASF GitHub Bot updated FLINK-40331:
-----------------------------------
Labels: pull-request-available (was: )
> [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
> Priority: Major
> Labels: pull-request-available
> 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)
