[jira] [Created] (FLINK-40331) [FLINK CDC][fluss] Add an opt-in forward routing mode for fluss sink

["Hongshun Wang (Jira)" <ji...@apache.org>]

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)