This is an automated email from the ASF dual-hosted git repository.
loserwang1024 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink-cdc.git
The following commit(s) were added to refs/heads/master by this push:
new 9f23c03569 [FLINK-40331][cdc-connect][fluss] Support forward shuffle
in fluss sink. (#4498)
9f23c03569 is described below
commit 9f23c035694d305ccf267cf2866f39e8e9713fa5
Author: Hongshun Wang <[email protected]>
AuthorDate: Tue Sep 8 10:15:54 2026 +0800
[FLINK-40331][cdc-connect][fluss] Support forward shuffle in fluss sink.
(#4498)
---
.../docs/connectors/pipeline-connectors/fluss.md | 7 +++
.../docs/connectors/pipeline-connectors/fluss.md | 7 +++
.../cdc/common/function/DefaultHashContext.java | 52 ++++++++++++++++++++
.../flink/cdc/common/function/HashFunction.java | 23 +++++++++
.../common/sink/ForwardHashFunctionProvider.java | 54 ++++++++++++++++++++
.../flink/cdc/common/function/HashContextTest.java | 55 +++++++++++++++++++++
.../sink/ForwardHashFunctionProviderTest.java | 57 ++++++++++++++++++++++
.../fluss/factory/FlussDataSinkFactory.java | 9 +++-
.../cdc/connectors/fluss/sink/FlussDataSink.java | 16 +++++-
.../fluss/sink/FlussDataSinkOptions.java | 17 +++++++
.../fluss/factory/FlussDataSinkFactoryTest.java | 21 ++++++++
.../BatchRegularPrePartitionOperator.java | 8 ++-
.../DistributedPrePartitionOperator.java | 20 +++++---
.../partitioning/RegularPrePartitionOperator.java | 8 ++-
.../partitioning/PrePartitionOperatorTest.java | 38 ++++++++++++++-
15 files changed, 380 insertions(+), 12 deletions(-)
diff --git a/docs/content.zh/docs/connectors/pipeline-connectors/fluss.md
b/docs/content.zh/docs/connectors/pipeline-connectors/fluss.md
index 1ac06c19c1..7fda6154bb 100644
--- a/docs/content.zh/docs/connectors/pipeline-connectors/fluss.md
+++ b/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> 对主键表按主键进行哈希分区，对日志表采用随机分发，将事件分散到各下游
subtask 以实现负载均衡。<code>FORWARD</code> 将数据事件发送到与上游 subtask 索引相同的下游 subtask，用于
Fluss 到 Fluss 的数据同步。上游数据分布必须与 Fluss 表的分桶策略保持一致，否则可能出现数据正确性问题。</td>
+ </tr>
<tr>
<td>bucket.key</td>
<td>optional</td>
diff --git a/docs/content/docs/connectors/pipeline-connectors/fluss.md
b/docs/content/docs/connectors/pipeline-connectors/fluss.md
index a2d7c53249..7bf8813aa3 100644
--- a/docs/content/docs/connectors/pipeline-connectors/fluss.md
+++ b/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 randomly distributes log table
events across downstream subtasks for balanced load. <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 ta [...]
+ </tr>
<tr>
<td>bucket.key</td>
<td>optional</td>
diff --git
a/flink-cdc-common/src/main/java/org/apache/flink/cdc/common/function/DefaultHashContext.java
b/flink-cdc-common/src/main/java/org/apache/flink/cdc/common/function/DefaultHashContext.java
new file mode 100644
index 0000000000..8afb520a23
--- /dev/null
+++
b/flink-cdc-common/src/main/java/org/apache/flink/cdc/common/function/DefaultHashContext.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.function;
+
+import org.apache.flink.cdc.common.annotation.Internal;
+import org.apache.flink.cdc.common.utils.Preconditions;
+
+/** Default implementation of {@link HashFunction.HashContext}. */
+@Internal
+public final class DefaultHashContext implements HashFunction.HashContext {
+
+ private final int sourceSubtaskIndex;
+ private final int downstreamParallelism;
+
+ public DefaultHashContext(int sourceSubtaskIndex, int
downstreamParallelism) {
+ Preconditions.checkArgument(
+ sourceSubtaskIndex >= 0,
+ "sourceSubtaskIndex must be greater than or equal to 0, but
was %s.",
+ sourceSubtaskIndex);
+ Preconditions.checkArgument(
+ downstreamParallelism > 0,
+ "downstreamParallelism must be greater than 0, but was %s.",
+ downstreamParallelism);
+ this.sourceSubtaskIndex = sourceSubtaskIndex;
+ this.downstreamParallelism = downstreamParallelism;
+ }
+
+ @Override
+ public int getSourceSubtaskIndex() {
+ return sourceSubtaskIndex;
+ }
+
+ @Override
+ public int getDownstreamParallelism() {
+ return downstreamParallelism;
+ }
+}
diff --git
a/flink-cdc-common/src/main/java/org/apache/flink/cdc/common/function/HashFunction.java
b/flink-cdc-common/src/main/java/org/apache/flink/cdc/common/function/HashFunction.java
index 323a3de1af..a487ef684d 100644
---
a/flink-cdc-common/src/main/java/org/apache/flink/cdc/common/function/HashFunction.java
+++
b/flink-cdc-common/src/main/java/org/apache/flink/cdc/common/function/HashFunction.java
@@ -27,5 +27,28 @@ import org.apache.flink.cdc.common.annotation.Internal;
@Internal
public interface HashFunction<T> {
+ /** Runtime context for calculating an event hash code. */
+ @Internal
+ interface HashContext {
+ int getSourceSubtaskIndex();
+
+ int getDownstreamParallelism();
+ }
+
+ /** Creates a hash context. */
+ static HashContext createContext(int sourceSubtaskIndex, int
downstreamParallelism) {
+ return new DefaultHashContext(sourceSubtaskIndex,
downstreamParallelism);
+ }
+
int hashcode(T event);
+
+ /**
+ * Calculates the hash code with the runtime hash context.
+ *
+ * <p>Implementations that do not depend on the context can continue
implementing {@link
+ * #hashcode(Object)} only.
+ */
+ default int hashcode(HashContext context, T event) {
+ return hashcode(event);
+ }
}
diff --git
a/flink-cdc-common/src/main/java/org/apache/flink/cdc/common/sink/ForwardHashFunctionProvider.java
b/flink-cdc-common/src/main/java/org/apache/flink/cdc/common/sink/ForwardHashFunctionProvider.java
new file mode 100644
index 0000000000..52b70e3edb
--- /dev/null
+++
b/flink-cdc-common/src/main/java/org/apache/flink/cdc/common/sink/ForwardHashFunctionProvider.java
@@ -0,0 +1,54 @@
+/*
+ * 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.HashFunction.HashContext;
+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) {
+ throw new UnsupportedOperationException(
+ "Forward hash function requires a HashContext.");
+ }
+
+ @Override
+ public int hashcode(HashContext context, DataChangeEvent event) {
+ return context.getSourceSubtaskIndex();
+ }
+ }
+}
diff --git
a/flink-cdc-common/src/test/java/org/apache/flink/cdc/common/function/HashContextTest.java
b/flink-cdc-common/src/test/java/org/apache/flink/cdc/common/function/HashContextTest.java
new file mode 100644
index 0000000000..9c30358f52
--- /dev/null
+++
b/flink-cdc-common/src/test/java/org/apache/flink/cdc/common/function/HashContextTest.java
@@ -0,0 +1,55 @@
+/*
+ * 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.function;
+
+import org.apache.flink.cdc.common.function.HashFunction.HashContext;
+
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/** Unit tests for {@link HashContext}. */
+class HashContextTest {
+
+ @Test
+ void testProperties() {
+ HashContext context = HashFunction.createContext(3, 5);
+
+ assertThat(context).isInstanceOf(DefaultHashContext.class);
+ assertThat(context.getSourceSubtaskIndex()).isEqualTo(3);
+ assertThat(context.getDownstreamParallelism()).isEqualTo(5);
+ }
+
+ @Test
+ void testRejectsNegativeSourceSubtaskIndex() {
+ assertThatThrownBy(() -> HashFunction.createContext(-1, 1))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("sourceSubtaskIndex");
+ }
+
+ @ParameterizedTest
+ @ValueSource(ints = {0, -1})
+ void testRejectsNonPositiveDownstreamParallelism(int
downstreamParallelism) {
+ assertThatThrownBy(() -> HashFunction.createContext(0,
downstreamParallelism))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("downstreamParallelism");
+ }
+}
diff --git
a/flink-cdc-common/src/test/java/org/apache/flink/cdc/common/sink/ForwardHashFunctionProviderTest.java
b/flink-cdc-common/src/test/java/org/apache/flink/cdc/common/sink/ForwardHashFunctionProviderTest.java
new file mode 100644
index 0000000000..6750ba8fdf
--- /dev/null
+++
b/flink-cdc-common/src/test/java/org/apache/flink/cdc/common/sink/ForwardHashFunctionProviderTest.java
@@ -0,0 +1,57 @@
+/*
+ * 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.data.GenericRecordData;
+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.schema.Schema;
+
+import org.junit.jupiter.api.Test;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/** Unit tests for {@link ForwardHashFunctionProvider}. */
+class ForwardHashFunctionProviderTest {
+
+ @Test
+ void testForwardingBySourceIndex() {
+ DataChangeEvent event =
+ DataChangeEvent.insertEvent(TableId.tableId("customers"),
GenericRecordData.of(1));
+ HashFunction<DataChangeEvent> hashFunction =
+ new ForwardHashFunctionProvider()
+ .getHashFunction(event.tableId(),
Schema.newBuilder().build());
+
+ assertThat(hashFunction.hashcode(HashFunction.createContext(3, 5),
event)).isEqualTo(3);
+ }
+
+ @Test
+ void testHashingWithoutContextFails() {
+ DataChangeEvent event =
+ DataChangeEvent.insertEvent(TableId.tableId("customers"),
GenericRecordData.of(1));
+ HashFunction<DataChangeEvent> hashFunction =
+ new ForwardHashFunctionProvider()
+ .getHashFunction(event.tableId(),
Schema.newBuilder().build());
+
+ assertThatThrownBy(() -> hashFunction.hashcode(event))
+ .isInstanceOf(UnsupportedOperationException.class)
+ .hasMessageContaining("HashContext");
+ }
+}
diff --git
a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-fluss/src/main/java/org/apache/flink/cdc/connectors/fluss/factory/FlussDataSinkFactory.java
b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-fluss/src/main/java/org/apache/flink/cdc/connectors/fluss/factory/FlussDataSinkFactory.java
index ca3a6a98ef..455bbdc171 100644
---
a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-fluss/src/main/java/org/apache/flink/cdc/connectors/fluss/factory/FlussDataSinkFactory.java
+++
b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-fluss/src/main/java/org/apache/flink/cdc/connectors/fluss/factory/FlussDataSinkFactory.java
@@ -36,6 +36,7 @@ import static
org.apache.flink.cdc.connectors.fluss.sink.FlussDataSinkOptions.BO
import static
org.apache.flink.cdc.connectors.fluss.sink.FlussDataSinkOptions.BUCKET_KEY;
import static
org.apache.flink.cdc.connectors.fluss.sink.FlussDataSinkOptions.BUCKET_NUMBER;
import static
org.apache.flink.cdc.connectors.fluss.sink.FlussDataSinkOptions.CLIENT_PROPERTIES_PREFIX;
+import static
org.apache.flink.cdc.connectors.fluss.sink.FlussDataSinkOptions.SINK_PARTITIONING_STRATEGY;
import static
org.apache.flink.cdc.connectors.fluss.sink.FlussDataSinkOptions.TABLE_PROPERTIES_PREFIX;
import static
org.apache.flink.cdc.connectors.fluss.utils.FlussConfigUtils.parseBucketKeys;
import static
org.apache.flink.cdc.connectors.fluss.utils.FlussConfigUtils.parseBucketNumber;
@@ -57,7 +58,12 @@ public class FlussDataSinkFactory implements DataSinkFactory
{
parseBucketKeys(factoryConfiguration.get(BUCKET_KEY));
Map<String, Integer> bucketNumMap =
parseBucketNumber(factoryConfiguration.get(BUCKET_NUMBER));
- return new FlussDataSink(flussClientConfig, tableProperties,
bucketKeysMap, bucketNumMap);
+ return new FlussDataSink(
+ flussClientConfig,
+ tableProperties,
+ bucketKeysMap,
+ bucketNumMap,
+ factoryConfiguration.get(SINK_PARTITIONING_STRATEGY));
}
@Override
@@ -77,6 +83,7 @@ public class FlussDataSinkFactory implements DataSinkFactory {
Set<ConfigOption<?>> options = new HashSet<>();
options.add(BUCKET_KEY);
options.add(BUCKET_NUMBER);
+ options.add(SINK_PARTITIONING_STRATEGY);
return options;
}
diff --git
a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-fluss/src/main/java/org/apache/flink/cdc/connectors/fluss/sink/FlussDataSink.java
b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-fluss/src/main/java/org/apache/flink/cdc/connectors/fluss/sink/FlussDataSink.java
index 4491e50321..8af81355c8 100644
---
a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-fluss/src/main/java/org/apache/flink/cdc/connectors/fluss/sink/FlussDataSink.java
+++
b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-fluss/src/main/java/org/apache/flink/cdc/connectors/fluss/sink/FlussDataSink.java
@@ -22,7 +22,9 @@ import
org.apache.flink.cdc.common.function.HashFunctionProvider;
import org.apache.flink.cdc.common.sink.DataSink;
import org.apache.flink.cdc.common.sink.EventSinkProvider;
import org.apache.flink.cdc.common.sink.FlinkSinkProvider;
+import org.apache.flink.cdc.common.sink.ForwardHashFunctionProvider;
import org.apache.flink.cdc.common.sink.MetadataApplier;
+import
org.apache.flink.cdc.connectors.fluss.sink.FlussDataSinkOptions.SinkPartitioningStrategy;
import org.apache.flink.cdc.connectors.fluss.sink.v2.FlussSink;
import org.apache.fluss.config.Configuration;
@@ -37,16 +39,19 @@ public class FlussDataSink implements DataSink {
private final Map<String, String> tableProperties;
private final Map<String, List<String>> bucketKeysMap;
private final Map<String, Integer> bucketNumMap;
+ private final SinkPartitioningStrategy sinkPartitioningStrategy;
public FlussDataSink(
Configuration flussClientConfig,
Map<String, String> tableProperties,
Map<String, List<String>> bucketKeysMap,
- Map<String, Integer> bucketNumMap) {
+ Map<String, Integer> bucketNumMap,
+ SinkPartitioningStrategy sinkPartitioningStrategy) {
this.flussClientConfig = flussClientConfig;
this.tableProperties = tableProperties;
this.bucketKeysMap = bucketKeysMap;
this.bucketNumMap = bucketNumMap;
+ this.sinkPartitioningStrategy = sinkPartitioningStrategy;
}
@Override
@@ -63,6 +68,13 @@ public class FlussDataSink implements DataSink {
@Override
public HashFunctionProvider<DataChangeEvent>
getDataChangeEventHashFunctionProvider() {
- return new FlussHashFunctionProvider();
+ switch (sinkPartitioningStrategy) {
+ case DEFAULT:
+ return new FlussHashFunctionProvider();
+ case FORWARD:
+ return new ForwardHashFunctionProvider();
+ }
+ throw new IllegalArgumentException(
+ "Unsupported sink partitioning strategy: " +
sinkPartitioningStrategy);
}
}
diff --git
a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-fluss/src/main/java/org/apache/flink/cdc/connectors/fluss/sink/FlussDataSinkOptions.java
b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-fluss/src/main/java/org/apache/flink/cdc/connectors/fluss/sink/FlussDataSinkOptions.java
index f14fc1b67f..51287fca46 100644
---
a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-fluss/src/main/java/org/apache/flink/cdc/connectors/fluss/sink/FlussDataSinkOptions.java
+++
b/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 randomly distributes log table events across downstream
subtasks for balanced load. "
+ + "FORWARD routes data events to
downstream subtasks with the same indices as upstream. "
+ + "FORWARD 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.");
+
public static final ConfigOption<String> BUCKET_KEY =
ConfigOptions.key("bucket.key")
.stringType()
@@ -54,4 +65,10 @@ public class FlussDataSinkOptions {
"The number of buckets of each Fluss table."
+ "Tables are separated by ';'. "
+ "Format:
database1.table1:4;database1.table2:8.");
+
+ /** The shuffle strategy for fluss sink. */
+ public enum SinkPartitioningStrategy {
+ DEFAULT,
+ FORWARD
+ }
}
diff --git
a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-fluss/src/test/java/org/apache/flink/cdc/connectors/fluss/factory/FlussDataSinkFactoryTest.java
b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-fluss/src/test/java/org/apache/flink/cdc/connectors/fluss/factory/FlussDataSinkFactoryTest.java
index 49ec01d300..7b287d122c 100644
---
a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-fluss/src/test/java/org/apache/flink/cdc/connectors/fluss/factory/FlussDataSinkFactoryTest.java
+++
b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-fluss/src/test/java/org/apache/flink/cdc/connectors/fluss/factory/FlussDataSinkFactoryTest.java
@@ -21,9 +21,12 @@ import
org.apache.flink.cdc.common.configuration.Configuration;
import org.apache.flink.cdc.common.factories.DataSinkFactory;
import org.apache.flink.cdc.common.factories.FactoryHelper;
import org.apache.flink.cdc.common.sink.DataSink;
+import org.apache.flink.cdc.common.sink.ForwardHashFunctionProvider;
import org.apache.flink.cdc.composer.utils.FactoryDiscoveryUtils;
import org.apache.flink.cdc.connectors.fluss.sink.FlussDataSink;
import org.apache.flink.cdc.connectors.fluss.sink.FlussDataSinkOptions;
+import
org.apache.flink.cdc.connectors.fluss.sink.FlussDataSinkOptions.SinkPartitioningStrategy;
+import org.apache.flink.cdc.connectors.fluss.sink.FlussHashFunctionProvider;
import org.apache.flink.table.api.ValidationException;
import org.apache.flink.shaded.guava31.com.google.common.collect.ImmutableMap;
@@ -47,6 +50,24 @@ public class FlussDataSinkFactoryTest {
new FactoryHelper.DefaultContext(
conf, conf,
Thread.currentThread().getContextClassLoader()));
Assertions.assertThat(dataSink).isInstanceOf(FlussDataSink.class);
+
Assertions.assertThat(dataSink.getDataChangeEventHashFunctionProvider())
+ .isInstanceOf(FlussHashFunctionProvider.class);
+ }
+
+ @Test
+ void testCreateDataSinkWithForwardPartitioningStrategy() {
+ DataSinkFactory sinkFactory =
+ FactoryDiscoveryUtils.getFactoryByIdentifier("fluss",
DataSinkFactory.class);
+ Configuration conf = createValidConfiguration();
+ conf.set(FlussDataSinkOptions.SINK_PARTITIONING_STRATEGY,
SinkPartitioningStrategy.FORWARD);
+
+ DataSink dataSink =
+ sinkFactory.createDataSink(
+ new FactoryHelper.DefaultContext(
+ conf, conf,
Thread.currentThread().getContextClassLoader()));
+
+
Assertions.assertThat(dataSink.getDataChangeEventHashFunctionProvider())
+ .isInstanceOf(ForwardHashFunctionProvider.class);
}
@Test
diff --git
a/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/partitioning/BatchRegularPrePartitionOperator.java
b/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/partitioning/BatchRegularPrePartitionOperator.java
index 1a2a697986..f5314fc676 100644
---
a/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/partitioning/BatchRegularPrePartitionOperator.java
+++
b/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/partitioning/BatchRegularPrePartitionOperator.java
@@ -23,6 +23,7 @@ import org.apache.flink.cdc.common.event.DataChangeEvent;
import org.apache.flink.cdc.common.event.Event;
import org.apache.flink.cdc.common.event.TableId;
import org.apache.flink.cdc.common.function.HashFunction;
+import org.apache.flink.cdc.common.function.HashFunction.HashContext;
import org.apache.flink.cdc.common.function.HashFunctionProvider;
import org.apache.flink.cdc.common.schema.Schema;
import org.apache.flink.cdc.runtime.operators.AbstractStreamOperatorAdapter;
@@ -56,6 +57,7 @@ public class BatchRegularPrePartitionOperator
private transient Map<TableId, HashFunction<DataChangeEvent>>
cachedHashFunctions;
private transient volatile Map<TableId, Schema> originalSchemaMap;
+ private transient HashContext hashContext;
public BatchRegularPrePartitionOperator(
int downstreamParallelism, HashFunctionProvider<DataChangeEvent>
hashFunctionProvider) {
@@ -69,6 +71,10 @@ public class BatchRegularPrePartitionOperator
super.open();
cachedHashFunctions = new HashMap<>();
originalSchemaMap = new HashMap<>();
+ hashContext =
+ HashFunction.createContext(
+
getRuntimeContext().getTaskInfo().getIndexOfThisSubtask(),
+ downstreamParallelism);
}
@Override
@@ -95,7 +101,7 @@ public class BatchRegularPrePartitionOperator
dataChangeEvent,
cachedHashFunctions
.get(dataChangeEvent.tableId())
- .hashcode(dataChangeEvent)
+ .hashcode(hashContext,
dataChangeEvent)
% downstreamParallelism)));
}
diff --git
a/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/partitioning/DistributedPrePartitionOperator.java
b/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/partitioning/DistributedPrePartitionOperator.java
index 9231abf1a8..7a984088b1 100644
---
a/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/partitioning/DistributedPrePartitionOperator.java
+++
b/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/partitioning/DistributedPrePartitionOperator.java
@@ -23,6 +23,7 @@ import org.apache.flink.cdc.common.event.Event;
import org.apache.flink.cdc.common.event.SchemaChangeEvent;
import org.apache.flink.cdc.common.event.TableId;
import org.apache.flink.cdc.common.function.HashFunction;
+import org.apache.flink.cdc.common.function.HashFunction.HashContext;
import org.apache.flink.cdc.common.function.HashFunctionProvider;
import org.apache.flink.cdc.common.schema.Schema;
import org.apache.flink.cdc.common.utils.SchemaUtils;
@@ -53,7 +54,7 @@ public class DistributedPrePartitionOperator
private transient Map<TableId, Schema> schemaMap;
private transient Map<TableId, HashFunction<DataChangeEvent>>
hashFunctionMap;
- private transient int subTaskId;
+ private transient HashContext hashContext;
public DistributedPrePartitionOperator(
int downstreamParallelism, HashFunctionProvider<DataChangeEvent>
hashFunctionProvider) {
@@ -65,7 +66,10 @@ public class DistributedPrePartitionOperator
@Override
public void open() throws Exception {
super.open();
- subTaskId =
RuntimeContextAdapter.getIndexOfThisSubtask(getRuntimeContext());
+ hashContext =
+ HashFunction.createContext(
+
RuntimeContextAdapter.getIndexOfThisSubtask(getRuntimeContext()),
+ downstreamParallelism);
schemaMap = new HashMap<>();
hashFunctionMap = new HashMap<>();
}
@@ -93,7 +97,9 @@ public class DistributedPrePartitionOperator
partitionBy((DataChangeEvent) event);
} else {
throw new IllegalStateException(
- subTaskId + "> PrePartition operator received an
unexpected event: " + event);
+ hashContext.getSourceSubtaskIndex()
+ + "> PrePartition operator received an unexpected
event: "
+ + event);
}
}
@@ -102,10 +108,10 @@ public class DistributedPrePartitionOperator
new StreamRecord<>(
PartitioningEvent.ofDistributed(
dataChangeEvent,
- subTaskId,
+ hashContext.getSourceSubtaskIndex(),
hashFunctionMap
.get(dataChangeEvent.tableId())
- .hashcode(dataChangeEvent)
+ .hashcode(hashContext,
dataChangeEvent)
% downstreamParallelism)));
}
@@ -115,7 +121,9 @@ public class DistributedPrePartitionOperator
// JVM
Event copiedEvent = EventSerializer.INSTANCE.copy(toBroadcast);
output.collect(
- new
StreamRecord<>(PartitioningEvent.ofDistributed(copiedEvent, subTaskId, i)));
+ new StreamRecord<>(
+ PartitioningEvent.ofDistributed(
+ copiedEvent,
hashContext.getSourceSubtaskIndex(), i)));
}
}
diff --git
a/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/partitioning/RegularPrePartitionOperator.java
b/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/partitioning/RegularPrePartitionOperator.java
index a5d81bfb19..6b72396b4d 100644
---
a/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/partitioning/RegularPrePartitionOperator.java
+++
b/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/partitioning/RegularPrePartitionOperator.java
@@ -24,6 +24,7 @@ import org.apache.flink.cdc.common.event.FlushEvent;
import org.apache.flink.cdc.common.event.SchemaChangeEvent;
import org.apache.flink.cdc.common.event.TableId;
import org.apache.flink.cdc.common.function.HashFunction;
+import org.apache.flink.cdc.common.function.HashFunction.HashContext;
import org.apache.flink.cdc.common.function.HashFunctionProvider;
import org.apache.flink.cdc.common.schema.Schema;
import org.apache.flink.cdc.runtime.operators.AbstractStreamOperatorAdapter;
@@ -62,6 +63,7 @@ public class RegularPrePartitionOperator extends
AbstractStreamOperatorAdapter<P
private transient SchemaEvolutionClient schemaEvolutionClient;
private transient LoadingCache<TableId, HashFunction<DataChangeEvent>>
cachedHashFunctions;
+ private transient HashContext hashContext;
public RegularPrePartitionOperator(
OperatorID schemaOperatorId,
@@ -80,6 +82,10 @@ public class RegularPrePartitionOperator extends
AbstractStreamOperatorAdapter<P
getContainingTask().getEnvironment().getOperatorCoordinatorEventGateway();
schemaEvolutionClient = new SchemaEvolutionClient(toCoordinator,
schemaOperatorId);
cachedHashFunctions = createCache();
+ hashContext =
+ HashFunction.createContext(
+
getRuntimeContext().getTaskInfo().getIndexOfThisSubtask(),
+ downstreamParallelism);
}
@Override
@@ -107,7 +113,7 @@ public class RegularPrePartitionOperator extends
AbstractStreamOperatorAdapter<P
dataChangeEvent,
cachedHashFunctions
.get(dataChangeEvent.tableId())
- .hashcode(dataChangeEvent)
+ .hashcode(hashContext,
dataChangeEvent)
% downstreamParallelism)));
}
diff --git
a/flink-cdc-runtime/src/test/java/org/apache/flink/cdc/runtime/partitioning/PrePartitionOperatorTest.java
b/flink-cdc-runtime/src/test/java/org/apache/flink/cdc/runtime/partitioning/PrePartitionOperatorTest.java
index 4273f97cdc..692e3835b1 100644
---
a/flink-cdc-runtime/src/test/java/org/apache/flink/cdc/runtime/partitioning/PrePartitionOperatorTest.java
+++
b/flink-cdc-runtime/src/test/java/org/apache/flink/cdc/runtime/partitioning/PrePartitionOperatorTest.java
@@ -25,6 +25,7 @@ import
org.apache.flink.cdc.common.event.SchemaChangeEventType;
import org.apache.flink.cdc.common.event.TableId;
import org.apache.flink.cdc.common.schema.Schema;
import
org.apache.flink.cdc.common.sink.DefaultDataChangeEventHashFunctionProvider;
+import org.apache.flink.cdc.common.sink.ForwardHashFunctionProvider;
import org.apache.flink.cdc.common.sink.TableIdHashFunctionProvider;
import org.apache.flink.cdc.common.types.DataTypes;
import org.apache.flink.cdc.common.types.RowType;
@@ -39,7 +40,7 @@ import java.util.Collections;
import static org.assertj.core.api.Assertions.assertThat;
-/** Unit test for {@link RegularPrePartitionOperator}. */
+/** Unit tests for pre-partition operators. */
class PrePartitionOperatorTest {
private static final TableId CUSTOMERS =
TableId.tableId("my_company", "my_branch", "customers");
@@ -140,6 +141,31 @@ class PrePartitionOperatorTest {
}
}
+ @Test
+ void testForwardingDataChangeEvent() throws Exception {
+ try (RegularEventOperatorTestHarness<RegularPrePartitionOperator,
PartitioningEvent>
+ testHarness = createDataForwardTestHarness()) {
+ testHarness.open();
+ testHarness.registerTableSchema(CUSTOMERS, CUSTOMERS_SCHEMA);
+
+ DataChangeEvent dataChangeEvent = createDataChangeEvent();
+ testHarness.getOperator().processElement(new
StreamRecord<>(dataChangeEvent));
+
+ assertThat(testHarness.getOutputRecords())
+ .containsExactly(
+ new
StreamRecord<>(PartitioningEvent.ofRegular(dataChangeEvent, 0)));
+ }
+ }
+
+ private DataChangeEvent createDataChangeEvent() {
+ BinaryRecordDataGenerator recordDataGenerator =
+ new BinaryRecordDataGenerator(((RowType)
CUSTOMERS_SCHEMA.toRowDataType()));
+ return DataChangeEvent.insertEvent(
+ CUSTOMERS,
+ recordDataGenerator.generate(
+ new Object[] {1, new BinaryStringData("Alice"),
12345678L}));
+ }
+
private int getPartitioningTarget(Schema schema, DataChangeEvent
dataChangeEvent) {
return new DefaultDataChangeEventHashFunctionProvider()
.getHashFunction(null, schema)
@@ -228,4 +254,14 @@ class PrePartitionOperatorTest {
new DefaultDataChangeEventHashFunctionProvider());
return RegularEventOperatorTestHarness.with(operator,
DOWNSTREAM_PARALLELISM);
}
+
+ private RegularEventOperatorTestHarness<RegularPrePartitionOperator,
PartitioningEvent>
+ createDataForwardTestHarness() {
+ RegularPrePartitionOperator operator =
+ new RegularPrePartitionOperator(
+ TestingSchemaRegistryGateway.SCHEMA_OPERATOR_ID,
+ DOWNSTREAM_PARALLELISM,
+ new ForwardHashFunctionProvider());
+ return RegularEventOperatorTestHarness.with(operator,
DOWNSTREAM_PARALLELISM);
+ }
}
