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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
