Mao Jiayi created FLINK-40513:
---------------------------------
Summary: Validate forward routing and optimize Fluss sink writer
with bucket affinity
Key: FLINK-40513
URL: https://issues.apache.org/jira/browse/FLINK-40513
Project: Flink
Issue Type: Improvement
Components: Flink CDC
Reporter: Mao Jiayi
h3. Background
FLINK-40331 / PR #4498 introduces an opt-in {{FORWARD}} routing strategy for
Fluss-to-Fluss synchronization. It preserves the upstream subtask distribution
and avoids repartitioning records before the Fluss sink.
However, FORWARD routing is beneficial and safe only when the source and target
tables have compatible bucket layouts. The Fluss sink currently does not
validate this assumption or explicitly expose bucket-affinity behavior.
h3. Proposed Changes
Enhance the Fluss sink when {{{}sink.partitioning.strategy=FORWARD{}}}:
# Validate that source and target tables have compatible distribution
settings, including bucket count, bucket keys, partition keys, table type and
bucketing algorithm.
# Validate at runtime that each record is routed to the expected target bucket
before writing.
# Fail fast when bucket affinity cannot be guaranteed, such as log tables
without deterministic bucket keys.
# Cache table metadata and bucket calculators in {{{}FlussSinkWriter{}}}.
# Reuse {{MultiTableWriter}} and its per-table/bucket batching so that each
sink subtask writes only its affinity bucket subset.
# Add metrics and logs for validated records, validation failures and active
buckets.
h3. Acceptance Criteria
* Compatible Fluss-to-Fluss FORWARD jobs write data correctly.
* Incompatible bucket layouts fail with a clear error instead of silently
degrading.
* Log tables without deterministic bucket keys are rejected in FORWARD mode.
* DEFAULT routing behavior remains unchanged.
* Unit and E2E tests cover successful and failed validation scenarios.
* The implementation is verified with both Flink 1.20 and Flink 2.x.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
