You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by "loserwang1024 (via GitHub)" <gi...@apache.org> on 2026/08/05 11:45:15 UTC
[PR] [FLINK-40331][cdc-connect][fluss] Support forward shuffle in fluss sink. [flink-cdc]
loserwang1024 opened a new pull request, #4498:
URL: https://github.com/apache/flink-cdc/pull/4498
<!--
Licensed to the Apache Software Foundation (ASF) under one
or more contributor license agreements. See the NOTICE file
distributed with this work for additional information
regarding copyright ownership. The ASF licenses this file
to you under the Apache License, Version 2.0 (the
"License"); you may not use this file except in compliance
with the License. You may obtain a copy of the License at
http://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing,
software distributed under the License is distributed on an
"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
KIND, either express or implied. See the License for the
specific language governing permissions and limitations
under the License.
-->
## What is the purpose of this pull request?
https://issues.apache.org/jira/browse/FLINK-40331
## Brief change log
Add sink.partitioning.strategy in fluss sink to support FORWARD.
---
## Verifying this change
<!-- Describe how this change can be verified. -->
This change is a trivial rework / code cleanup without any test coverage.
*(or)*
This change added tests and can be verified as follows:
- *Added/Updated unit tests in `...*`*
- *Added/Updated integration tests in `...*`*
- *Manually tested by ...*
## Documentation
- Does this pull request introduce a new feature? (yes / no)
- If yes, how is the feature documented? (not applicable / docs / JavaDocs / not documented)
---
##### Was generative AI tooling used to co-author this PR?
<!--
If generative AI tooling has been used in the process of authoring this PR, please
change the checkbox below to `[X]` followed by the name of the tool, and uncomment the
"Generated-by" line. See the ASF Generative Tooling Guidance for details:
https://www.apache.org/legal/generative-tooling.html
You are responsible for the quality and correctness of every change in this PR
regardless of the tooling used. Low-effort AI-generated PRs will be closed.
-->
- [ ] Yes (please specify the tool below)
<!--
Generated-by: [Tool Name and Version]
-->
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-40331][cdc-connect][fluss] Support forward shuffle in fluss sink. [flink-cdc]
Posted by "naivedogger (via GitHub)" <gi...@apache.org>.
naivedogger commented on code in PR #4498:
URL: https://github.com/apache/flink-cdc/pull/4498#discussion_r3892361201
##########
docs/content.zh/docs/connectors/pipeline-connectors/fluss.md:
##########
@@ -99,6 +99,13 @@ Pipeline Connector Options
<td>String</td>
<td>用于建立与 Fluss 集群初始连接的主机/端口对列表。 </td>
</tr>
+ <tr>
+ <td>sink.partitioning.strategy</td>
+ <td>optional</td>
+ <td style="word-wrap: break-word;">DEFAULT</td>
+ <td>String</td>
+ <td>DataChangeEvent 路由使用的分区策略。可选值为 <code>DEFAULT</code> 和 <code>FORWARD</code>。<code>DEFAULT</code> 对主键表按主键 hash，对 log 表采用 round-robin。<code>FORWARD</code> 将数据事件发送到与上游 subtask 索引相同的下游 subtask，用于 Fluss 到 Fluss 的数据同步。上游数据分布必须与 Fluss 表的分桶策略保持一致，否则可能出现数据正确性问题。</td>
Review Comment:
The docs and ConfigOption description say DEFAULT uses round-robin for log tables, but the actual implementation in FlussHashFunction uses ThreadLocalRandom.current().nextInt(), which is a random value, not round-robin. Could we align the docs here?
##########
docs/content/docs/connectors/pipeline-connectors/fluss.md:
##########
@@ -100,6 +100,13 @@ Pipeline Connector Options
<td>String</td>
<td>The bootstrap servers for the Fluss sink connection. </td>
</tr>
+ <tr>
+ <td>sink.partitioning.strategy</td>
+ <td>optional</td>
+ <td style="word-wrap: break-word;">DEFAULT</td>
+ <td>String</td>
+ <td>The partitioning strategy for DataChangeEvent routing. Available values are <code>DEFAULT</code> and <code>FORWARD</code>. <code>DEFAULT</code> hashes primary key tables by primary keys and routes log tables in round-robin mode. <code>FORWARD</code> routes data events to downstream subtasks with the same indices as upstream and is intended for Fluss-to-Fluss data synchronization. The upstream data distribution must match the Fluss table bucket distribution; otherwise, data correctness issues may occur.</td>
Review Comment:
ditto
##########
flink-cdc-common/src/main/java/org/apache/flink/cdc/common/sink/ForwardHashFunctionProvider.java:
##########
@@ -0,0 +1,52 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.cdc.common.sink;
+
+import org.apache.flink.cdc.common.annotation.Internal;
+import org.apache.flink.cdc.common.event.DataChangeEvent;
+import org.apache.flink.cdc.common.event.TableId;
+import org.apache.flink.cdc.common.function.HashFunction;
+import org.apache.flink.cdc.common.function.HashFunctionProvider;
+import org.apache.flink.cdc.common.schema.Schema;
+
+import javax.annotation.Nullable;
+
+/** A {@link HashFunctionProvider} that preserves the upstream subtask distribution. */
+@Internal
+public class ForwardHashFunctionProvider implements HashFunctionProvider<DataChangeEvent> {
+
+ private static final long serialVersionUID = 1L;
+
+ @Override
+ public HashFunction<DataChangeEvent> getHashFunction(@Nullable TableId tableId, Schema schema) {
+ return new ForwardHashFunction();
+ }
+
+ private static class ForwardHashFunction implements HashFunction<DataChangeEvent> {
+
+ @Override
+ public int hashcode(DataChangeEvent event) {
+ return 0;
Review Comment:
I see that ForwardHashFunction overrides both hashcode(int sourceIndex, DataChangeEvent event) and hashcode(DataChangeEvent event). The single-arg version returns 0.
My understanding is that hashcode(int sourceIndex, T event) is a default method on the interface, so the only reason to implement the single-arg version is that it's the abstract contract. But this raises two questions:
1. Is the single-arg override actually needed? If every call site in the pipeline now uses the two-arg overload, can we just leave the single-arg version unimplemented (i.e., rely on the default)? Or does something in the framework force us to provide it?
2. If we do need to implement it, shouldn't it throw rather than silently return 0? Returning 0 means that if someone accidentally regresses a call site back to hashcode(event), every record from every source subtask gets routed to sink subtask 0.
##########
flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-fluss/src/main/java/org/apache/flink/cdc/connectors/fluss/sink/FlussDataSinkOptions.java:
##########
@@ -33,6 +33,17 @@ public class FlussDataSinkOptions {
.noDefaultValue()
.withDescription("The bootstrap servers for the Fluss sink connection.");
+ public static final ConfigOption<SinkPartitioningStrategy> SINK_PARTITIONING_STRATEGY =
+ ConfigOptions.key("sink.partitioning.strategy")
+ .enumType(SinkPartitioningStrategy.class)
+ .defaultValue(SinkPartitioningStrategy.DEFAULT)
+ .withDescription(
+ "Partitioning strategy for DataChangeEvent routing. "
+ + "DEFAULT hashes primary key tables by primary keys and routes log tables in round-robin mode. "
Review Comment:
ditto
##########
flink-cdc-common/src/main/java/org/apache/flink/cdc/common/function/HashFunction.java:
##########
@@ -28,4 +28,14 @@
public interface HashFunction<T> {
int hashcode(T event);
+
+ /**
+ * Calculates the hash code with the upstream source subtask index.
+ *
+ * <p>Implementations that do not depend on the source subtask index can continue implementing
+ * {@link #hashcode(Object)} only.
+ */
+ default int hashcode(int sourceIndex, T event) {
Review Comment:
+1
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-40331][cdc-connect][fluss] Support forward shuffle in fluss sink. [flink-cdc]
Posted by "loserwang1024 (via GitHub)" <gi...@apache.org>.
loserwang1024 commented on code in PR #4498:
URL: https://github.com/apache/flink-cdc/pull/4498#discussion_r3892706277
##########
flink-cdc-common/src/main/java/org/apache/flink/cdc/common/function/HashFunction.java:
##########
@@ -28,4 +28,14 @@
public interface HashFunction<T> {
int hashcode(T event);
+
+ /**
+ * Calculates the hash code with the upstream source subtask index.
+ *
+ * <p>Implementations that do not depend on the source subtask index can continue implementing
+ * {@link #hashcode(Object)} only.
+ */
+ default int hashcode(int sourceIndex, T event) {
Review Comment:
done
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-40331][cdc-connect][fluss] Support forward shuffle in fluss sink. [flink-cdc]
Posted by "loserwang1024 (via GitHub)" <gi...@apache.org>.
loserwang1024 merged PR #4498:
URL: https://github.com/apache/flink-cdc/pull/4498
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-40331][cdc-connect][fluss] Support forward shuffle in fluss sink. [flink-cdc]
Posted by "lvyanquan (via GitHub)" <gi...@apache.org>.
lvyanquan commented on code in PR #4498:
URL: https://github.com/apache/flink-cdc/pull/4498#discussion_r3851108940
##########
flink-cdc-common/src/main/java/org/apache/flink/cdc/common/function/HashFunction.java:
##########
@@ -28,4 +28,14 @@
public interface HashFunction<T> {
int hashcode(T event);
+
+ /**
+ * Calculates the hash code with the upstream source subtask index.
+ *
+ * <p>Implementations that do not depend on the source subtask index can continue implementing
+ * {@link #hashcode(Object)} only.
+ */
+ default int hashcode(int sourceIndex, T event) {
Review Comment:
Hi @loserwang1024, thanks for the PR! One suggestion on the `HashFunction` change: instead of a raw `int sourceIndex`, consider passing a small immutable `HashContext` object:
```java
@Internal
public class HashContext {
private final int sourceSubtaskIndex;
private final int downstreamParallelism;
public HashContext(int sourceSubtaskIndex, int downstreamParallelism) { ... }
public int getSourceSubtaskIndex() { ... }
public int getDownstreamParallelism() { ... }
}
```
```java
default int hashcode(HashContext context, T event) {
return hashcode(event);
}
```
Reasons:
1. **Self-documenting call sites** — `hashcode(subTaskId, event)` gives no hint what the bare `int` means; `hashcode(hashContext, event)` does.
2. **Extensible without signature churn** — future strategies (bucket-aware mapping, fallback when `sourceIndex >= downstreamParallelism`) can read from the context instead of stacking more `int` overloads.
3. **Fails fast instead of silently skewing** — `ForwardHashFunction.hashcode(event)` can throw `UnsupportedOperationException` rather than returning `0` and routing everything to subtask 0 if someone calls the single-arg method by mistake.
The context can be created once in `open()` and reused, so there's no per-record allocation cost. Since `HashFunction` is `@Internal`, this is the cheapest time to make the change. WDYT?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-40331][cdc-connect][fluss] Support forward shuffle in fluss sink. [flink-cdc]
Posted by "naivedogger (via GitHub)" <gi...@apache.org>.
naivedogger commented on code in PR #4498:
URL: https://github.com/apache/flink-cdc/pull/4498#discussion_r3892456964
##########
flink-cdc-common/src/main/java/org/apache/flink/cdc/common/function/HashFunction.java:
##########
@@ -28,4 +28,14 @@
public interface HashFunction<T> {
int hashcode(T event);
+
+ /**
+ * Calculates the hash code with the upstream source subtask index.
+ *
+ * <p>Implementations that do not depend on the source subtask index can continue implementing
+ * {@link #hashcode(Object)} only.
+ */
+ default int hashcode(int sourceIndex, T event) {
Review Comment:
+1
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-40331][cdc-connect][fluss] Support forward shuffle in fluss sink. [flink-cdc]
Posted by "loserwang1024 (via GitHub)" <gi...@apache.org>.
loserwang1024 commented on code in PR #4498:
URL: https://github.com/apache/flink-cdc/pull/4498#discussion_r3892707355
##########
docs/content/docs/connectors/pipeline-connectors/fluss.md:
##########
@@ -100,6 +100,13 @@ Pipeline Connector Options
<td>String</td>
<td>The bootstrap servers for the Fluss sink connection. </td>
</tr>
+ <tr>
+ <td>sink.partitioning.strategy</td>
+ <td>optional</td>
+ <td style="word-wrap: break-word;">DEFAULT</td>
+ <td>String</td>
+ <td>The partitioning strategy for DataChangeEvent routing. Available values are <code>DEFAULT</code> and <code>FORWARD</code>. <code>DEFAULT</code> hashes primary key tables by primary keys and routes log tables in round-robin mode. <code>FORWARD</code> routes data events to downstream subtasks with the same indices as upstream and is intended for Fluss-to-Fluss data synchronization. The upstream data distribution must match the Fluss table bucket distribution; otherwise, data correctness issues may occur.</td>
Review Comment:
done
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org