You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by le...@apache.org on 2025/07/07 11:48:51 UTC
(flink-cdc) branch master updated: [FLINK-37958][pipeline-connector/fluss] Introduce Fluss Pipeline Sink connector
This is an automated email from the ASF dual-hosted git repository.
leonard 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 2139824f6 [FLINK-37958][pipeline-connector/fluss] Introduce Fluss Pipeline Sink connector
2139824f6 is described below
commit 2139824f6d554f083f7608797fff8ad771647da5
Author: Hongshun Wang <12...@users.noreply.github.com>
AuthorDate: Mon Jul 7 19:48:46 2025 +0800
[FLINK-37958][pipeline-connector/fluss] Introduce Fluss Pipeline Sink connector
This closes #4053
---
.github/workflows/flink_cdc_base.yml | 7 +
.github/workflows/flink_cdc_ci.yml | 2 +-
.github/workflows/flink_cdc_ci_nightly.yml | 2 +-
.../flink-cdc-pipeline-connector-fluss/pom.xml | 136 +++++
.../fluss/factory/FlussDataSinkFactory.java | 109 ++++
.../cdc/connectors/fluss/sink/CdcAsFlussRow.java | 179 +++++++
.../connectors/fluss/sink/FlussConfigUtils.java | 81 +++
.../cdc/connectors/fluss/sink/FlussDataSink.java | 68 +++
.../fluss/sink/FlussDataSinkOptions.java | 57 ++
.../fluss/sink/FlussEventSerializationSchema.java | 184 +++++++
.../fluss/sink/FlussHashFunctionProvider.java | 101 ++++
.../fluss/sink/FlussMetaDataApplier.java | 181 +++++++
.../cdc/connectors/fluss/sink/v2/FlussEvent.java | 49 ++
.../fluss/sink/v2/FlussEventSerializer.java | 35 ++
.../fluss/sink/v2/FlussOperationType.java | 52 ++
.../connectors/fluss/sink/v2/FlussRowWithOp.java | 104 ++++
.../cdc/connectors/fluss/sink/v2/FlussSink.java | 62 +++
.../connectors/fluss/sink/v2/FlussSinkWriter.java | 258 +++++++++
.../fluss/sink/v2/metrics/WarppedFlussCounter.java | 59 +++
.../fluss/sink/v2/metrics/WrappedFlussGauge.java | 39 ++
.../sink/v2/metrics/WrapperFlussHistogram.java | 98 ++++
.../fluss/sink/v2/metrics/WrapperFlussMeter.java | 54 ++
.../v2/metrics/WrapperFlussMetricRegistry.java | 151 ++++++
.../connectors/fluss/utils/FlussConversions.java | 255 +++++++++
.../org.apache.flink.cdc.common.factories.Factory | 16 +
.../cdc/connectors/fluss/FlussPipelineITCase.java | 576 +++++++++++++++++++++
.../fluss/factory/FlussDataSinkFactoryTest.java | 154 ++++++
.../sink/FlussEventSerializationSchemaTest.java | 265 ++++++++++
.../fluss/sink/FlussMetadataApplierTest.java | 561 ++++++++++++++++++++
.../connectors/fluss/sink/v2/FlussSinkITCase.java | 576 +++++++++++++++++++++
.../src/test/resources/log4j2-test.properties | 25 +
.../flink-cdc-pipeline-connectors/pom.xml | 1 +
32 files changed, 4495 insertions(+), 2 deletions(-)
diff --git a/.github/workflows/flink_cdc_base.yml b/.github/workflows/flink_cdc_base.yml
index 1b0cdabb2..e408c26e7 100644
--- a/.github/workflows/flink_cdc_base.yml
+++ b/.github/workflows/flink_cdc_base.yml
@@ -61,6 +61,7 @@ env:
flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-oceanbase,\
flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-paimon,\
flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-starrocks,\
+ flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-fluss,\
flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-values"
MODULES_MYSQL_SOURCE: "\
@@ -126,6 +127,9 @@ env:
MODULES_MAXCOMPUTE: "\
flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-maxcompute"
+ MODULES_FLUSS: "\
+ flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-fluss"
+
MODULES_PIPELINE_E2E: "\
flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests"
@@ -247,6 +251,9 @@ jobs:
("starrocks")
modules=${{ env.MODULES_STARROCKS }}
;;
+ ("fluss")
+ modules=${{ env.MODULES_FLUSS }}
+ ;;
("pipeline_e2e")
compile_modules="${{ env.MODULES_CORE }},${{ env.MODULES_PIPELINE_CONNECTORS }},${{ env.MODULES_MYSQL_SOURCE }},${{ env.MODULES_POSTGRES }},${{ env.MODULES_ORACLE }},${{ env.MODULES_MONGODB }},${{ env.MODULES_SQLSERVER }},${{ env.MODULES_TIDB }},${{ env.MODULES_OCEANBASE_SOURCE }},${{ env.MODULES_DB2 }},${{ env.MODULES_VITESS }},${{ env.MODULES_PIPELINE_E2E }}"
modules=${{ env.MODULES_PIPELINE_E2E }}
diff --git a/.github/workflows/flink_cdc_ci.yml b/.github/workflows/flink_cdc_ci.yml
index d4a7fefcc..70d91d8ea 100644
--- a/.github/workflows/flink_cdc_ci.yml
+++ b/.github/workflows/flink_cdc_ci.yml
@@ -68,7 +68,7 @@ jobs:
uses: ./.github/workflows/flink_cdc_base.yml
with:
java-versions: "[8]"
- modules: "['mysql-pipeline', 'oceanbase-pipeline', 'doris', 'elasticsearch', 'iceberg', 'kafka', 'maxcompute', 'paimon', 'starrocks']"
+ modules: "['mysql-pipeline', 'oceanbase-pipeline', 'doris', 'elasticsearch', 'iceberg', 'kafka', 'maxcompute', 'paimon', 'starrocks', 'fluss']"
source-ut:
name: Source Unit Tests
uses: ./.github/workflows/flink_cdc_base.yml
diff --git a/.github/workflows/flink_cdc_ci_nightly.yml b/.github/workflows/flink_cdc_ci_nightly.yml
index c17058413..9d12e69ef 100644
--- a/.github/workflows/flink_cdc_ci_nightly.yml
+++ b/.github/workflows/flink_cdc_ci_nightly.yml
@@ -60,7 +60,7 @@ jobs:
uses: ./.github/workflows/flink_cdc_base.yml
with:
java-versions: "[11]"
- modules: "['mysql-pipeline', 'oceanbase-pipeline', 'doris', 'elasticsearch', 'iceberg', 'kafka', 'maxcompute', 'paimon', 'starrocks']"
+ modules: "['mysql-pipeline', 'oceanbase-pipeline', 'doris', 'elasticsearch', 'iceberg', 'kafka', 'maxcompute', 'paimon', 'starrocks', 'fluss']"
source-ut:
if: github.repository == 'apache/flink-cdc'
name: Source Unit Tests
diff --git a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-fluss/pom.xml b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-fluss/pom.xml
new file mode 100644
index 000000000..0ebfbc9d2
--- /dev/null
+++ b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-fluss/pom.xml
@@ -0,0 +1,136 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+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.
+-->
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+ xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+ xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+ <modelVersion>4.0.0</modelVersion>
+ <parent>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-cdc-pipeline-connectors</artifactId>
+ <version>${revision}</version>
+ </parent>
+
+ <artifactId>flink-cdc-pipeline-connector-fluss</artifactId>
+ <packaging>jar</packaging>
+
+ <name>flink-cdc-pipeline-connector-fluss</name>
+ <url>http://maven.apache.org</url>
+
+
+ <properties>
+ <fluss.version>0.7.0</fluss.version>
+ </properties>
+
+ <dependencies>
+ <dependency>
+ <groupId>com.alibaba.fluss</groupId>
+ <artifactId>fluss-client</artifactId>
+ <version>${fluss.version}</version>
+ </dependency>
+
+
+ <!-- Test dependencies -->
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-cdc-composer</artifactId>
+ <version>${project.version}</version>
+ <scope>test</scope>
+ </dependency>
+
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-test-utils</artifactId>
+ <version>${flink.version}</version>
+ <scope>test</scope>
+ </dependency>
+
+ <dependency>
+ <groupId>com.alibaba.fluss</groupId>
+ <artifactId>fluss-server</artifactId>
+ <version>${fluss.version}</version>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>com.alibaba.fluss</groupId>
+ <artifactId>fluss-server</artifactId>
+ <version>${fluss.version}</version>
+ <type>test-jar</type>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>com.alibaba.fluss</groupId>
+ <artifactId>fluss-test-utils</artifactId>
+ <version>${fluss.version}</version>
+ <scope>test</scope>
+ </dependency>
+ <!-- In Flink CDC project has Pipeline Sink Connector for Fluss. we import fluss-fink for Fluss Sink Connector just for test purpose -->
+ <dependency>
+ <groupId>com.alibaba.fluss</groupId>
+ <artifactId>fluss-flink-common</artifactId>
+ <version>${fluss.version}</version>
+ <type>test-jar</type>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>com.alibaba.fluss</groupId>
+ <artifactId>fluss-flink-1.20</artifactId>
+ <version>${fluss.version}</version>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-table-planner_2.12</artifactId>
+ <version>${flink.version}</version>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-cdc-pipeline-connector-values</artifactId>
+ <version>${project.version}</version>
+ <scope>test</scope>
+ </dependency>
+ </dependencies>
+
+ <build>
+ <plugins>
+ <plugin>
+ <groupId>org.apache.maven.plugins</groupId>
+ <artifactId>maven-shade-plugin</artifactId>
+ <version>${maven.shade.plugin.version}</version>
+ <executions>
+ <execution>
+ <id>shade-flink</id>
+ <phase>package</phase>
+ <goals>
+ <goal>shade</goal>
+ </goals>
+ <configuration>
+ <shadeTestJar>false</shadeTestJar>
+ <artifactSet>
+ <includes>
+ <include>com.alibaba.fluss:*</include>
+ </includes>
+ </artifactSet>
+ </configuration>
+ </execution>
+ </executions>
+ </plugin>
+ </plugins>
+ </build>
+
+</project>
\ No newline at end of file
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
new file mode 100644
index 000000000..754eb693a
--- /dev/null
+++ 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
@@ -0,0 +1,109 @@
+/*
+ * 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.connectors.fluss.factory;
+
+import org.apache.flink.cdc.common.configuration.ConfigOption;
+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.connectors.fluss.sink.FlussDataSink;
+
+import com.alibaba.fluss.config.ConfigOptions;
+import com.alibaba.fluss.config.Configuration;
+
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+
+import static org.apache.flink.cdc.connectors.fluss.sink.FlussConfigUtils.parseBucketKeys;
+import static org.apache.flink.cdc.connectors.fluss.sink.FlussConfigUtils.parseBucketNumber;
+import static org.apache.flink.cdc.connectors.fluss.sink.FlussDataSinkOptions.BOOTSTRAP_SERVERS;
+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.TABLE_PROPERTIES_PREFIX;
+
+/** Factory for creating configured instances of {@link FlussDataSink}. */
+public class FlussDataSinkFactory implements DataSinkFactory {
+ public static final String IDENTIFIER = "fluss";
+
+ @Override
+ public DataSink createDataSink(Context context) {
+ FactoryHelper.createFactoryHelper(this, context)
+ .validateExcept(CLIENT_PROPERTIES_PREFIX, TABLE_PROPERTIES_PREFIX);
+ org.apache.flink.cdc.common.configuration.Configuration factoryConfiguration =
+ context.getFactoryConfiguration();
+
+ Configuration flussClientConfig = toFlussClientConfig(factoryConfiguration.toMap());
+ Map<String, String> tableProperties = toFlussTableOptions(factoryConfiguration.toMap());
+ Map<String, List<String>> bucketKeysMap =
+ parseBucketKeys(factoryConfiguration.get(BUCKET_KEY));
+ Map<String, Integer> bucketNumMap =
+ parseBucketNumber(factoryConfiguration.get(BUCKET_NUMBER));
+ return new FlussDataSink(flussClientConfig, tableProperties, bucketKeysMap, bucketNumMap);
+ }
+
+ @Override
+ public String identifier() {
+ return IDENTIFIER;
+ }
+
+ @Override
+ public Set<ConfigOption<?>> requiredOptions() {
+ Set<ConfigOption<?>> options = new HashSet<>();
+ options.add(BOOTSTRAP_SERVERS);
+ return options;
+ }
+
+ @Override
+ public Set<ConfigOption<?>> optionalOptions() {
+ Set<ConfigOption<?>> options = new HashSet<>();
+ options.add(BUCKET_KEY);
+ options.add(BUCKET_NUMBER);
+ return options;
+ }
+
+ private static Configuration toFlussClientConfig(Map<String, String> tableOptions) {
+ Configuration flussConfig = new Configuration();
+ flussConfig.setString(
+ ConfigOptions.BOOTSTRAP_SERVERS.key(), tableOptions.get(BOOTSTRAP_SERVERS.key()));
+
+ // forward all client configs
+ tableOptions.forEach(
+ (key, value) -> {
+ if (key.startsWith(CLIENT_PROPERTIES_PREFIX)) {
+ flussConfig.setString(key.substring("properties.".length()), value);
+ }
+ });
+ return flussConfig;
+ }
+
+ private static Map<String, String> toFlussTableOptions(Map<String, String> tableOptions) {
+ Map<String, String> flussTableConfig = new HashMap<>();
+ // forward all client configs
+ tableOptions.forEach(
+ (key, value) -> {
+ if (key.startsWith(TABLE_PROPERTIES_PREFIX)) {
+ flussTableConfig.put(key.substring("properties.".length()), value);
+ }
+ });
+ return flussTableConfig;
+ }
+}
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/CdcAsFlussRow.java b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-fluss/src/main/java/org/apache/flink/cdc/connectors/fluss/sink/CdcAsFlussRow.java
new file mode 100644
index 000000000..5891d58a7
--- /dev/null
+++ b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-fluss/src/main/java/org/apache/flink/cdc/connectors/fluss/sink/CdcAsFlussRow.java
@@ -0,0 +1,179 @@
+/*
+ * 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.connectors.fluss.sink;
+
+import org.apache.flink.cdc.common.annotation.Internal;
+import org.apache.flink.cdc.common.annotation.VisibleForTesting;
+import org.apache.flink.cdc.common.data.DecimalData;
+import org.apache.flink.cdc.common.data.LocalZonedTimestampData;
+import org.apache.flink.cdc.common.data.RecordData;
+import org.apache.flink.cdc.common.data.TimestampData;
+
+import com.alibaba.fluss.row.BinaryString;
+import com.alibaba.fluss.row.Decimal;
+import com.alibaba.fluss.row.InternalRow;
+import com.alibaba.fluss.row.TimestampLtz;
+import com.alibaba.fluss.row.TimestampNtz;
+
+import java.util.Map;
+import java.util.stream.Collectors;
+import java.util.stream.IntStream;
+
+/** Wraps a CDC {@link RecordData} as a Fluss {@link InternalRow}. */
+@Internal
+public class CdcAsFlussRow implements InternalRow {
+
+ private final RecordData cdcRecord;
+
+ /**
+ * Index mapping from Fluss Type to Cdc Type which will be used to convert CDC type to Fluss.
+ */
+ private final Map<Integer, Integer> indexMapping;
+
+ private final Integer rowFieldCount;
+
+ private CdcAsFlussRow(
+ RecordData cdcRecord, Integer rowFieldCount, Map<Integer, Integer> indexMapping) {
+ this.cdcRecord = cdcRecord;
+ this.rowFieldCount = rowFieldCount;
+ this.indexMapping = indexMapping;
+ }
+
+ public static CdcAsFlussRow replace(
+ RecordData cdcRecord, Integer rowFieldCount, Map<Integer, Integer> indexMapping) {
+ return new CdcAsFlussRow(cdcRecord, rowFieldCount, indexMapping);
+ }
+
+ public static CdcAsFlussRow replace(RecordData cdcRecord) {
+ return new CdcAsFlussRow(
+ cdcRecord,
+ cdcRecord.getArity(),
+ IntStream.range(0, cdcRecord.getArity())
+ .boxed()
+ .collect(Collectors.toMap(i -> i, i -> i)));
+ }
+
+ @Override
+ public int getFieldCount() {
+ return rowFieldCount;
+ }
+
+ @Override
+ public boolean isNullAt(int pos) {
+ if (indexMapping.get(pos) == null) {
+ return true;
+ }
+
+ return cdcRecord.isNullAt(indexMapping.get(pos));
+ }
+
+ @Override
+ public boolean getBoolean(int pos) {
+ return cdcRecord.getBoolean(indexMapping.get(pos));
+ }
+
+ @Override
+ public byte getByte(int pos) {
+ return cdcRecord.getByte(indexMapping.get(pos));
+ }
+
+ @Override
+ public short getShort(int pos) {
+ return cdcRecord.getShort(indexMapping.get(pos));
+ }
+
+ @Override
+ public int getInt(int pos) {
+ return cdcRecord.getInt(indexMapping.get(pos));
+ }
+
+ @Override
+ public long getLong(int pos) {
+ return cdcRecord.getLong(indexMapping.get(pos));
+ }
+
+ @Override
+ public float getFloat(int pos) {
+ return cdcRecord.getFloat(indexMapping.get(pos));
+ }
+
+ @Override
+ public double getDouble(int pos) {
+ return cdcRecord.getDouble(indexMapping.get(pos));
+ }
+
+ @Override
+ public BinaryString getChar(int pos, int length) {
+ return BinaryString.fromBytes(cdcRecord.getString(indexMapping.get(pos)).toBytes());
+ }
+
+ @Override
+ public BinaryString getString(int pos) {
+ return BinaryString.fromBytes(cdcRecord.getString(indexMapping.get(pos)).toBytes());
+ }
+
+ @Override
+ public Decimal getDecimal(int pos, int precision, int scale) {
+ return fromFlinkDecimal(cdcRecord.getDecimal(indexMapping.get(pos), precision, scale));
+ }
+
+ public static Decimal fromFlinkDecimal(DecimalData decimal) {
+ if (decimal.isCompact()) {
+ return Decimal.fromUnscaledLong(
+ decimal.toUnscaledLong(), decimal.precision(), decimal.scale());
+ } else {
+ return Decimal.fromBigDecimal(
+ decimal.toBigDecimal(), decimal.precision(), decimal.scale());
+ }
+ }
+
+ @Override
+ public TimestampNtz getTimestampNtz(int pos, int precision) {
+ TimestampData timestamp = cdcRecord.getTimestamp(indexMapping.get(pos), precision);
+ return TimestampNtz.fromMillis(
+ timestamp.getMillisecond(), timestamp.getNanoOfMillisecond());
+ }
+
+ @Override
+ public TimestampLtz getTimestampLtz(int pos, int precision) {
+ // Fluss ltz maybe mapping from CDC ltz or
+ LocalZonedTimestampData timestamp =
+ cdcRecord.getLocalZonedTimestampData(indexMapping.get(pos), precision);
+ return TimestampLtz.fromInstant(timestamp.toInstant());
+ }
+
+ @Override
+ public byte[] getBinary(int pos, int length) {
+ return cdcRecord.getBinary(indexMapping.get(pos));
+ }
+
+ @Override
+ public byte[] getBytes(int pos) {
+ return cdcRecord.getBinary(indexMapping.get(pos));
+ }
+
+ @VisibleForTesting
+ public RecordData getCdcRecord() {
+ return cdcRecord;
+ }
+
+ @VisibleForTesting
+ public Map<Integer, Integer> getIndexMapping() {
+ return indexMapping;
+ }
+}
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/FlussConfigUtils.java b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-fluss/src/main/java/org/apache/flink/cdc/connectors/fluss/sink/FlussConfigUtils.java
new file mode 100644
index 000000000..000cabe08
--- /dev/null
+++ b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-fluss/src/main/java/org/apache/flink/cdc/connectors/fluss/sink/FlussConfigUtils.java
@@ -0,0 +1,81 @@
+/*
+ * 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.connectors.fluss.sink;
+
+import javax.annotation.Nullable;
+
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+/** Utils for parsing fluss yaml sink options. */
+public class FlussConfigUtils {
+ public static Map<String, List<String>> parseBucketKeys(@Nullable String rawValue)
+ throws IllegalArgumentException {
+ Map<String, List<String>> result = new HashMap<>();
+ if (rawValue == null || rawValue.isEmpty()) {
+ return result;
+ }
+
+ for (String tableBucketKeyStr : rawValue.split(";")) {
+ if (tableBucketKeyStr.trim().isEmpty()) {
+ continue;
+ }
+
+ String[] tableAndBucketKeys = tableBucketKeyStr.trim().split(":", 2);
+ if (tableAndBucketKeys.length != 2) {
+ throw new IllegalArgumentException("Invalid bucket key configuration: " + rawValue);
+ }
+
+ String table = tableAndBucketKeys[0].trim();
+ List<String> keys = Arrays.asList(tableAndBucketKeys[1].trim().split(","));
+ result.put(table, keys);
+ }
+ return result;
+ }
+
+ public static Map<String, Integer> parseBucketNumber(@Nullable String rawValue)
+ throws IllegalArgumentException {
+ Map<String, Integer> result = new HashMap<>();
+ if (rawValue == null || rawValue.isEmpty()) {
+ return result;
+ }
+
+ for (String tableBucketNumStr : rawValue.split(";")) {
+ if (tableBucketNumStr.trim().isEmpty()) {
+ continue;
+ }
+ String[] kv = tableBucketNumStr.trim().split(":", 2);
+ if (kv.length != 2) {
+ throw new IllegalArgumentException(
+ "Invalid bucket number configuration: " + rawValue);
+ }
+
+ String table = kv[0].trim();
+ try {
+ int value = Integer.parseInt(kv[1].trim());
+ result.put(table, value);
+ } catch (NumberFormatException ignored) {
+ throw new IllegalArgumentException(
+ "Invalid bucket number configuration: " + rawValue);
+ }
+ }
+ return result;
+ }
+}
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
new file mode 100644
index 000000000..e03caf8e6
--- /dev/null
+++ 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
@@ -0,0 +1,68 @@
+/*
+ * 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.connectors.fluss.sink;
+
+import org.apache.flink.cdc.common.event.DataChangeEvent;
+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.MetadataApplier;
+import org.apache.flink.cdc.connectors.fluss.sink.v2.FlussSink;
+
+import com.alibaba.fluss.config.Configuration;
+
+import java.util.List;
+import java.util.Map;
+
+/** A DataSink implementation for Fluss. */
+public class FlussDataSink implements DataSink {
+
+ private final Configuration flussClientConfig;
+ private final Map<String, String> tableProperties;
+ private final Map<String, List<String>> bucketKeysMap;
+ private final Map<String, Integer> bucketNumMap;
+
+ public FlussDataSink(
+ Configuration flussClientConfig,
+ Map<String, String> tableProperties,
+ Map<String, List<String>> bucketKeysMap,
+ Map<String, Integer> bucketNumMap) {
+ this.flussClientConfig = flussClientConfig;
+ this.tableProperties = tableProperties;
+ this.bucketKeysMap = bucketKeysMap;
+ this.bucketNumMap = bucketNumMap;
+ }
+
+ @Override
+ public EventSinkProvider getEventSinkProvider() {
+ return FlinkSinkProvider.of(
+ new FlussSink<>(flussClientConfig, new FlussEventSerializationSchema()));
+ }
+
+ @Override
+ public MetadataApplier getMetadataApplier() {
+ return new FlussMetaDataApplier(
+ flussClientConfig, tableProperties, bucketKeysMap, bucketNumMap);
+ }
+
+ @Override
+ public HashFunctionProvider<DataChangeEvent> getDataChangeEventHashFunctionProvider() {
+ return new FlussHashFunctionProvider();
+ }
+}
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
new file mode 100644
index 000000000..f14fc1b67
--- /dev/null
+++ 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
@@ -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.connectors.fluss.sink;
+
+import org.apache.flink.cdc.common.configuration.ConfigOption;
+import org.apache.flink.cdc.common.configuration.ConfigOptions;
+
+/** Options for {@link FlussDataSink}. */
+public class FlussDataSinkOptions {
+
+ // prefix for passing properties for fluss options.
+ public static final String TABLE_PROPERTIES_PREFIX = "properties.table.";
+ public static final String CLIENT_PROPERTIES_PREFIX = "properties.client.";
+
+ public static final ConfigOption<String> BOOTSTRAP_SERVERS =
+ ConfigOptions.key("bootstrap.servers")
+ .stringType()
+ .noDefaultValue()
+ .withDescription("The bootstrap servers for the Fluss sink connection.");
+
+ public static final ConfigOption<String> BUCKET_KEY =
+ ConfigOptions.key("bucket.key")
+ .stringType()
+ .noDefaultValue()
+ .withDescription(
+ "Specific the distribution policy of each Fluss table. "
+ + "Tables are separated by ';', and bucket keys are separated by ','. "
+ + "Format: database1.table1:key1,key2;database1.table2:key3. \""
+ + "Data will be distributed to each bucket according to the hash value of bucket-key (It must be a subset of the primary keys excluding partition keys of the primary key table). "
+ + "If the table has a primary key and a bucket key is not specified, the bucket key will be used as primary key(excluding the partition key). "
+ + "If the table has no primary key and the bucket key is not specified, "
+ + "the data will be distributed to each bucket randomly.");
+
+ public static final ConfigOption<String> BUCKET_NUMBER =
+ ConfigOptions.key("bucket.num")
+ .stringType()
+ .noDefaultValue()
+ .withDescription(
+ "The number of buckets of each Fluss table."
+ + "Tables are separated by ';'. "
+ + "Format: database1.table1:4;database1.table2:8.");
+}
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/FlussEventSerializationSchema.java b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-fluss/src/main/java/org/apache/flink/cdc/connectors/fluss/sink/FlussEventSerializationSchema.java
new file mode 100644
index 000000000..d946916b6
--- /dev/null
+++ b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-fluss/src/main/java/org/apache/flink/cdc/connectors/fluss/sink/FlussEventSerializationSchema.java
@@ -0,0 +1,184 @@
+/*
+ * 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.connectors.fluss.sink;
+
+import org.apache.flink.cdc.common.event.CreateTableEvent;
+import org.apache.flink.cdc.common.event.DataChangeEvent;
+import org.apache.flink.cdc.common.event.Event;
+import org.apache.flink.cdc.common.event.OperationType;
+import org.apache.flink.cdc.common.event.SchemaChangeEvent;
+import org.apache.flink.cdc.common.event.TableId;
+import org.apache.flink.cdc.common.utils.Preconditions;
+import org.apache.flink.cdc.connectors.fluss.sink.v2.FlussEvent;
+import org.apache.flink.cdc.connectors.fluss.sink.v2.FlussEventSerializer;
+import org.apache.flink.cdc.connectors.fluss.sink.v2.FlussRowWithOp;
+
+import com.alibaba.fluss.client.Connection;
+import com.alibaba.fluss.client.table.Table;
+import com.alibaba.fluss.metadata.TablePath;
+import com.alibaba.fluss.types.DataType;
+
+import java.io.IOException;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import static org.apache.flink.cdc.connectors.fluss.sink.v2.FlussOperationType.APPEND;
+import static org.apache.flink.cdc.connectors.fluss.sink.v2.FlussOperationType.DELETE;
+import static org.apache.flink.cdc.connectors.fluss.sink.v2.FlussOperationType.UPSERT;
+import static org.apache.flink.cdc.connectors.fluss.utils.FlussConversions.sameCdcColumnsIgnoreCommentAndDefaultValue;
+import static org.apache.flink.cdc.connectors.fluss.utils.FlussConversions.toFlussSchema;
+
+/** Serialization schema that converts a CDC data record to a Fluss event. */
+public class FlussEventSerializationSchema implements FlussEventSerializer<Event> {
+ private static final long serialVersionUID = 1L;
+
+ private transient Map<TableId, TableSchemaInfo> tableInfoMap;
+ private transient Connection connection;
+
+ @Override
+ public void open(Connection connection) {
+ this.tableInfoMap = new HashMap<>();
+ this.connection = connection;
+ }
+
+ @Override
+ public FlussEvent serialize(Event event) throws IOException {
+ if (event instanceof SchemaChangeEvent) {
+ applySchemaChangeEvent((SchemaChangeEvent) event);
+ return new FlussEvent(getTablePath(((SchemaChangeEvent) event).tableId()), null, true);
+ } else if (event instanceof DataChangeEvent) {
+ FlussRowWithOp rowWithOp = applyDataChangeEvent((DataChangeEvent) event);
+ return new FlussEvent(
+ getTablePath(((DataChangeEvent) event).tableId()),
+ Collections.singletonList(rowWithOp),
+ false);
+
+ } else {
+ throw new UnsupportedOperationException("Don't support event " + event);
+ }
+ }
+
+ private void applySchemaChangeEvent(SchemaChangeEvent event) {
+ TableId tableId = event.tableId();
+ if (event instanceof CreateTableEvent) {
+ org.apache.flink.cdc.common.schema.Schema newSchema =
+ ((CreateTableEvent) event).getSchema();
+ // if the table is not exist or the schema is changed, update the table info.
+ if (!tableInfoMap.containsKey(tableId)
+ || !sameCdcColumnsIgnoreCommentAndDefaultValue(
+ tableInfoMap.get(tableId).upstreamCdcSchema, newSchema)) {
+ Table table = connection.getTable(getTablePath(tableId));
+ TableSchemaInfo newSchemaInfo =
+ new TableSchemaInfo(newSchema, table.getTableInfo().getSchema());
+ tableInfoMap.put(tableId, newSchemaInfo);
+ }
+ } else {
+ // TODO: Logics for altering tables are not supported yet.
+ // This is anticipated to be supported in Fluss version 0.8.0.
+ throw new RuntimeException(
+ "Schema change type not supported. Only CreateTableEvent is allowed at the moment.");
+ }
+ }
+
+ private FlussRowWithOp applyDataChangeEvent(DataChangeEvent record) {
+ OperationType op = record.op();
+ TableSchemaInfo tableSchemaInfo = tableInfoMap.get(record.tableId());
+ Preconditions.checkNotNull(
+ tableSchemaInfo, "Table schema not found for table " + record.tableId());
+ int flussFieldCount =
+ tableSchemaInfo.downStreamFlusstreamSchema.getRowType().getFieldCount();
+ boolean hasPrimaryKey = !tableSchemaInfo.upstreamCdcSchema.primaryKeys().isEmpty();
+ switch (op) {
+ case INSERT:
+ case UPDATE:
+ case REPLACE:
+ return new FlussRowWithOp(
+ CdcAsFlussRow.replace(
+ record.after(), flussFieldCount, tableSchemaInfo.indexMapping),
+ hasPrimaryKey ? UPSERT : APPEND);
+ case DELETE:
+ return new FlussRowWithOp(
+ CdcAsFlussRow.replace(
+ record.before(), flussFieldCount, tableSchemaInfo.indexMapping),
+ DELETE);
+ default:
+ throw new IllegalArgumentException("Unsupported row kind: " + op);
+ }
+ }
+
+ private TablePath getTablePath(TableId tableId) {
+ return TablePath.of(tableId.getSchemaName(), tableId.getTableName());
+ }
+
+ private static class TableSchemaInfo {
+ org.apache.flink.cdc.common.schema.Schema upstreamCdcSchema;
+ com.alibaba.fluss.metadata.Schema downStreamFlusstreamSchema;
+ Map<Integer, Integer> indexMapping;
+
+ private TableSchemaInfo(
+ org.apache.flink.cdc.common.schema.Schema upstreamCdcSchema,
+ com.alibaba.fluss.metadata.Schema downStreamFlusstreamSchema) {
+ this.upstreamCdcSchema = upstreamCdcSchema;
+ this.downStreamFlusstreamSchema = downStreamFlusstreamSchema;
+ this.indexMapping =
+ sanityCheckAndGenerateIndexMapping(
+ toFlussSchema(upstreamCdcSchema), downStreamFlusstreamSchema);
+ }
+ }
+
+ static Map<Integer, Integer> sanityCheckAndGenerateIndexMapping(
+ com.alibaba.fluss.metadata.Schema inferredFlussSchema,
+ com.alibaba.fluss.metadata.Schema currentFlussnewSchema) {
+ List<String> inferredSchemaColumnNames = inferredFlussSchema.getColumnNames();
+ Map<String, Integer> reverseIndex = new HashMap<>();
+ for (int i = 0; i < inferredSchemaColumnNames.size(); i++) {
+ reverseIndex.put(inferredSchemaColumnNames.get(i), i);
+ }
+
+ List<String> currentSchemaColumnNames = currentFlussnewSchema.getColumnNames();
+ Map<Integer, Integer> indexMapping = new HashMap<>();
+ for (int newSchemaIndex = 0;
+ newSchemaIndex < currentSchemaColumnNames.size();
+ newSchemaIndex++) {
+ String columnName = currentSchemaColumnNames.get(newSchemaIndex);
+ if (reverseIndex.get(columnName) != null) {
+ Integer oldSchemaIndex = reverseIndex.get(columnName);
+ indexMapping.put(newSchemaIndex, oldSchemaIndex);
+
+ // Currently, we only support mismatched column counts between upstream and
+ // downstream, but not mismatched data types, to prevent errors caused by type
+ // changes.
+ // In the future, meta applier will be used to handle column changes.
+ DataType oldDataType = inferredFlussSchema.getRowType().getTypeAt(oldSchemaIndex);
+ DataType newDataType = currentFlussnewSchema.getRowType().getTypeAt(newSchemaIndex);
+ if (!oldDataType.copy(false).equals(newDataType.copy(false))) {
+ throw new IllegalArgumentException(
+ "The data type of column "
+ + columnName
+ + " is changed from "
+ + oldDataType
+ + " to "
+ + newDataType);
+ }
+ }
+ }
+ return indexMapping;
+ }
+}
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/FlussHashFunctionProvider.java b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-fluss/src/main/java/org/apache/flink/cdc/connectors/fluss/sink/FlussHashFunctionProvider.java
new file mode 100644
index 000000000..c043ab791
--- /dev/null
+++ b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-fluss/src/main/java/org/apache/flink/cdc/connectors/fluss/sink/FlussHashFunctionProvider.java
@@ -0,0 +1,101 @@
+/*
+ * 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.connectors.fluss.sink;
+
+import org.apache.flink.cdc.common.data.RecordData;
+import org.apache.flink.cdc.common.event.DataChangeEvent;
+import org.apache.flink.cdc.common.event.OperationType;
+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;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Objects;
+import java.util.Optional;
+import java.util.concurrent.ThreadLocalRandom;
+
+/** {@link HashFunctionProvider} implementation for {@link FlussDataSink}. */
+public class FlussHashFunctionProvider implements HashFunctionProvider<DataChangeEvent> {
+ @Override
+ public HashFunction<DataChangeEvent> getHashFunction(@Nullable TableId tableId, Schema schema) {
+ return new FlussHashFunction(schema);
+ }
+
+ static class FlussHashFunction implements HashFunction<DataChangeEvent> {
+ private final List<RecordData.FieldGetter> primaryKeyGetters;
+
+ public FlussHashFunction(Schema schema) {
+ primaryKeyGetters = createFieldGetters(schema);
+ }
+
+ @Override
+ public int hashcode(DataChangeEvent event) {
+ List<Object> objectsToHash = new ArrayList<>();
+ // Table ID
+ TableId tableId = event.tableId();
+ Optional.ofNullable(tableId.getNamespace()).ifPresent(objectsToHash::add);
+ Optional.ofNullable(tableId.getSchemaName()).ifPresent(objectsToHash::add);
+ objectsToHash.add(tableId.getTableName());
+
+ // Primary key
+ if (!primaryKeyGetters.isEmpty()) {
+ RecordData data =
+ event.op().equals(OperationType.DELETE) ? event.before() : event.after();
+ for (RecordData.FieldGetter primaryKeyGetter : primaryKeyGetters) {
+ objectsToHash.add(primaryKeyGetter.getFieldOrNull(data));
+ }
+ } else {
+ // Avoid sending all events to the same subtask when table has no primary key.
+ objectsToHash.add(ThreadLocalRandom.current().nextInt());
+ }
+
+ // Calculate hash
+ return (Objects.hash(objectsToHash.toArray()) * 31) & 0x7FFFFFFF;
+ }
+
+ private List<RecordData.FieldGetter> createFieldGetters(Schema schema) {
+ List<RecordData.FieldGetter> fieldGetters =
+ new ArrayList<>(schema.primaryKeys().size());
+ schema.primaryKeys().stream()
+ .mapToInt(
+ pk -> {
+ int index = schema.getColumnNames().indexOf(pk);
+ if (index == -1) {
+ throw new IllegalStateException(
+ String.format(
+ "Unable to find column \"%s\" which is defined as primary key",
+ pk));
+ }
+ return index;
+ })
+ .forEach(
+ primaryKeyPosition ->
+ fieldGetters.add(
+ RecordData.createFieldGetter(
+ schema.getColumns()
+ .get(primaryKeyPosition)
+ .getType(),
+ primaryKeyPosition)));
+ return fieldGetters;
+ }
+ }
+}
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/FlussMetaDataApplier.java b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-fluss/src/main/java/org/apache/flink/cdc/connectors/fluss/sink/FlussMetaDataApplier.java
new file mode 100644
index 000000000..9ab7a6f78
--- /dev/null
+++ b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-fluss/src/main/java/org/apache/flink/cdc/connectors/fluss/sink/FlussMetaDataApplier.java
@@ -0,0 +1,181 @@
+/*
+ * 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.connectors.fluss.sink;
+
+import org.apache.flink.cdc.common.event.CreateTableEvent;
+import org.apache.flink.cdc.common.event.DropTableEvent;
+import org.apache.flink.cdc.common.event.SchemaChangeEvent;
+import org.apache.flink.cdc.common.event.SchemaChangeEventType;
+import org.apache.flink.cdc.common.event.SchemaChangeEventTypeFamily;
+import org.apache.flink.cdc.common.event.TableId;
+import org.apache.flink.cdc.common.sink.MetadataApplier;
+import org.apache.flink.table.api.ValidationException;
+
+import com.alibaba.fluss.client.Connection;
+import com.alibaba.fluss.client.ConnectionFactory;
+import com.alibaba.fluss.client.admin.Admin;
+import com.alibaba.fluss.config.Configuration;
+import com.alibaba.fluss.metadata.DatabaseDescriptor;
+import com.alibaba.fluss.metadata.TableDescriptor;
+import com.alibaba.fluss.metadata.TableInfo;
+import com.alibaba.fluss.metadata.TablePath;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.Arrays;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+import static org.apache.flink.cdc.common.event.SchemaChangeEventType.CREATE_TABLE;
+import static org.apache.flink.cdc.common.event.SchemaChangeEventType.DROP_TABLE;
+import static org.apache.flink.cdc.connectors.fluss.utils.FlussConversions.toFlussTable;
+
+/** {@link MetadataApplier} for fluss. */
+public class FlussMetaDataApplier implements MetadataApplier {
+ private static final Logger LOG = LoggerFactory.getLogger(FlussMetaDataApplier.class);
+ private final Configuration flussClientConfig;
+ private final Map<String, String> tableProperties;
+ private final Map<String, List<String>> bucketKeysMap;
+ private final Map<String, Integer> bucketNumMap;
+ private Set<SchemaChangeEventType> enabledEventTypes =
+ new HashSet<>(Arrays.asList(CREATE_TABLE, DROP_TABLE));
+
+ public FlussMetaDataApplier(
+ Configuration flussClientConfig,
+ Map<String, String> tableProperties,
+ Map<String, List<String>> bucketKeysMap,
+ Map<String, Integer> bucketNumMap) {
+ this.flussClientConfig = flussClientConfig;
+ this.tableProperties = tableProperties;
+ this.bucketKeysMap = bucketKeysMap;
+ this.bucketNumMap = bucketNumMap;
+ }
+
+ @Override
+ public MetadataApplier setAcceptedSchemaEvolutionTypes(
+ Set<SchemaChangeEventType> schemaEvolutionTypes) {
+ enabledEventTypes = schemaEvolutionTypes;
+ return this;
+ }
+
+ @Override
+ public boolean acceptsSchemaEvolutionType(SchemaChangeEventType schemaChangeEventType) {
+ return enabledEventTypes.contains(schemaChangeEventType);
+ }
+
+ @Override
+ public Set<SchemaChangeEventType> getSupportedSchemaEvolutionTypes() {
+ return Arrays.stream(SchemaChangeEventTypeFamily.TABLE).collect(Collectors.toSet());
+ }
+
+ @Override
+ public void applySchemaChange(SchemaChangeEvent schemaChangeEvent) {
+ LOG.info("fluss metadata applier receive schemaChangeEvent {}", schemaChangeEvent);
+ if (schemaChangeEvent instanceof CreateTableEvent) {
+ CreateTableEvent createTableEvent = (CreateTableEvent) schemaChangeEvent;
+ applyCreateTable(createTableEvent);
+ } else if (schemaChangeEvent instanceof DropTableEvent) {
+ DropTableEvent dropTableEvent = (DropTableEvent) schemaChangeEvent;
+ applyDropTable(dropTableEvent);
+ } else {
+ throw new IllegalArgumentException(
+ "fluss metadata applier only support CreateTableEvent now but receives "
+ + schemaChangeEvent);
+ }
+ }
+
+ private void applyCreateTable(CreateTableEvent event) {
+ try (Connection connection = ConnectionFactory.createConnection(flussClientConfig);
+ Admin admin = connection.getAdmin()) {
+ TableId tableId = event.tableId();
+ TablePath tablePath = new TablePath(tableId.getSchemaName(), tableId.getTableName());
+ String tableIdentifier = tablePath.getDatabaseName() + "." + tablePath.getTableName();
+ List<String> bucketKeys = bucketKeysMap.get(tableIdentifier);
+ Integer bucketNum = bucketNumMap.get(tableIdentifier);
+ TableDescriptor inferredFlussTable =
+ toFlussTable(event.getSchema(), bucketKeys, bucketNum, tableProperties);
+ admin.createDatabase(tablePath.getDatabaseName(), DatabaseDescriptor.EMPTY, true);
+ if (!admin.tableExists(tablePath).get()) {
+ admin.createTable(tablePath, inferredFlussTable, false).get();
+ } else {
+ TableInfo currentTableInfo = admin.getTableInfo(tablePath).get();
+ // sanity check to prevent unexpected table schema evolution.
+ sanityCheck(inferredFlussTable, currentTableInfo);
+ }
+ } catch (Exception e) {
+ LOG.error("Failed to apply schema change {}", event, e);
+ throw new RuntimeException(e);
+ }
+ }
+
+ private void applyDropTable(DropTableEvent event) {
+ try (Connection connection = ConnectionFactory.createConnection(flussClientConfig);
+ Admin admin = connection.getAdmin()) {
+ TableId tableId = event.tableId();
+ TablePath tablePath = new TablePath(tableId.getSchemaName(), tableId.getTableName());
+ admin.dropTable(tablePath, true).get();
+ } catch (Exception e) {
+ LOG.error("Failed to apply schema change {}", event, e);
+ throw new RuntimeException(e);
+ }
+ }
+
+ private void sanityCheck(TableDescriptor inferredFlussTable, TableInfo currentTableInfo) {
+ List<String> inferredPrimaryKeyColumnNames =
+ inferredFlussTable.getSchema().getPrimaryKeyColumnNames().stream()
+ .sorted()
+ .collect(Collectors.toList());
+ List<String> currentPrimaryKeyColumnNames =
+ currentTableInfo.getSchema().getPrimaryKeyColumnNames().stream()
+ .sorted()
+ .collect(Collectors.toList());
+ if (!inferredPrimaryKeyColumnNames.equals(currentPrimaryKeyColumnNames)) {
+ throw new ValidationException(
+ "The table schema inffered by Flink CDC is not matched with current Fluss table schema. "
+ + "\n New Fluss table's primary keys : "
+ + inferredPrimaryKeyColumnNames
+ + "\n Current Fluss's primary keys: "
+ + currentPrimaryKeyColumnNames);
+ }
+
+ List<String> inferredBucketKeys = inferredFlussTable.getBucketKeys();
+ List<String> currentBucketKeys = currentTableInfo.getBucketKeys();
+ if (!inferredBucketKeys.equals(currentBucketKeys)) {
+ throw new ValidationException(
+ "The table schema inffered by Flink CDC is not matched with current Fluss table schema. "
+ + "\n New Fluss table's bucket keys : "
+ + inferredBucketKeys
+ + "\n Current Fluss's bucket keys: "
+ + currentBucketKeys);
+ }
+
+ List<String> inferredPartitionKeys = inferredFlussTable.getPartitionKeys();
+ List<String> currentPartitionKeys = currentTableInfo.getPartitionKeys();
+ if (!inferredPartitionKeys.equals(currentPartitionKeys)) {
+ throw new ValidationException(
+ "The table schema inffered by Flink CDC is not matched with current Fluss table schema. "
+ + "\n New Fluss table's partition keys : "
+ + inferredPartitionKeys
+ + "\n Current Fluss's partition keys: "
+ + currentPartitionKeys);
+ }
+ }
+}
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/v2/FlussEvent.java b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-fluss/src/main/java/org/apache/flink/cdc/connectors/fluss/sink/v2/FlussEvent.java
new file mode 100644
index 000000000..75df49fa6
--- /dev/null
+++ b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-fluss/src/main/java/org/apache/flink/cdc/connectors/fluss/sink/v2/FlussEvent.java
@@ -0,0 +1,49 @@
+/*
+ * 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.connectors.fluss.sink.v2;
+
+import com.alibaba.fluss.metadata.TablePath;
+
+import java.util.List;
+
+/** FlussEvent is a wrapper for a list of genericRows. */
+public class FlussEvent {
+ private final TablePath tablePath;
+ private final List<FlussRowWithOp> rowWithOps;
+ // if true, means that table schema has changed right before this genericRow.
+ boolean shouldRefreshSchema;
+
+ public FlussEvent(
+ TablePath tablePath, List<FlussRowWithOp> rowWithOps, boolean shouldRefreshSchema) {
+ this.tablePath = tablePath;
+ this.rowWithOps = rowWithOps;
+ this.shouldRefreshSchema = shouldRefreshSchema;
+ }
+
+ public TablePath getTablePath() {
+ return tablePath;
+ }
+
+ public List<FlussRowWithOp> getRowWithOps() {
+ return rowWithOps;
+ }
+
+ public boolean isShouldRefreshSchema() {
+ return shouldRefreshSchema;
+ }
+}
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/v2/FlussEventSerializer.java b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-fluss/src/main/java/org/apache/flink/cdc/connectors/fluss/sink/v2/FlussEventSerializer.java
new file mode 100644
index 000000000..dadf0d2ee
--- /dev/null
+++ b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-fluss/src/main/java/org/apache/flink/cdc/connectors/fluss/sink/v2/FlussEventSerializer.java
@@ -0,0 +1,35 @@
+/*
+ * 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.connectors.fluss.sink.v2;
+
+import com.alibaba.fluss.client.Connection;
+
+import java.io.IOException;
+import java.io.Serializable;
+
+/**
+ * Serializer to serialize the input record to a {@link FlussEvent} for {@link FlussSinkWriter}.
+ *
+ * @param <InputT> The type of the input record which comes from the upstream and will be
+ * transformed into a FlussEvent here.
+ */
+public interface FlussEventSerializer<InputT> extends Serializable {
+ void open(Connection connection) throws IOException;
+
+ FlussEvent serialize(InputT in) throws IOException;
+}
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/v2/FlussOperationType.java b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-fluss/src/main/java/org/apache/flink/cdc/connectors/fluss/sink/v2/FlussOperationType.java
new file mode 100644
index 000000000..e8d1df9ef
--- /dev/null
+++ b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-fluss/src/main/java/org/apache/flink/cdc/connectors/fluss/sink/v2/FlussOperationType.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.connectors.fluss.sink.v2;
+
+/* This file is based on source code of Apache Fluss Project (https://fluss.apache.org/), licensed by the Apache
+ * Software Foundation (ASF) under the Apache License, Version 2.0. See the NOTICE file distributed with this work for
+ * additional information regarding copyright ownership. */
+
+/**
+ * Enumeration of row operation types used in Fluss-Flink data processing.
+ *
+ * <p>This enum represents the type of operation associated with a row, such as an append (insert),
+ * upsert (update or insert), delete, or ignore. It is used to indicate how a row should be
+ * interpreted or processed in downstream systems.
+ *
+ * <ul>
+ * <li>{@link #APPEND} - Represents an append-only (insert) operation.
+ * <li>{@link #UPSERT} - Represents an upsert operation (update or insert).
+ * <li>{@link #DELETE} - Represents a delete operation.
+ * <li>{@link #IGNORE} - Represents an operation that should be ignored.
+ * </ul>
+ *
+ * @see FlussRowWithOp
+ */
+public enum FlussOperationType {
+ /** Represents an append-only (insert) operation. */
+ APPEND,
+
+ /** Represents an upsert operation (update or insert). */
+ UPSERT,
+
+ /** Represents a delete operation. */
+ DELETE,
+
+ /** Represents an operation that should be ignored. */
+ IGNORE
+}
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/v2/FlussRowWithOp.java b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-fluss/src/main/java/org/apache/flink/cdc/connectors/fluss/sink/v2/FlussRowWithOp.java
new file mode 100644
index 000000000..0dedaf6cc
--- /dev/null
+++ b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-fluss/src/main/java/org/apache/flink/cdc/connectors/fluss/sink/v2/FlussRowWithOp.java
@@ -0,0 +1,104 @@
+/*
+ * 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.connectors.fluss.sink.v2;
+
+import com.alibaba.fluss.row.InternalRow;
+
+import javax.annotation.Nullable;
+
+import java.util.Objects;
+
+import static com.alibaba.fluss.utils.Preconditions.checkNotNull;
+
+/* This file is based on source code of Apache Fluss Project (https://fluss.apache.org/), licensed by the Apache
+ * Software Foundation (ASF) under the Apache License, Version 2.0. See the NOTICE file distributed with this work for
+ * additional information regarding copyright ownership. */
+
+/**
+ * A wrapper class that associates an {@link InternalRow} with an {@link FlussOperationType} for use
+ * in Fluss-Flink data processing.
+ *
+ * <p>This class is used to represent a row of data along with its corresponding operation type,
+ * such as APPEND, UPSERT, or DELETE, as defined by {@link FlussOperationType}.
+ *
+ * @see InternalRow
+ * @see FlussOperationType
+ */
+public class FlussRowWithOp {
+ /** The internal row data. */
+ private final InternalRow row;
+
+ /** The type of operation associated with this row (e.g., APPEND, UPSERT, DELETE). */
+ private final FlussOperationType opType;
+
+ /**
+ * Constructs a {@code RowWithOp} with the specified internal row and operation type.
+ *
+ * @param row the internal row data (must not be null)
+ * @param opType the operation type (must not be null)
+ * @throws NullPointerException if {@code row} or {@code opType} is null
+ */
+ public FlussRowWithOp(InternalRow row, @Nullable FlussOperationType opType) {
+ this.row = checkNotNull(row, "row cannot be null");
+ this.opType = checkNotNull(opType, "opType cannot be null");
+ }
+
+ /**
+ * Returns the internal row data.
+ *
+ * @return the internal row
+ */
+ public InternalRow getRow() {
+ return row;
+ }
+
+ /**
+ * Returns the operation type associated with this row.
+ *
+ * @return the operation type
+ */
+ public FlussOperationType getOperationType() {
+ return opType;
+ }
+
+ /**
+ * Indicates whether some other object is "equal to" this one. Two {@code RowWithOp} objects are
+ * considered equal if their internal rows and operation types are equal.
+ *
+ * @param o the reference object with which to compare
+ * @return {@code true} if this object is the same as the obj argument; {@code false} otherwise
+ */
+ @Override
+ public boolean equals(Object o) {
+ if (o == null || getClass() != o.getClass()) {
+ return false;
+ }
+ FlussRowWithOp rowWithOp = (FlussRowWithOp) o;
+ return Objects.equals(row, rowWithOp.row) && opType == rowWithOp.opType;
+ }
+
+ /**
+ * Returns a hash code value for the object.
+ *
+ * @return a hash code value for this object
+ */
+ @Override
+ public int hashCode() {
+ return Objects.hash(row, opType);
+ }
+}
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/v2/FlussSink.java b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-fluss/src/main/java/org/apache/flink/cdc/connectors/fluss/sink/v2/FlussSink.java
new file mode 100644
index 000000000..4e6d789e9
--- /dev/null
+++ b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-fluss/src/main/java/org/apache/flink/cdc/connectors/fluss/sink/v2/FlussSink.java
@@ -0,0 +1,62 @@
+/*
+ * 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.connectors.fluss.sink.v2;
+
+import org.apache.flink.api.connector.sink2.Sink;
+import org.apache.flink.api.connector.sink2.SinkWriter;
+import org.apache.flink.api.connector.sink2.WriterInitContext;
+import org.apache.flink.runtime.metrics.groups.InternalSinkWriterMetricGroup;
+
+import com.alibaba.fluss.config.Configuration;
+
+import java.io.IOException;
+
+/* This file is based on source code of Apache Fluss Project (https://fluss.apache.org/), licensed by the Apache
+ * Software Foundation (ASF) under the Apache License, Version 2.0. See the NOTICE file distributed with this work for
+ * additional information regarding copyright ownership. */
+
+/**
+ * Flink sink for Fluss.
+ *
+ * @param <InputT> The type of the input elements.
+ */
+public class FlussSink<InputT> implements Sink<InputT> {
+ private final FlussEventSerializer<InputT> serializer;
+ private final Configuration flussConfig;
+
+ public FlussSink(Configuration flussConfig, FlussEventSerializer<InputT> serializer) {
+ this.serializer = serializer;
+ this.flussConfig = flussConfig;
+ }
+
+ @Deprecated
+ public SinkWriter<InputT> createWriter(InitContext context) throws IOException {
+ FlussSinkWriter<InputT> flinkSinkWriter =
+ new FlussSinkWriter<>(flussConfig, context.getMailboxExecutor(), serializer);
+ flinkSinkWriter.initialize(InternalSinkWriterMetricGroup.wrap(context.metricGroup()));
+ return flinkSinkWriter;
+ }
+
+ @Override
+ public SinkWriter<InputT> createWriter(WriterInitContext context) throws IOException {
+ FlussSinkWriter<InputT> flinkSinkWriter =
+ new FlussSinkWriter<>(flussConfig, context.getMailboxExecutor(), serializer);
+ flinkSinkWriter.initialize(InternalSinkWriterMetricGroup.wrap(context.metricGroup()));
+ return flinkSinkWriter;
+ }
+}
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/v2/FlussSinkWriter.java b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-fluss/src/main/java/org/apache/flink/cdc/connectors/fluss/sink/v2/FlussSinkWriter.java
new file mode 100644
index 000000000..dab9e6ed5
--- /dev/null
+++ b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-fluss/src/main/java/org/apache/flink/cdc/connectors/fluss/sink/v2/FlussSinkWriter.java
@@ -0,0 +1,258 @@
+/*
+ * 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.connectors.fluss.sink.v2;
+
+import org.apache.flink.api.common.operators.MailboxExecutor;
+import org.apache.flink.api.connector.sink2.SinkWriter;
+import org.apache.flink.cdc.connectors.fluss.sink.v2.metrics.WrapperFlussMetricRegistry;
+import org.apache.flink.metrics.Counter;
+import org.apache.flink.metrics.groups.SinkWriterMetricGroup;
+
+import com.alibaba.fluss.client.Connection;
+import com.alibaba.fluss.client.ConnectionFactory;
+import com.alibaba.fluss.client.table.Table;
+import com.alibaba.fluss.client.table.writer.AppendWriter;
+import com.alibaba.fluss.client.table.writer.TableWriter;
+import com.alibaba.fluss.client.table.writer.UpsertWriter;
+import com.alibaba.fluss.config.Configuration;
+import com.alibaba.fluss.metadata.TablePath;
+import com.alibaba.fluss.metrics.Gauge;
+import com.alibaba.fluss.metrics.Metric;
+import com.alibaba.fluss.metrics.MetricNames;
+import com.alibaba.fluss.row.InternalRow;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.CompletableFuture;
+
+/** Base class for Flink {@link SinkWriter} implementations in Fluss. */
+public class FlussSinkWriter<InputT> implements SinkWriter<InputT> {
+
+ protected static final Logger LOG = LoggerFactory.getLogger(FlussSinkWriter.class);
+
+ private final Configuration flussConfig;
+ private final MailboxExecutor mailboxExecutor;
+ private final FlussEventSerializer<InputT> flussRecordSerializer;
+
+ private transient Connection connection;
+ protected transient WrapperFlussMetricRegistry flinkMetricRegistry;
+
+ protected transient SinkWriterMetricGroup metricGroup;
+
+ private transient Counter numRecordsOutCounter;
+ private transient Counter numRecordsOutErrorsCounter;
+ private volatile Throwable asyncWriterException;
+
+ private final Map<TablePath, TableWriter> writerMap;
+ private final Map<TablePath, Table> tableMap;
+
+ public FlussSinkWriter(
+ Configuration flussConfig,
+ MailboxExecutor mailboxExecutor,
+ FlussEventSerializer<InputT> flussRecordSerializer) {
+ this.flussConfig = flussConfig;
+ this.mailboxExecutor = mailboxExecutor;
+ this.flussRecordSerializer = flussRecordSerializer;
+ this.writerMap = new HashMap<>();
+ this.tableMap = new HashMap<>();
+ }
+
+ public void initialize(SinkWriterMetricGroup metricGroup) throws IOException {
+ LOG.info("Opening Fluss with config {}", flussConfig);
+ this.metricGroup = metricGroup;
+ flinkMetricRegistry =
+ new WrapperFlussMetricRegistry(
+ metricGroup, Collections.singleton(MetricNames.WRITER_SEND_LATENCY_MS));
+ connection = ConnectionFactory.createConnection(flussConfig, flinkMetricRegistry);
+ flussRecordSerializer.open(connection);
+
+ initMetrics();
+ }
+
+ protected void initMetrics() {
+ numRecordsOutCounter = metricGroup.getNumRecordsSendCounter();
+ numRecordsOutErrorsCounter = metricGroup.getNumRecordsOutErrorsCounter();
+ metricGroup.setCurrentSendTimeGauge(this::computeSendTime);
+ }
+
+ @Override
+ public void write(InputT inputValue, Context context) throws IOException {
+ checkAsyncException();
+
+ try {
+ FlussEvent flussEvent = flussRecordSerializer.serialize(inputValue);
+
+ TablePath tablePath = flussEvent.getTablePath();
+
+ if (flussEvent.isShouldRefreshSchema() || !writerMap.containsKey(tablePath)) {
+ // refresh table schema
+ if (tableMap.containsKey(tablePath)) {
+ Table table = tableMap.remove(tablePath);
+ writerMap.remove(tablePath);
+ table.close();
+ }
+
+ Table table = connection.getTable(tablePath);
+ TableWriter writer;
+ if (table.getTableInfo().hasPrimaryKey()) {
+ writer = table.newUpsert().createWriter();
+ } else {
+ writer = table.newAppend().createWriter();
+ }
+ tableMap.put(tablePath, table);
+ writerMap.put(tablePath, writer);
+ }
+
+ List<FlussRowWithOp> rowWithOps = flussEvent.getRowWithOps();
+ if (rowWithOps == null) {
+ return;
+ }
+ for (FlussRowWithOp rowWithOp : rowWithOps) {
+ FlussOperationType opType = rowWithOp.getOperationType();
+ InternalRow row = rowWithOp.getRow();
+ if (opType == FlussOperationType.IGNORE) {
+ // skip writing the row
+ return;
+ }
+ CompletableFuture<?> writeFuture =
+ write(writerMap.get(tablePath), opType, row, tablePath);
+ writeFuture.whenComplete(
+ (ignored, throwable) -> {
+ if (throwable != null) {
+ if (this.asyncWriterException == null) {
+ this.asyncWriterException = throwable;
+ }
+
+ // Checking for exceptions from previous writes
+ mailboxExecutor.execute(
+ this::checkAsyncException, "Update error metric");
+ }
+ });
+
+ numRecordsOutCounter.inc();
+ }
+
+ } catch (Exception e) {
+ throw new IOException(e.getMessage(), e);
+ }
+ }
+
+ private CompletableFuture<?> write(
+ TableWriter writer, FlussOperationType opType, InternalRow row, TablePath tablePath)
+ throws IOException {
+ if (writer instanceof UpsertWriter) {
+ UpsertWriter upsertWriter = (UpsertWriter) writer;
+ if (opType == FlussOperationType.UPSERT) {
+ return upsertWriter.upsert(row);
+ } else if (opType == FlussOperationType.DELETE) {
+ return upsertWriter.delete(row);
+ } else {
+ throw new UnsupportedOperationException(
+ String.format(
+ "Unsupported operation type: %s for primary key table %s",
+ opType, tablePath));
+ }
+ } else if (writer instanceof AppendWriter) {
+ AppendWriter appendWriter = (AppendWriter) writer;
+ if (opType == FlussOperationType.APPEND) {
+ return appendWriter.append(row);
+ } else {
+ throw new UnsupportedOperationException(
+ String.format(
+ "Unsupported operation type: %s for log table %s",
+ opType, tablePath));
+ }
+ } else {
+ throw new UnsupportedOperationException(
+ String.format(
+ "Unsupported writer type: %s for table %s",
+ writer.getClass(), tablePath));
+ }
+ }
+
+ public void flush(boolean endOfInput) throws IOException {
+ for (TableWriter writer : writerMap.values()) {
+ writer.flush();
+ checkAsyncException();
+ }
+ }
+
+ @Override
+ public void close() throws Exception {
+ LOG.info("Closing Fluss sink function.");
+ try {
+ for (Table table : tableMap.values()) {
+ table.close();
+ }
+
+ tableMap.clear();
+
+ if (connection != null) {
+ connection.close();
+ }
+ } catch (Exception e) {
+ LOG.warn("Exception occurs while closing Fluss Connection.", e);
+ }
+ connection = null;
+
+ if (flinkMetricRegistry != null) {
+ flinkMetricRegistry.close();
+ }
+ flinkMetricRegistry = null;
+
+ // Rethrow exception for the case in which close is called before writer() and flush().
+ checkAsyncException();
+
+ LOG.info("Finished closing Fluss sink function.");
+ }
+
+ private long computeSendTime() {
+ if (flinkMetricRegistry == null) {
+ return -1;
+ }
+
+ Metric writerSendLatencyMs =
+ flinkMetricRegistry.getFlussMetric(MetricNames.WRITER_SEND_LATENCY_MS);
+ if (writerSendLatencyMs == null) {
+ return -1;
+ }
+
+ return ((Gauge<Long>) writerSendLatencyMs).getValue();
+ }
+
+ /**
+ * This method should only be invoked in the mailbox thread since the counter is not volatile.
+ * Logic needs to be invoked by write AND flush since we support various semantics.
+ */
+ protected void checkAsyncException() throws IOException {
+ // reset this exception since we could close the writer later on
+ Throwable throwable = asyncWriterException;
+ if (throwable != null) {
+ asyncWriterException = null;
+ numRecordsOutErrorsCounter.inc();
+ LOG.error("Exception occurs while write row to fluss.", throwable);
+ throw new IOException(
+ "One or more Fluss Writer send requests have encountered exception", throwable);
+ }
+ }
+}
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/v2/metrics/WarppedFlussCounter.java b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-fluss/src/main/java/org/apache/flink/cdc/connectors/fluss/sink/v2/metrics/WarppedFlussCounter.java
new file mode 100644
index 000000000..8f3c1e3d6
--- /dev/null
+++ b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-fluss/src/main/java/org/apache/flink/cdc/connectors/fluss/sink/v2/metrics/WarppedFlussCounter.java
@@ -0,0 +1,59 @@
+/*
+ * 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.connectors.fluss.sink.v2.metrics;
+
+import org.apache.flink.metrics.Counter;
+
+/* This file is based on source code of Apache Fluss Project (https://fluss.apache.org/), licensed by the Apache
+ * Software Foundation (ASF) under the Apache License, Version 2.0. See the NOTICE file distributed with this work for
+ * additional information regarding copyright ownership. */
+
+/** An implementation of Flink's {@link Counter} which wraps Fluss's Counter. */
+public class WarppedFlussCounter implements Counter {
+
+ private final com.alibaba.fluss.metrics.Counter flussCounter;
+
+ public WarppedFlussCounter(com.alibaba.fluss.metrics.Counter flussCounter) {
+ this.flussCounter = flussCounter;
+ }
+
+ @Override
+ public void inc() {
+ flussCounter.inc();
+ }
+
+ @Override
+ public void inc(long n) {
+ flussCounter.inc(n);
+ }
+
+ @Override
+ public void dec() {
+ flussCounter.dec();
+ }
+
+ @Override
+ public void dec(long n) {
+ flussCounter.dec(n);
+ }
+
+ @Override
+ public long getCount() {
+ return flussCounter.getCount();
+ }
+}
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/v2/metrics/WrappedFlussGauge.java b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-fluss/src/main/java/org/apache/flink/cdc/connectors/fluss/sink/v2/metrics/WrappedFlussGauge.java
new file mode 100644
index 000000000..1b5df21d0
--- /dev/null
+++ b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-fluss/src/main/java/org/apache/flink/cdc/connectors/fluss/sink/v2/metrics/WrappedFlussGauge.java
@@ -0,0 +1,39 @@
+/*
+ * 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.connectors.fluss.sink.v2.metrics;
+
+import org.apache.flink.metrics.Gauge;
+
+/* This file is based on source code of Apache Fluss Project (https://fluss.apache.org/), licensed by the Apache
+ * Software Foundation (ASF) under the Apache License, Version 2.0. See the NOTICE file distributed with this work for
+ * additional information regarding copyright ownership. */
+
+/** An implementation of Flink's {@link Gauge} which wraps Fluss's Gauge. */
+public class WrappedFlussGauge<T> implements Gauge<T> {
+
+ private final com.alibaba.fluss.metrics.Gauge<T> flussGauge;
+
+ public WrappedFlussGauge(com.alibaba.fluss.metrics.Gauge<T> flussGauge) {
+ this.flussGauge = flussGauge;
+ }
+
+ @Override
+ public T getValue() {
+ return flussGauge.getValue();
+ }
+}
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/v2/metrics/WrapperFlussHistogram.java b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-fluss/src/main/java/org/apache/flink/cdc/connectors/fluss/sink/v2/metrics/WrapperFlussHistogram.java
new file mode 100644
index 000000000..a6ce22838
--- /dev/null
+++ b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-fluss/src/main/java/org/apache/flink/cdc/connectors/fluss/sink/v2/metrics/WrapperFlussHistogram.java
@@ -0,0 +1,98 @@
+/*
+ * 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.connectors.fluss.sink.v2.metrics;
+
+import org.apache.flink.metrics.Histogram;
+import org.apache.flink.metrics.HistogramStatistics;
+
+/* This file is based on source code of Apache Fluss Project (https://fluss.apache.org/), licensed by the Apache
+ * Software Foundation (ASF) under the Apache License, Version 2.0. See the NOTICE file distributed with this work for
+ * additional information regarding copyright ownership. */
+
+/** An implementation of Flink's {@link Histogram} which wraps Fluss's Histogram. */
+public class WrapperFlussHistogram implements Histogram {
+
+ private final com.alibaba.fluss.metrics.Histogram flussHistogram;
+
+ public WrapperFlussHistogram(com.alibaba.fluss.metrics.Histogram flussHistogram) {
+ this.flussHistogram = flussHistogram;
+ }
+
+ @Override
+ public void update(long n) {
+ flussHistogram.update(n);
+ }
+
+ @Override
+ public long getCount() {
+ return flussHistogram.getCount();
+ }
+
+ @Override
+ public HistogramStatistics getStatistics() {
+
+ flussHistogram.getStatistics();
+
+ return null;
+ }
+
+ private static class FlinkHistogramStatistics extends HistogramStatistics {
+
+ private final com.alibaba.fluss.metrics.HistogramStatistics flussHistogramStatistics;
+
+ public FlinkHistogramStatistics(
+ com.alibaba.fluss.metrics.HistogramStatistics flussHistogramStatistics) {
+ this.flussHistogramStatistics = flussHistogramStatistics;
+ }
+
+ @Override
+ public double getQuantile(double quantile) {
+ return flussHistogramStatistics.getQuantile(quantile);
+ }
+
+ @Override
+ public long[] getValues() {
+ return flussHistogramStatistics.getValues();
+ }
+
+ @Override
+ public int size() {
+ return flussHistogramStatistics.size();
+ }
+
+ @Override
+ public double getMean() {
+ return flussHistogramStatistics.getMean();
+ }
+
+ @Override
+ public double getStdDev() {
+ return flussHistogramStatistics.getStdDev();
+ }
+
+ @Override
+ public long getMax() {
+ return flussHistogramStatistics.getMax();
+ }
+
+ @Override
+ public long getMin() {
+ return flussHistogramStatistics.getMin();
+ }
+ }
+}
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/v2/metrics/WrapperFlussMeter.java b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-fluss/src/main/java/org/apache/flink/cdc/connectors/fluss/sink/v2/metrics/WrapperFlussMeter.java
new file mode 100644
index 000000000..40bac46a8
--- /dev/null
+++ b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-fluss/src/main/java/org/apache/flink/cdc/connectors/fluss/sink/v2/metrics/WrapperFlussMeter.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.connectors.fluss.sink.v2.metrics;
+
+import org.apache.flink.metrics.Meter;
+
+/* This file is based on source code of Apache Fluss Project (https://fluss.apache.org/), licensed by the Apache
+ * Software Foundation (ASF) under the Apache License, Version 2.0. See the NOTICE file distributed with this work for
+ * additional information regarding copyright ownership. */
+
+/** An implementation of Flink's {@link Meter} which wraps Fluss's Meter. */
+public class WrapperFlussMeter implements Meter {
+
+ private final com.alibaba.fluss.metrics.Meter flussMeter;
+
+ public WrapperFlussMeter(com.alibaba.fluss.metrics.Meter flussMeter) {
+ this.flussMeter = flussMeter;
+ }
+
+ @Override
+ public void markEvent() {
+ flussMeter.markEvent();
+ }
+
+ @Override
+ public void markEvent(long n) {
+ flussMeter.markEvent(n);
+ }
+
+ @Override
+ public double getRate() {
+ return flussMeter.getRate();
+ }
+
+ @Override
+ public long getCount() {
+ return flussMeter.getCount();
+ }
+}
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/v2/metrics/WrapperFlussMetricRegistry.java b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-fluss/src/main/java/org/apache/flink/cdc/connectors/fluss/sink/v2/metrics/WrapperFlussMetricRegistry.java
new file mode 100644
index 000000000..6b33202ef
--- /dev/null
+++ b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-fluss/src/main/java/org/apache/flink/cdc/connectors/fluss/sink/v2/metrics/WrapperFlussMetricRegistry.java
@@ -0,0 +1,151 @@
+/*
+ * 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.connectors.fluss.sink.v2.metrics;
+
+import org.apache.flink.metrics.MetricGroup;
+
+import com.alibaba.fluss.metrics.CharacterFilter;
+import com.alibaba.fluss.metrics.Counter;
+import com.alibaba.fluss.metrics.Gauge;
+import com.alibaba.fluss.metrics.Histogram;
+import com.alibaba.fluss.metrics.Meter;
+import com.alibaba.fluss.metrics.Metric;
+import com.alibaba.fluss.metrics.groups.AbstractMetricGroup;
+import com.alibaba.fluss.metrics.registry.MetricRegistry;
+
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.LinkedHashMap;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.CompletableFuture;
+
+/* This file is based on source code of Apache Fluss Project (https://fluss.apache.org/), licensed by the Apache
+ * Software Foundation (ASF) under the Apache License, Version 2.0. See the NOTICE file distributed with this work for
+ * additional information regarding copyright ownership. */
+
+/**
+ * An implementation of {@link MetricRegistry} which registers all metrics into Flink's metric
+ * system. It's mainly used for Fluss client to register its metrics to Flink's metric system.
+ *
+ * <p>All metrics of the Fluss client are registered under group "fluss", which is a child group of
+ * {@link org.apache.flink.metrics.groups.OperatorMetricGroup}.
+ *
+ * <p>For example, the metrics sendLatencyMs will be reported in metric:
+ * "{some_parent_groups}.operator.fluss.client_writer.client_id.sendLatencyMs"
+ */
+public class WrapperFlussMetricRegistry implements MetricRegistry {
+
+ public static final String FLUSS_GROUP_NAME = "fluss";
+ private static final Character FIELD_DELIMITER = '_';
+
+ private final MetricGroup metricGroupForFluss;
+ private final Map<String, Metric> metrics;
+ private final Set<String> exposedMetricNames;
+
+ public WrapperFlussMetricRegistry(
+ MetricGroup flinkOperatorMetricGroup, Set<String> exposedMetricNames) {
+ this.metricGroupForFluss = flinkOperatorMetricGroup.addGroup(FLUSS_GROUP_NAME);
+ this.metrics = new HashMap<>();
+ this.exposedMetricNames = exposedMetricNames;
+ }
+
+ @Override
+ public int getNumberReporters() {
+ // only flink as the reporter
+ return 1;
+ }
+
+ @Override
+ public void register(Metric metric, String metricName, AbstractMetricGroup group) {
+ // use the logical group name from Fluss's metrics group as the group name
+ // of Flink's metric group the metrics will be registered in
+ String logicalGroupName =
+ group.getLogicalScope(CharacterFilter.NO_OP_FILTER, FIELD_DELIMITER);
+ MetricGroup currentMetricGroup = metricGroupForFluss.addGroup(logicalGroupName);
+ // we need to put all the variables of the Fluss's metrics group to Flink's metrics group
+ for (Map.Entry<String, String> variablesEntry : getVariables(group).entrySet()) {
+ currentMetricGroup =
+ currentMetricGroup.addGroup(variablesEntry.getKey(), variablesEntry.getValue());
+ }
+
+ // now, register to the Flink's metrics group
+ registerMetric(currentMetricGroup, metric, metricName);
+ }
+
+ /** Exposes the metrics of Fluss metics group for flink. */
+ public Metric getFlussMetric(String metricName) {
+ return metrics.get(metricName);
+ }
+
+ /**
+ * Get all the variables of the group. It'll get the variables of the group from parent to
+ * children to keep the orders of variables.
+ *
+ * @return the orders map of the variables
+ */
+ private Map<String, String> getVariables(AbstractMetricGroup group) {
+ if (group == null) {
+ return Collections.emptyMap();
+ }
+ Map<String, String> variablesMap = new LinkedHashMap<>(getVariables(group.getParent()));
+ for (Map.Entry<String, String> variablesEntry : group.getAllVariables().entrySet()) {
+ // only if it hasn't contains the variable, put the variable to the map
+ if (!variablesMap.containsKey(variablesEntry.getKey())) {
+ variablesMap.put(variablesEntry.getKey(), variablesEntry.getValue());
+ }
+ }
+ return variablesMap;
+ }
+
+ private void registerMetric(MetricGroup metricGroup, Metric metric, String metricName) {
+ switch (metric.getMetricType()) {
+ case COUNTER:
+ metricGroup.counter(metricName, new WarppedFlussCounter((Counter) metric));
+ break;
+ case METER:
+ metricGroup.meter(metricName, new WrapperFlussMeter((Meter) metric));
+ break;
+ case GAUGE:
+ metricGroup.gauge(metricName, new WrappedFlussGauge<>((Gauge<?>) metric));
+ break;
+ case HISTOGRAM:
+ metricGroup.histogram(metricName, new WrapperFlussHistogram((Histogram) metric));
+ break;
+ }
+
+ if (exposedMetricNames.contains(metricName)) {
+ metrics.put(metricName, metric);
+ }
+ }
+
+ @Override
+ public void unregister(Metric metric, String metricName, AbstractMetricGroup group) {
+ // do nothing since the metric is actually registered into Flink's metric system,
+ // it's fine to not unregister in here now since when Fluss's writer/scanner needs to
+ // unregister metrics, it means the writer/scanner needs to be closed along with closing
+ // operator
+ }
+
+ @Override
+ public CompletableFuture<Void> closeAsync() {
+ ((org.apache.flink.runtime.metrics.groups.AbstractMetricGroup<?>) metricGroupForFluss)
+ .close();
+ return CompletableFuture.completedFuture(null);
+ }
+}
diff --git a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-fluss/src/main/java/org/apache/flink/cdc/connectors/fluss/utils/FlussConversions.java b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-fluss/src/main/java/org/apache/flink/cdc/connectors/fluss/utils/FlussConversions.java
new file mode 100644
index 000000000..eb311dd0e
--- /dev/null
+++ b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-fluss/src/main/java/org/apache/flink/cdc/connectors/fluss/utils/FlussConversions.java
@@ -0,0 +1,255 @@
+/*
+ * 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.connectors.fluss.utils;
+
+import org.apache.flink.cdc.common.types.ArrayType;
+import org.apache.flink.cdc.common.types.BigIntType;
+import org.apache.flink.cdc.common.types.BinaryType;
+import org.apache.flink.cdc.common.types.BooleanType;
+import org.apache.flink.cdc.common.types.CharType;
+import org.apache.flink.cdc.common.types.DateType;
+import org.apache.flink.cdc.common.types.DecimalType;
+import org.apache.flink.cdc.common.types.DoubleType;
+import org.apache.flink.cdc.common.types.FloatType;
+import org.apache.flink.cdc.common.types.IntType;
+import org.apache.flink.cdc.common.types.LocalZonedTimestampType;
+import org.apache.flink.cdc.common.types.MapType;
+import org.apache.flink.cdc.common.types.RowType;
+import org.apache.flink.cdc.common.types.SmallIntType;
+import org.apache.flink.cdc.common.types.TimeType;
+import org.apache.flink.cdc.common.types.TimestampType;
+import org.apache.flink.cdc.common.types.TinyIntType;
+import org.apache.flink.cdc.common.types.VarBinaryType;
+import org.apache.flink.cdc.common.types.VarCharType;
+import org.apache.flink.cdc.common.types.ZonedTimestampType;
+import org.apache.flink.util.CollectionUtil;
+
+import com.alibaba.fluss.annotation.VisibleForTesting;
+import com.alibaba.fluss.metadata.Schema;
+import com.alibaba.fluss.metadata.TableDescriptor;
+
+import javax.annotation.Nullable;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
+
+/** Converter from Flink's type to Fluss's type. */
+public class FlussConversions {
+ private static final CdcTypeToFlussType TO_FLUSS_TYPE_INSTANCE = new CdcTypeToFlussType();
+
+ public static TableDescriptor toFlussTable(
+ org.apache.flink.cdc.common.schema.Schema cdcSchema,
+ List<String> bucketKeys,
+ @Nullable Integer bucketNum,
+ Map<String, String> tableProperties) {
+ // first build schema with cdc columns
+ Schema schema = toFlussSchema(cdcSchema);
+
+ if (CollectionUtil.isNullOrEmpty(bucketKeys)) {
+ // use primary keys - partition keys
+ bucketKeys =
+ schema.getPrimaryKey()
+ .map(
+ pk -> {
+ List<String> keys = new ArrayList<>(pk.getColumnNames());
+ keys.removeAll(cdcSchema.partitionKeys());
+ return keys;
+ })
+ .orElse(Collections.emptyList());
+ }
+
+ return TableDescriptor.builder()
+ .schema(schema)
+ .partitionedBy(cdcSchema.partitionKeys())
+ .distributedBy(bucketNum, bucketKeys)
+ .comment(cdcSchema.comment())
+ .properties(tableProperties)
+ .build();
+ }
+
+ public static com.alibaba.fluss.metadata.Schema toFlussSchema(
+ org.apache.flink.cdc.common.schema.Schema cdcSchema) {
+ Schema.Builder schemBuilder = Schema.newBuilder();
+ if (!CollectionUtil.isNullOrEmpty(cdcSchema.primaryKeys())) {
+ schemBuilder.primaryKey(cdcSchema.primaryKeys());
+ }
+
+ Schema schema =
+ schemBuilder
+ .fromColumns(
+ cdcSchema.getColumns().stream()
+ .map(
+ column ->
+ new Schema.Column(
+ column.getName(),
+ toFlussType(column.getType()),
+ column.getComment()))
+ .collect(Collectors.toList()))
+ .build();
+ return schema;
+ }
+
+ @VisibleForTesting
+ private static com.alibaba.fluss.types.DataType toFlussType(
+ org.apache.flink.cdc.common.types.DataType flinkDataType) {
+ return flinkDataType.accept(TO_FLUSS_TYPE_INSTANCE);
+ }
+
+ public static Boolean sameCdcColumnsIgnoreCommentAndDefaultValue(
+ org.apache.flink.cdc.common.schema.Schema oldSchema,
+ org.apache.flink.cdc.common.schema.Schema newSchema) {
+ List<org.apache.flink.cdc.common.schema.Column> upstreamColumns = oldSchema.getColumns();
+ List<org.apache.flink.cdc.common.schema.Column> physicalColumns = newSchema.getColumns();
+ if (upstreamColumns.size() != physicalColumns.size()) {
+ return false;
+ }
+ for (int i = 0; i < physicalColumns.size(); i++) {
+ org.apache.flink.cdc.common.schema.Column upstreamColumn = upstreamColumns.get(i);
+ org.apache.flink.cdc.common.schema.Column physicalColumn = physicalColumns.get(i);
+ // Case sensitive.
+ if (!upstreamColumn.getName().equals(physicalColumn.getName())) {
+ return false;
+ }
+ if (!upstreamColumn.getType().equals(physicalColumn.getType())) {
+ return false;
+ }
+ }
+ return true;
+ }
+
+ private static class CdcTypeToFlussType
+ implements org.apache.flink.cdc.common.types.DataTypeVisitor<
+ com.alibaba.fluss.types.DataType> {
+
+ @Override
+ public com.alibaba.fluss.types.DataType visit(CharType charType) {
+ return new com.alibaba.fluss.types.CharType(
+ charType.isNullable(), charType.getLength());
+ }
+
+ @Override
+ public com.alibaba.fluss.types.DataType visit(VarCharType varCharType) {
+ // fluss not support varchar type
+ return new com.alibaba.fluss.types.StringType(varCharType.isNullable());
+ }
+
+ @Override
+ public com.alibaba.fluss.types.DataType visit(BooleanType booleanType) {
+ return new com.alibaba.fluss.types.BooleanType(booleanType.isNullable());
+ }
+
+ @Override
+ public com.alibaba.fluss.types.DataType visit(BinaryType binaryType) {
+ return new com.alibaba.fluss.types.BinaryType(
+ binaryType.isNullable(), binaryType.getLength());
+ }
+
+ @Override
+ public com.alibaba.fluss.types.DataType visit(VarBinaryType varBinaryType) {
+ // fluss not support varbinary type
+ return new com.alibaba.fluss.types.BytesType(varBinaryType.isNullable());
+ }
+
+ @Override
+ public com.alibaba.fluss.types.DataType visit(DecimalType decimalType) {
+ return new com.alibaba.fluss.types.DecimalType(
+ decimalType.isNullable(), decimalType.getPrecision(), decimalType.getScale());
+ }
+
+ @Override
+ public com.alibaba.fluss.types.DataType visit(TinyIntType tinyIntType) {
+ return new com.alibaba.fluss.types.TinyIntType(tinyIntType.isNullable());
+ }
+
+ @Override
+ public com.alibaba.fluss.types.DataType visit(SmallIntType smallIntType) {
+ return new com.alibaba.fluss.types.SmallIntType(smallIntType.isNullable());
+ }
+
+ @Override
+ public com.alibaba.fluss.types.DataType visit(IntType intType) {
+ return new com.alibaba.fluss.types.IntType(intType.isNullable());
+ }
+
+ @Override
+ public com.alibaba.fluss.types.DataType visit(BigIntType bigIntType) {
+ return new com.alibaba.fluss.types.BigIntType(bigIntType.isNullable());
+ }
+
+ @Override
+ public com.alibaba.fluss.types.DataType visit(FloatType floatType) {
+ return new com.alibaba.fluss.types.FloatType(floatType.isNullable());
+ }
+
+ @Override
+ public com.alibaba.fluss.types.DataType visit(DoubleType doubleType) {
+ return new com.alibaba.fluss.types.DoubleType(doubleType.isNullable());
+ }
+
+ @Override
+ public com.alibaba.fluss.types.DataType visit(DateType dateType) {
+ return new com.alibaba.fluss.types.DateType(dateType.isNullable());
+ }
+
+ @Override
+ public com.alibaba.fluss.types.DataType visit(TimeType timeType) {
+ return new com.alibaba.fluss.types.TimeType(
+ timeType.isNullable(), timeType.getPrecision());
+ }
+
+ @Override
+ public com.alibaba.fluss.types.DataType visit(TimestampType timestampType) {
+ return new com.alibaba.fluss.types.TimestampType(
+ timestampType.isNullable(), timestampType.getPrecision());
+ }
+
+ @Override
+ public com.alibaba.fluss.types.DataType visit(ZonedTimestampType zonedTimestampType) {
+ throw new UnsupportedOperationException(
+ "Unsupported data type in fluss " + zonedTimestampType);
+ }
+
+ @Override
+ public com.alibaba.fluss.types.DataType visit(
+ LocalZonedTimestampType localZonedTimestampType) {
+ return new com.alibaba.fluss.types.LocalZonedTimestampType(
+ localZonedTimestampType.isNullable(), localZonedTimestampType.getPrecision());
+ }
+
+ @Override
+ public com.alibaba.fluss.types.DataType visit(ArrayType arrayType) {
+ throw new UnsupportedOperationException(
+ "Unsupported data type in fluss version under 0.7: " + arrayType);
+ }
+
+ @Override
+ public com.alibaba.fluss.types.DataType visit(MapType mapType) {
+ throw new UnsupportedOperationException(
+ "Unsupported data type in fluss version under 0.7: " + mapType);
+ }
+
+ @Override
+ public com.alibaba.fluss.types.DataType visit(RowType rowType) {
+ throw new UnsupportedOperationException(
+ "Unsupported data type in fluss version under 0.7: " + rowType);
+ }
+ }
+}
diff --git a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-fluss/src/main/resources/META-INF/services/org.apache.flink.cdc.common.factories.Factory b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-fluss/src/main/resources/META-INF/services/org.apache.flink.cdc.common.factories.Factory
new file mode 100644
index 000000000..204b6f59f
--- /dev/null
+++ b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-fluss/src/main/resources/META-INF/services/org.apache.flink.cdc.common.factories.Factory
@@ -0,0 +1,16 @@
+# 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.
+
+org.apache.flink.cdc.connectors.fluss.factory.FlussDataSinkFactory
diff --git a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-fluss/src/test/java/org/apache/flink/cdc/connectors/fluss/FlussPipelineITCase.java b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-fluss/src/test/java/org/apache/flink/cdc/connectors/fluss/FlussPipelineITCase.java
new file mode 100644
index 000000000..a625f0fd9
--- /dev/null
+++ b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-fluss/src/test/java/org/apache/flink/cdc/connectors/fluss/FlussPipelineITCase.java
@@ -0,0 +1,576 @@
+/*
+ * 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.connectors.fluss;
+
+import org.apache.flink.cdc.common.configuration.Configuration;
+import org.apache.flink.cdc.common.data.binary.BinaryStringData;
+import org.apache.flink.cdc.common.event.CreateTableEvent;
+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.pipeline.PipelineOptions;
+import org.apache.flink.cdc.common.pipeline.SchemaChangeBehavior;
+import org.apache.flink.cdc.common.schema.Schema;
+import org.apache.flink.cdc.common.types.DataTypes;
+import org.apache.flink.cdc.common.types.RowType;
+import org.apache.flink.cdc.composer.PipelineExecution;
+import org.apache.flink.cdc.composer.definition.PipelineDef;
+import org.apache.flink.cdc.composer.definition.SinkDef;
+import org.apache.flink.cdc.composer.definition.SourceDef;
+import org.apache.flink.cdc.composer.flink.FlinkPipelineComposer;
+import org.apache.flink.cdc.connectors.values.factory.ValuesDataFactory;
+import org.apache.flink.cdc.connectors.values.source.ValuesDataSourceHelper;
+import org.apache.flink.cdc.connectors.values.source.ValuesDataSourceOptions;
+import org.apache.flink.cdc.runtime.typeutils.BinaryRecordDataGenerator;
+import org.apache.flink.runtime.testutils.MiniClusterResourceConfiguration;
+import org.apache.flink.table.api.EnvironmentSettings;
+import org.apache.flink.table.api.TableEnvironment;
+import org.apache.flink.table.api.config.ExecutionConfigOptions;
+import org.apache.flink.test.junit5.MiniClusterExtension;
+import org.apache.flink.types.Row;
+import org.apache.flink.util.CloseableIterator;
+
+import com.alibaba.fluss.config.ConfigOptions;
+import com.alibaba.fluss.config.MemorySize;
+import com.alibaba.fluss.metadata.DataLakeFormat;
+import com.alibaba.fluss.server.testutils.FlussClusterExtension;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.RegisterExtension;
+
+import java.time.Duration;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import static com.alibaba.fluss.config.ConfigOptions.BOOTSTRAP_SERVERS;
+import static com.alibaba.fluss.flink.source.testutils.FlinkRowAssertionsUtils.assertResultsIgnoreOrder;
+import static com.alibaba.fluss.server.testutils.FlussClusterExtension.BUILTIN_DATABASE;
+import static org.apache.flink.cdc.connectors.values.source.ValuesDataSourceHelper.TABLE_1;
+import static org.apache.flink.cdc.connectors.values.source.ValuesDataSourceHelper.TABLE_2;
+import static org.apache.flink.configuration.CoreOptions.ALWAYS_PARENT_FIRST_LOADER_PATTERNS_ADDITIONAL;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/** ITCase for Fluss Pipeline. */
+public class FlussPipelineITCase {
+ private static final int MAX_PARALLELISM = 4;
+
+ // Always use parent-first classloader for CDC classes.
+ // The reason is that ValuesDatabase uses static field for holding data, we need to make sure
+ // the class is loaded by AppClassloader so that we can verify data in the test case.
+ private static final org.apache.flink.configuration.Configuration MINI_CLUSTER_CONFIG =
+ new org.apache.flink.configuration.Configuration();
+
+ /**
+ * Use {@link MiniClusterExtension} to reduce the overhead of restarting the MiniCluster for
+ * every test case.
+ */
+ @RegisterExtension
+ static final MiniClusterExtension MINI_CLUSTER_RESOURCE =
+ new MiniClusterExtension(
+ new MiniClusterResourceConfiguration.Builder()
+ .setNumberTaskManagers(1)
+ .setNumberSlotsPerTaskManager(MAX_PARALLELISM)
+ .build());
+
+ /**
+ * Use {@link FlussClusterExtension} to start a Fluss cluster with sasl authentication for every
+ * test case.
+ */
+ @RegisterExtension
+ public static final FlussClusterExtension FLUSS_CLUSTER_EXTENSION =
+ FlussClusterExtension.builder()
+ .setClusterConf(initConfig())
+ .setCoordinatorServerListeners("FLUSS://localhost:0, CLIENT://localhost:0")
+ .setTabletServerListeners("FLUSS://localhost:0, CLIENT://localhost:0")
+ .setNumOfTabletServers(3)
+ .build();
+
+ static {
+ MINI_CLUSTER_CONFIG.set(
+ ALWAYS_PARENT_FIRST_LOADER_PATTERNS_ADDITIONAL,
+ Collections.singletonList("org.apache.flink.cdc"));
+ }
+
+ static final String CATALOG_NAME = "test_catalog";
+ static final String DEFAULT_DB = "default_schema";
+
+ protected TableEnvironment tBatchEnv;
+
+ @BeforeEach
+ void before() {
+ // open a catalog so that we can get table from the catalog
+ String bootstrapServers = FLUSS_CLUSTER_EXTENSION.getBootstrapServers();
+
+ // create batch table environment
+ tBatchEnv =
+ TableEnvironment.create(EnvironmentSettings.newInstance().inBatchMode().build());
+ tBatchEnv.executeSql(
+ String.format(
+ "create catalog %s with ('type' = 'fluss', '%s' = '%s')",
+ CATALOG_NAME, BOOTSTRAP_SERVERS.key(), bootstrapServers));
+ tBatchEnv.executeSql("use catalog " + CATALOG_NAME);
+ tBatchEnv
+ .getConfig()
+ .set(ExecutionConfigOptions.TABLE_EXEC_RESOURCE_DEFAULT_PARALLELISM, 2);
+ // create database
+ tBatchEnv.executeSql("create database " + DEFAULT_DB);
+ tBatchEnv.useDatabase(DEFAULT_DB);
+ }
+
+ @AfterEach
+ void after() {
+ tBatchEnv.useDatabase(BUILTIN_DATABASE);
+ tBatchEnv.executeSql(String.format("drop database %s cascade", DEFAULT_DB));
+ }
+
+ @Test
+ void testSinglePrimaryTable() throws Exception {
+ List<List<Event>> eventOfSplits = new ArrayList<>();
+ List<Event> split1 = new ArrayList<>();
+ Schema schema =
+ Schema.newBuilder()
+ .physicalColumn("col1", DataTypes.STRING())
+ .physicalColumn("col2", DataTypes.STRING())
+ .primaryKey("col1")
+ .build();
+ CreateTableEvent createTableEvent = new CreateTableEvent(TABLE_1, schema);
+ split1.add(createTableEvent);
+ BinaryRecordDataGenerator generator =
+ new BinaryRecordDataGenerator(RowType.of(DataTypes.STRING(), DataTypes.STRING()));
+ DataChangeEvent insertEvent1 =
+ DataChangeEvent.insertEvent(
+ TABLE_1,
+ generator.generate(
+ new Object[] {
+ BinaryStringData.fromString("1"),
+ BinaryStringData.fromString("a")
+ }));
+ split1.add(insertEvent1);
+ DataChangeEvent insertEvent2 =
+ DataChangeEvent.insertEvent(
+ TABLE_1,
+ generator.generate(
+ new Object[] {
+ BinaryStringData.fromString("2"),
+ BinaryStringData.fromString("b")
+ }));
+ split1.add(insertEvent2);
+ DataChangeEvent insertEvent3 =
+ DataChangeEvent.insertEvent(
+ TABLE_1,
+ generator.generate(
+ new Object[] {
+ BinaryStringData.fromString("3"),
+ BinaryStringData.fromString("c")
+ }));
+ split1.add(insertEvent3);
+ eventOfSplits.add(split1);
+ DataChangeEvent updateEvent =
+ DataChangeEvent.updateEvent(
+ TABLE_1,
+ generator.generate(
+ new Object[] {
+ BinaryStringData.fromString("2"),
+ BinaryStringData.fromString("b")
+ }),
+ generator.generate(
+ new Object[] {
+ BinaryStringData.fromString("2"),
+ BinaryStringData.fromString("b2")
+ }));
+ DataChangeEvent deleteEvent =
+ DataChangeEvent.deleteEvent(
+ TABLE_1,
+ generator.generate(
+ new Object[] {
+ BinaryStringData.fromString("1"),
+ BinaryStringData.fromString("a")
+ }));
+ eventOfSplits.add(Collections.singletonList(updateEvent));
+ eventOfSplits.add(Collections.singletonList(deleteEvent));
+
+ composeAndExecute(eventOfSplits);
+ checkResult(TABLE_1, Arrays.asList("+I[2, b2]", "+I[3, c]"));
+ }
+
+ @Test
+ void testSingleLogTable() throws Exception {
+ List<List<Event>> eventOfSplits = new ArrayList<>();
+ List<Event> split1 = new ArrayList<>();
+ Schema schema =
+ Schema.newBuilder()
+ .physicalColumn("col1", DataTypes.STRING())
+ .physicalColumn("col2", DataTypes.STRING())
+ .build();
+ CreateTableEvent createTableEvent = new CreateTableEvent(TABLE_1, schema);
+ split1.add(createTableEvent);
+ BinaryRecordDataGenerator generator =
+ new BinaryRecordDataGenerator(RowType.of(DataTypes.STRING(), DataTypes.STRING()));
+ DataChangeEvent insertEvent1 =
+ DataChangeEvent.insertEvent(
+ TABLE_1,
+ generator.generate(
+ new Object[] {
+ BinaryStringData.fromString("1"),
+ BinaryStringData.fromString("a")
+ }));
+ split1.add(insertEvent1);
+ DataChangeEvent insertEvent2 =
+ DataChangeEvent.insertEvent(
+ TABLE_1,
+ generator.generate(
+ new Object[] {
+ BinaryStringData.fromString("2"),
+ BinaryStringData.fromString("b")
+ }));
+ split1.add(insertEvent2);
+ DataChangeEvent insertEvent3 =
+ DataChangeEvent.insertEvent(
+ TABLE_1,
+ generator.generate(
+ new Object[] {
+ BinaryStringData.fromString("3"),
+ BinaryStringData.fromString("c")
+ }));
+ split1.add(insertEvent3);
+ eventOfSplits.add(split1);
+
+ composeAndExecute(eventOfSplits);
+ checkResult(TABLE_1, Arrays.asList("+I[1, a]", "+I[2, b]", "+I[3, c]"));
+ }
+
+ @Test
+ void testSingleLogTableWithSchemaChange() {
+ assertThatThrownBy(() -> composeAndExecute(ValuesDataSourceHelper.singleSplitSingleTable()))
+ .rootCause()
+ .hasMessageContaining(
+ "fluss metadata applier only support CreateTableEvent now but receives AddColumnEvent");
+ }
+
+ @Test
+ void testLackUsernameAndPassword() {
+ assertThatThrownBy(
+ () ->
+ composeAndExecute(
+ ValuesDataSourceHelper.singleSplitSingleTable(),
+ Collections.singletonMap(
+ BOOTSTRAP_SERVERS.key(), getBootstrapServers())))
+ .rootCause()
+ .hasMessageContaining(
+ "The connection has not completed authentication yet. This may be caused by a missing or incorrect configuration of 'client.security.protocol' on the client side.");
+ }
+
+ @Test
+ void testWrongTableOptions() {
+ Map<String, String> sinkOption = new HashMap<>();
+ sinkOption.put(BOOTSTRAP_SERVERS.key(), getBootstrapServers());
+ sinkOption.put(BOOTSTRAP_SERVERS.key(), getBootstrapServers());
+ sinkOption.put("properties.client.security.protocol", "sasl");
+ sinkOption.put("properties.client.security.sasl.mechanism", "PLAIN");
+ sinkOption.put("properties.client.security.sasl.username", "guest");
+ sinkOption.put("properties.client.security.sasl.password", "password2");
+ sinkOption.put("properties.table.non-key", "non-key-value");
+ assertThatThrownBy(
+ () ->
+ composeAndExecute(
+ ValuesDataSourceHelper.singleSplitSingleTable(),
+ sinkOption))
+ .rootCause()
+ .hasMessageContaining("'table.non-key' is not a Fluss table property");
+ }
+
+ @Test
+ void testMultiTables() throws Exception {
+ List<List<Event>> eventOfSplits = new ArrayList<>();
+ List<Event> split1 = new ArrayList<>();
+ BinaryRecordDataGenerator generator =
+ new BinaryRecordDataGenerator(RowType.of(DataTypes.STRING(), DataTypes.STRING()));
+
+ // create table
+ Schema schema =
+ Schema.newBuilder()
+ .physicalColumn("col1", DataTypes.STRING())
+ .physicalColumn("col2", DataTypes.STRING())
+ .primaryKey("col1")
+ .build();
+ CreateTableEvent createTableEvent = new CreateTableEvent(TABLE_1, schema);
+ split1.add(createTableEvent);
+ DataChangeEvent insertTable1Event1 =
+ DataChangeEvent.insertEvent(
+ TABLE_1,
+ generator.generate(
+ new Object[] {
+ BinaryStringData.fromString("1"),
+ BinaryStringData.fromString("1")
+ }));
+ split1.add(insertTable1Event1);
+ CreateTableEvent createTableEvent2 = new CreateTableEvent(TABLE_2, schema);
+ split1.add(createTableEvent2);
+ DataChangeEvent insertTable1Event2 =
+ DataChangeEvent.insertEvent(
+ TABLE_1,
+ generator.generate(
+ new Object[] {
+ BinaryStringData.fromString("2"),
+ BinaryStringData.fromString("2")
+ }));
+ split1.add(insertTable1Event2);
+ DataChangeEvent insertTable2Event1 =
+ DataChangeEvent.insertEvent(
+ TABLE_2,
+ generator.generate(
+ new Object[] {
+ BinaryStringData.fromString("1"),
+ BinaryStringData.fromString("1")
+ }));
+ split1.add(insertTable2Event1);
+ DataChangeEvent insertTable1Event3 =
+ DataChangeEvent.insertEvent(
+ TABLE_1,
+ generator.generate(
+ new Object[] {
+ BinaryStringData.fromString("3"),
+ BinaryStringData.fromString("3")
+ }));
+ split1.add(insertTable1Event3);
+
+ DataChangeEvent insertTable2Event2 =
+ DataChangeEvent.insertEvent(
+ TABLE_2,
+ generator.generate(
+ new Object[] {
+ BinaryStringData.fromString("2"),
+ BinaryStringData.fromString("2")
+ }));
+ split1.add(insertTable2Event2);
+ DataChangeEvent insertTabl2Event3 =
+ DataChangeEvent.insertEvent(
+ TABLE_2,
+ generator.generate(
+ new Object[] {
+ BinaryStringData.fromString("3"),
+ BinaryStringData.fromString("3")
+ }));
+ split1.add(insertTabl2Event3);
+ eventOfSplits.add(split1);
+
+ composeAndExecute(eventOfSplits);
+ checkResult(TABLE_1, Arrays.asList("+I[1, 1]", "+I[2, 2]", "+I[3, 3]"));
+ checkResult(TABLE_2, Arrays.asList("+I[1, 1]", "+I[2, 2]", "+I[3, 3]"));
+ }
+
+ @Test
+ void testInsertExistTableWithMoreColumns() throws Exception {
+ // create a fluss table with 2 columns but then cdc read a source table with 3 columns
+ tBatchEnv
+ .executeSql(
+ String.format(
+ "CREATE TABLE %s.%s (\n"
+ + " col2 STRING,\n"
+ + " col1 STRING,\n"
+ + " PRIMARY KEY (col1) NOT ENFORCED \n"
+ + ");",
+ TABLE_1.getSchemaName(), TABLE_1.getTableName()))
+ .await();
+
+ List<List<Event>> eventOfSplits = new ArrayList<>();
+ List<Event> split1 = new ArrayList<>();
+ Schema schema =
+ Schema.newBuilder()
+ .physicalColumn("col1", DataTypes.STRING())
+ .physicalColumn("col2", DataTypes.STRING())
+ .physicalColumn("col3", DataTypes.STRING())
+ .primaryKey("col1")
+ .build();
+ CreateTableEvent createTableEvent = new CreateTableEvent(TABLE_1, schema);
+ split1.add(createTableEvent);
+ BinaryRecordDataGenerator generator =
+ new BinaryRecordDataGenerator(
+ RowType.of(DataTypes.STRING(), DataTypes.STRING(), DataTypes.STRING()));
+ DataChangeEvent insertEvent1 =
+ DataChangeEvent.insertEvent(
+ TABLE_1,
+ generator.generate(
+ new Object[] {
+ BinaryStringData.fromString("1"),
+ BinaryStringData.fromString("a"),
+ BinaryStringData.fromString("1")
+ }));
+ split1.add(insertEvent1);
+ DataChangeEvent insertEvent2 =
+ DataChangeEvent.insertEvent(
+ TABLE_1,
+ generator.generate(
+ new Object[] {
+ BinaryStringData.fromString("2"),
+ BinaryStringData.fromString("2"),
+ BinaryStringData.fromString("2")
+ }));
+ split1.add(insertEvent2);
+ eventOfSplits.add(split1);
+
+ composeAndExecute(eventOfSplits);
+ checkResult(TABLE_1, Arrays.asList("+I[a, 1]", "+I[2, 2]"));
+ }
+
+ @Test
+ void testInsertExistTableWithLessColumns() throws Exception {
+ // create a fluss table with 2 columns but then cdc read a source table with 3 columns
+ tBatchEnv
+ .executeSql(
+ String.format(
+ "CREATE TABLE %s.%s (\n"
+ + " col1 STRING,\n"
+ + " col2 STRING,\n"
+ + " col3 STRING,\n"
+ + " PRIMARY KEY (col1) NOT ENFORCED \n"
+ + ");",
+ TABLE_1.getSchemaName(), TABLE_1.getTableName()))
+ .await();
+
+ List<List<Event>> eventOfSplits = new ArrayList<>();
+ List<Event> split1 = new ArrayList<>();
+ Schema schema =
+ Schema.newBuilder()
+ .physicalColumn("col1", DataTypes.STRING())
+ .physicalColumn("col2", DataTypes.STRING())
+ .primaryKey("col1")
+ .build();
+ CreateTableEvent createTableEvent = new CreateTableEvent(TABLE_1, schema);
+ split1.add(createTableEvent);
+ BinaryRecordDataGenerator generator =
+ new BinaryRecordDataGenerator(RowType.of(DataTypes.STRING(), DataTypes.STRING()));
+ DataChangeEvent insertEvent1 =
+ DataChangeEvent.insertEvent(
+ TABLE_1,
+ generator.generate(
+ new Object[] {
+ BinaryStringData.fromString("1"),
+ BinaryStringData.fromString("a")
+ }));
+ split1.add(insertEvent1);
+ DataChangeEvent insertEvent2 =
+ DataChangeEvent.insertEvent(
+ TABLE_1,
+ generator.generate(
+ new Object[] {
+ BinaryStringData.fromString("2"),
+ BinaryStringData.fromString("2")
+ }));
+ split1.add(insertEvent2);
+ eventOfSplits.add(split1);
+
+ composeAndExecute(eventOfSplits);
+ checkResult(TABLE_1, Arrays.asList("+I[1, a, null]", "+I[2, 2, null]"));
+ }
+
+ private void composeAndExecute(List<List<Event>> customSourceEvents) throws Exception {
+ Map<String, String> sinkOption = new HashMap<>();
+ sinkOption.put(BOOTSTRAP_SERVERS.key(), getBootstrapServers());
+ sinkOption.put("properties.client.security.protocol", "sasl");
+ sinkOption.put("properties.client.security.sasl.mechanism", "PLAIN");
+ sinkOption.put("properties.client.security.sasl.username", "guest");
+ sinkOption.put("properties.client.security.sasl.password", "password2");
+ composeAndExecute(customSourceEvents, sinkOption);
+ }
+
+ private void composeAndExecute(
+ List<List<Event>> customSourceEvents, Map<String, String> sinkOption) throws Exception {
+ FlinkPipelineComposer composer = FlinkPipelineComposer.ofMiniCluster();
+
+ // Setup value source
+ Configuration sourceConfig = new Configuration();
+ sourceConfig.set(
+ ValuesDataSourceOptions.EVENT_SET_ID,
+ ValuesDataSourceHelper.EventSetId.CUSTOM_SOURCE_EVENTS);
+ SourceDef sourceDef =
+ new SourceDef(ValuesDataFactory.IDENTIFIER, "Value Source", sourceConfig);
+ ValuesDataSourceHelper.setSourceEvents(customSourceEvents);
+
+ // Setup value sink
+ SinkDef sinkDef = new SinkDef("fluss", "Fluss Sink", Configuration.fromMap(sinkOption));
+
+ // Setup pipeline
+ Configuration pipelineConfig = new Configuration();
+ pipelineConfig.set(PipelineOptions.PIPELINE_PARALLELISM, 4);
+ pipelineConfig.set(
+ PipelineOptions.PIPELINE_SCHEMA_CHANGE_BEHAVIOR, SchemaChangeBehavior.EVOLVE);
+ PipelineDef pipelineDef =
+ new PipelineDef(
+ sourceDef,
+ sinkDef,
+ Collections.emptyList(),
+ Collections.emptyList(),
+ Collections.emptyList(),
+ pipelineConfig);
+
+ // Execute the pipeline
+ PipelineExecution execution = composer.compose(pipelineDef);
+ execution.execute();
+ }
+
+ private void checkResult(TableId tableId, List<String> expectedRows) {
+ CloseableIterator<Row> rowIter =
+ tBatchEnv
+ .executeSql(
+ String.format(
+ "select * from %s.%s limit %d",
+ tableId.getSchemaName(),
+ tableId.getTableName(),
+ expectedRows.size()))
+ .collect();
+ assertResultsIgnoreOrder(rowIter, expectedRows, true);
+ }
+
+ private static com.alibaba.fluss.config.Configuration initConfig() {
+ com.alibaba.fluss.config.Configuration conf = new com.alibaba.fluss.config.Configuration();
+ conf.setInt(ConfigOptions.DEFAULT_REPLICATION_FACTOR, 3);
+ // set a shorter interval for testing purpose
+ conf.set(ConfigOptions.KV_SNAPSHOT_INTERVAL, Duration.ofSeconds(1));
+ // set a shorter max lag time to make tests in FlussFailServerTableITCase faster
+ conf.set(ConfigOptions.LOG_REPLICA_MAX_LAG_TIME, Duration.ofSeconds(10));
+ // set default datalake format for the cluster and enable datalake tables
+ conf.set(ConfigOptions.DATALAKE_FORMAT, DataLakeFormat.PAIMON);
+
+ conf.set(ConfigOptions.CLIENT_WRITER_BUFFER_MEMORY_SIZE, MemorySize.parse("1mb"));
+ conf.set(ConfigOptions.CLIENT_WRITER_BATCH_SIZE, MemorySize.parse("1kb"));
+
+ // set security information.
+ conf.setString(ConfigOptions.SERVER_SECURITY_PROTOCOL_MAP.key(), "CLIENT:sasl");
+ conf.setString("security.sasl.enabled.mechanisms", "plain");
+ conf.setString(
+ "security.sasl.plain.jaas.config",
+ "com.alibaba.fluss.security.auth.sasl.plain.PlainLoginModule required "
+ + " user_root=\"password\" "
+ + " user_guest=\"password2\";");
+ return conf;
+ }
+
+ String getBootstrapServers() {
+ return String.join(
+ ",",
+ FLUSS_CLUSTER_EXTENSION
+ .getClientConfig("CLIENT")
+ .get(ConfigOptions.BOOTSTRAP_SERVERS));
+ }
+}
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
new file mode 100644
index 000000000..49ec01d30
--- /dev/null
+++ 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
@@ -0,0 +1,154 @@
+/*
+ * 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.connectors.fluss.factory;
+
+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.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.table.api.ValidationException;
+
+import org.apache.flink.shaded.guava31.com.google.common.collect.ImmutableMap;
+
+import org.assertj.core.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/** Test for {@link FlussDataSinkFactory}. */
+public class FlussDataSinkFactoryTest {
+ @Test
+ void testCreateDataSink() {
+ DataSinkFactory sinkFactory =
+ FactoryDiscoveryUtils.getFactoryByIdentifier("fluss", DataSinkFactory.class);
+ Assertions.assertThat(sinkFactory).isInstanceOf(FlussDataSinkFactory.class);
+
+ Configuration conf = createValidConfiguration();
+ DataSink dataSink =
+ sinkFactory.createDataSink(
+ new FactoryHelper.DefaultContext(
+ conf, conf, Thread.currentThread().getContextClassLoader()));
+ Assertions.assertThat(dataSink).isInstanceOf(FlussDataSink.class);
+ }
+
+ @Test
+ void testUnsupportedOption() {
+ DataSinkFactory sinkFactory =
+ FactoryDiscoveryUtils.getFactoryByIdentifier("fluss", DataSinkFactory.class);
+ Assertions.assertThat(sinkFactory).isInstanceOf(FlussDataSinkFactory.class);
+
+ Configuration conf =
+ Configuration.fromMap(
+ ImmutableMap.<String, String>builder()
+ .put(FlussDataSinkOptions.BOOTSTRAP_SERVERS.key(), "localhost:9123")
+ .put("unsupported_key", "unsupported_value")
+ .build());
+
+ Assertions.assertThatThrownBy(
+ () ->
+ sinkFactory.createDataSink(
+ new FactoryHelper.DefaultContext(
+ conf,
+ conf,
+ Thread.currentThread().getContextClassLoader())))
+ .isInstanceOf(ValidationException.class)
+ .hasMessageContaining(
+ "Unsupported options found for 'fluss'.\n\n"
+ + "Unsupported options:\n\n"
+ + "unsupported_key");
+ }
+
+ @Test
+ void testPrefixRequireOption() {
+ DataSinkFactory sinkFactory =
+ FactoryDiscoveryUtils.getFactoryByIdentifier("fluss", DataSinkFactory.class);
+ Assertions.assertThat(sinkFactory).isInstanceOf(FlussDataSinkFactory.class);
+
+ Configuration conf =
+ Configuration.fromMap(
+ ImmutableMap.<String, String>builder()
+ .put(FlussDataSinkOptions.BOOTSTRAP_SERVERS.key(), "localhost:9123")
+ .put("properties.table.option1", "value1")
+ .put("properties.client.option1", "value1")
+ .build());
+ DataSink dataSink =
+ sinkFactory.createDataSink(
+ new FactoryHelper.DefaultContext(
+ conf, conf, Thread.currentThread().getContextClassLoader()));
+ Assertions.assertThat(dataSink).isInstanceOf(FlussDataSink.class);
+ }
+
+ @Test
+ void testWrongBucketKeyAndBucketNum() {
+ DataSinkFactory sinkFactory =
+ FactoryDiscoveryUtils.getFactoryByIdentifier("fluss", DataSinkFactory.class);
+ Assertions.assertThat(sinkFactory).isInstanceOf(FlussDataSinkFactory.class);
+ assertThatThrownBy(
+ () ->
+ sinkFactory.createDataSink(
+ new FactoryHelper.DefaultContext(
+ Configuration.fromMap(
+ ImmutableMap.<String, String>builder()
+ .put(
+ FlussDataSinkOptions
+ .BOOTSTRAP_SERVERS
+ .key(),
+ "localhost:9123")
+ .put(
+ FlussDataSinkOptions
+ .BUCKET_KEY
+ .key(),
+ "database1.table1;a,b")
+ .build()),
+ new Configuration(),
+ Thread.currentThread().getContextClassLoader())))
+ .hasMessageContaining("Invalid bucket key configuration: database1.table1;a,b");
+ assertThatThrownBy(
+ () ->
+ sinkFactory.createDataSink(
+ new FactoryHelper.DefaultContext(
+ Configuration.fromMap(
+ ImmutableMap.<String, String>builder()
+ .put(
+ FlussDataSinkOptions
+ .BOOTSTRAP_SERVERS
+ .key(),
+ "localhost:9123")
+ .put(
+ FlussDataSinkOptions
+ .BUCKET_NUMBER
+ .key(),
+ "database1.table1: 11a")
+ .build()),
+ new Configuration(),
+ Thread.currentThread().getContextClassLoader())))
+ .hasMessageContaining("Invalid bucket number configuration: database1.table1: 11a");
+ }
+
+ private Configuration createValidConfiguration() {
+ return Configuration.fromMap(
+ ImmutableMap.<String, String>builder()
+ .put(FlussDataSinkOptions.BOOTSTRAP_SERVERS.key(), "localhost:9123")
+ .put(FlussDataSinkOptions.BUCKET_KEY.key(), "database1.table1:a,b")
+ .put(FlussDataSinkOptions.BUCKET_NUMBER.key(), "database1.table1:2")
+ .build());
+ }
+}
diff --git a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-fluss/src/test/java/org/apache/flink/cdc/connectors/fluss/sink/FlussEventSerializationSchemaTest.java b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-fluss/src/test/java/org/apache/flink/cdc/connectors/fluss/sink/FlussEventSerializationSchemaTest.java
new file mode 100644
index 000000000..4d7e40657
--- /dev/null
+++ b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-fluss/src/test/java/org/apache/flink/cdc/connectors/fluss/sink/FlussEventSerializationSchemaTest.java
@@ -0,0 +1,265 @@
+/*
+ * 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.connectors.fluss.sink;
+
+import org.apache.flink.cdc.common.data.DecimalData;
+import org.apache.flink.cdc.common.data.LocalZonedTimestampData;
+import org.apache.flink.cdc.common.data.RecordData;
+import org.apache.flink.cdc.common.data.TimestampData;
+import org.apache.flink.cdc.common.data.binary.BinaryStringData;
+import org.apache.flink.cdc.common.event.CreateTableEvent;
+import org.apache.flink.cdc.common.event.DataChangeEvent;
+import org.apache.flink.cdc.common.event.TableId;
+import org.apache.flink.cdc.common.schema.Schema;
+import org.apache.flink.cdc.common.types.BooleanType;
+import org.apache.flink.cdc.common.types.DataType;
+import org.apache.flink.cdc.common.types.DateType;
+import org.apache.flink.cdc.common.types.DecimalType;
+import org.apache.flink.cdc.common.types.FloatType;
+import org.apache.flink.cdc.common.types.IntType;
+import org.apache.flink.cdc.common.types.LocalZonedTimestampType;
+import org.apache.flink.cdc.common.types.TimestampType;
+import org.apache.flink.cdc.common.types.VarCharType;
+import org.apache.flink.cdc.connectors.fluss.sink.v2.FlussEvent;
+import org.apache.flink.cdc.connectors.fluss.sink.v2.FlussOperationType;
+import org.apache.flink.cdc.connectors.fluss.sink.v2.FlussRowWithOp;
+import org.apache.flink.cdc.runtime.typeutils.BinaryRecordDataGenerator;
+
+import com.alibaba.fluss.client.Connection;
+import com.alibaba.fluss.client.ConnectionFactory;
+import com.alibaba.fluss.metadata.TablePath;
+import com.alibaba.fluss.server.testutils.FlussClusterExtension;
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.RegisterExtension;
+
+import java.math.BigDecimal;
+import java.sql.Timestamp;
+import java.time.LocalDate;
+import java.time.LocalDateTime;
+import java.time.ZoneOffset;
+import java.util.Collections;
+import java.util.List;
+import java.util.Objects;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** Test for {@link FlussEventSerializationSchema}. */
+public class FlussEventSerializationSchemaTest {
+ private static FlussEventSerializationSchema serializer;
+
+ @RegisterExtension
+ public static final FlussClusterExtension FLUSS_CLUSTER_EXTENSION =
+ FlussClusterExtension.builder().setNumOfTabletServers(3).build();
+
+ private static Connection conn;
+ private static FlussMetaDataApplier flussMetaDataApplier;
+
+ @BeforeAll
+ static void setup() {
+ conn = ConnectionFactory.createConnection(FLUSS_CLUSTER_EXTENSION.getClientConfig());
+ flussMetaDataApplier =
+ new FlussMetaDataApplier(
+ FLUSS_CLUSTER_EXTENSION.getClientConfig(),
+ Collections.emptyMap(),
+ Collections.emptyMap(),
+ Collections.emptyMap());
+ serializer = new FlussEventSerializationSchema();
+ serializer.open(conn);
+ }
+
+ @AfterAll
+ static void tearDown() throws Exception {
+ if (flussMetaDataApplier != null) {
+ flussMetaDataApplier.close();
+ }
+ if (conn != null) {
+ conn.close();
+ }
+ }
+
+ @Test
+ void testMixedSchemaAndDataChanges() throws Exception {
+ // 1. create table1, and insert/delete/update data
+ TableId table1 = TableId.parse("test.tbl1");
+ Schema schema1 =
+ Schema.newBuilder()
+ .physicalColumn("col1", new IntType())
+ .physicalColumn("col2", new BooleanType())
+ .physicalColumn("col3", new TimestampType())
+ .primaryKey("col1")
+ .build();
+ CreateTableEvent createTableEvent1 = new CreateTableEvent(table1, schema1);
+ flussMetaDataApplier.applySchemaChange(createTableEvent1);
+ verifySchemaChangeEvent(table1, serializer.serialize(createTableEvent1));
+
+ BinaryRecordDataGenerator generator1 =
+ new BinaryRecordDataGenerator(
+ schema1.getColumnDataTypes().toArray(new DataType[0]));
+
+ RecordData record =
+ generator1.generate(
+ new Object[] {
+ 1,
+ true,
+ TimestampData.fromMillis(
+ Timestamp.valueOf("2023-11-27 18:00:00").getTime())
+ });
+ DataChangeEvent insertEvent1 = DataChangeEvent.insertEvent(table1, record);
+
+ verifyDataChangeEvent(
+ table1,
+ new FlussRowWithOp(CdcAsFlussRow.replace(record), FlussOperationType.UPSERT),
+ serializer.serialize(insertEvent1));
+
+ record =
+ generator1.generate(
+ new Object[] {
+ 2,
+ false,
+ TimestampData.fromMillis(
+ Timestamp.valueOf("2023-11-27 19:00:00").getTime())
+ });
+
+ DataChangeEvent deleteEvent1 = DataChangeEvent.deleteEvent(table1, record);
+ verifyDataChangeEvent(
+ table1,
+ new FlussRowWithOp(CdcAsFlussRow.replace(record), FlussOperationType.DELETE),
+ serializer.serialize(deleteEvent1));
+
+ record =
+ generator1.generate(
+ new Object[] {
+ 3,
+ true,
+ TimestampData.fromMillis(
+ Timestamp.valueOf("2023-11-27 21:00:00").getTime())
+ });
+ DataChangeEvent updateEvent1 =
+ DataChangeEvent.updateEvent(
+ table1,
+ generator1.generate(
+ new Object[] {
+ 3,
+ false,
+ TimestampData.fromMillis(
+ Timestamp.valueOf("2023-11-27 20:00:00").getTime())
+ }),
+ record);
+
+ verifyDataChangeEvent(
+ table1,
+ new FlussRowWithOp(CdcAsFlussRow.replace(record), FlussOperationType.UPSERT),
+ serializer.serialize(updateEvent1));
+
+ // 2. create table2, and insert data
+ TableId table2 = TableId.parse("test.tbl2");
+ Schema schema2 =
+ Schema.newBuilder()
+ .physicalColumn("col1", new DateType())
+ .physicalColumn("col2", new FloatType())
+ .physicalColumn("col3", new VarCharType(20))
+ .physicalColumn("col4", new DecimalType(20, 5))
+ .physicalColumn("col5", new LocalZonedTimestampType())
+ .primaryKey("col1")
+ .build();
+ CreateTableEvent createTableEvent2 = new CreateTableEvent(table2, schema2);
+ flussMetaDataApplier.applySchemaChange(createTableEvent2);
+ verifySchemaChangeEvent(table2, serializer.serialize(createTableEvent2));
+
+ BinaryRecordDataGenerator generator2 =
+ new BinaryRecordDataGenerator(
+ schema2.getColumnDataTypes().toArray(new DataType[0]));
+ record =
+ generator2.generate(
+ new Object[] {
+ (int) LocalDate.of(2023, 11, 27).toEpochDay(),
+ 3.4f,
+ BinaryStringData.fromString("insert table2"),
+ DecimalData.fromBigDecimal(new BigDecimal("83.23"), 20, 5),
+ LocalZonedTimestampData.fromInstant(
+ LocalDateTime.of(2023, 11, 27, 21, 0, 0)
+ .toInstant(ZoneOffset.of("+10")))
+ });
+ DataChangeEvent insertEvent2 = DataChangeEvent.insertEvent(table2, record);
+ verifyDataChangeEvent(
+ table2,
+ new FlussRowWithOp(CdcAsFlussRow.replace(record), FlussOperationType.UPSERT),
+ serializer.serialize(insertEvent2));
+
+ record =
+ generator2.generate(
+ new Object[] {
+ 4,
+ 3.4f,
+ BinaryStringData.fromString("insert table2"),
+ DecimalData.fromBigDecimal(new BigDecimal("83.23"), 20, 5),
+ LocalZonedTimestampData.fromInstant(
+ LocalDateTime.of(2023, 11, 27, 21, 0, 0)
+ .toInstant(ZoneOffset.of("+10")))
+ });
+ DataChangeEvent deleteEvent2 = DataChangeEvent.deleteEvent(table2, record);
+ verifyDataChangeEvent(
+ table2,
+ new FlussRowWithOp(CdcAsFlussRow.replace(record), FlussOperationType.DELETE),
+ Objects.requireNonNull(serializer.serialize(deleteEvent2)));
+ }
+
+ private void verifySchemaChangeEvent(TableId tableId, FlussEvent flussEvent) throws Exception {
+ verifySerializeResult(
+ new FlussEvent(
+ TablePath.of(tableId.getSchemaName(), tableId.getTableName()), null, true),
+ flussEvent);
+ }
+
+ private void verifyDataChangeEvent(TableId tableId, FlussRowWithOp row, FlussEvent flussEvent)
+ throws Exception {
+ verifySerializeResult(
+ new FlussEvent(
+ TablePath.of(tableId.getSchemaName(), tableId.getTableName()),
+ Collections.singletonList(row),
+ false),
+ flussEvent);
+ }
+
+ private void verifySerializeResult(FlussEvent expectedEvent, FlussEvent actualflussEvent) {
+
+ assertThat(actualflussEvent.getTablePath()).isEqualTo(expectedEvent.getTablePath());
+ assertThat(actualflussEvent.isShouldRefreshSchema())
+ .isEqualTo(expectedEvent.isShouldRefreshSchema());
+ List<FlussRowWithOp> actualRowWithOps = actualflussEvent.getRowWithOps();
+ List<FlussRowWithOp> expectedRowWithOps = expectedEvent.getRowWithOps();
+ if (actualRowWithOps == null) {
+ assertThat(expectedRowWithOps).isNull();
+ return;
+ }
+ assertThat(actualRowWithOps.size()).isEqualTo(expectedRowWithOps.size());
+ for (int i = 0; i < actualRowWithOps.size(); i++) {
+ FlussRowWithOp actualRowWithOp = actualRowWithOps.get(i);
+ FlussRowWithOp expectedRowWithOp = expectedRowWithOps.get(i);
+ assertThat(actualRowWithOp.getOperationType())
+ .isEqualTo(expectedRowWithOp.getOperationType());
+ assertThat(actualRowWithOp.getRow()).isInstanceOf(CdcAsFlussRow.class);
+ CdcAsFlussRow actualRow = (CdcAsFlussRow) actualRowWithOp.getRow();
+ CdcAsFlussRow expectedRow = (CdcAsFlussRow) expectedRowWithOp.getRow();
+ assertThat(actualRow.getCdcRecord()).isEqualTo(expectedRow.getCdcRecord());
+ assertThat(actualRow.getIndexMapping()).isEqualTo(expectedRow.getIndexMapping());
+ }
+ }
+}
diff --git a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-fluss/src/test/java/org/apache/flink/cdc/connectors/fluss/sink/FlussMetadataApplierTest.java b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-fluss/src/test/java/org/apache/flink/cdc/connectors/fluss/sink/FlussMetadataApplierTest.java
new file mode 100644
index 000000000..69e26bd59
--- /dev/null
+++ b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-fluss/src/test/java/org/apache/flink/cdc/connectors/fluss/sink/FlussMetadataApplierTest.java
@@ -0,0 +1,561 @@
+/*
+ * 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.connectors.fluss.sink;
+
+import org.apache.flink.cdc.common.event.CreateTableEvent;
+import org.apache.flink.cdc.common.event.DropTableEvent;
+import org.apache.flink.cdc.common.event.TableId;
+import org.apache.flink.cdc.common.schema.Schema;
+import org.apache.flink.cdc.common.types.DataTypes;
+import org.apache.flink.cdc.common.types.IntType;
+
+import com.alibaba.fluss.client.Connection;
+import com.alibaba.fluss.client.ConnectionFactory;
+import com.alibaba.fluss.client.admin.Admin;
+import com.alibaba.fluss.exception.InvalidConfigException;
+import com.alibaba.fluss.metadata.TableDescriptor;
+import com.alibaba.fluss.metadata.TableInfo;
+import com.alibaba.fluss.metadata.TablePath;
+import com.alibaba.fluss.server.testutils.FlussClusterExtension;
+import com.alibaba.fluss.types.RowType;
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.RegisterExtension;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
+
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.ExecutionException;
+
+import static com.alibaba.fluss.config.ConfigOptions.TABLE_REPLICATION_FACTOR;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/** Test for {@link FlussMetaDataApplier}. */
+public class FlussMetadataApplierTest {
+ @RegisterExtension
+ private static final FlussClusterExtension FLUSS_CLUSTER_EXTENSION =
+ FlussClusterExtension.builder().setNumOfTabletServers(3).build();
+
+ private static Connection conn;
+ private static Admin admin;
+ private static final String DATABASE_NAME = "test_database";
+
+ @BeforeAll
+ static void setup() {
+ conn = ConnectionFactory.createConnection(FLUSS_CLUSTER_EXTENSION.getClientConfig());
+ admin = conn.getAdmin();
+ }
+
+ @AfterAll
+ static void tearDown() throws Exception {
+ if (admin != null) {
+ admin.close();
+ }
+ if (conn != null) {
+ conn.close();
+ }
+ }
+
+ @AfterEach
+ void cleanup() throws ExecutionException, InterruptedException {
+ admin.dropDatabase(DATABASE_NAME, true, true).get();
+ }
+
+ @ParameterizedTest
+ @ValueSource(booleans = {true, false})
+ void testCreateTableAllTypes(boolean primaryKeyTable) throws Exception {
+ String[] fieldNames =
+ new String[] {
+ "binary_col",
+ "varbinary_col",
+ "bytes_col",
+ "boolean_col",
+ "tinyint_col",
+ "smallint_col",
+ "int_col",
+ "bigint_col",
+ "float_col",
+ "double_col",
+ "decimal_col",
+ "char_col",
+ "varchar_col",
+ "string_col",
+ "date_col",
+ "time_col",
+ "timestamp_col",
+ "timestamp_ltz_col",
+ };
+
+ org.apache.flink.cdc.common.types.DataType[] cdcDataTypes =
+ new org.apache.flink.cdc.common.types.DataType[] {
+ DataTypes.BINARY(10),
+ DataTypes.VARBINARY(10),
+ DataTypes.BYTES(),
+ DataTypes.BOOLEAN(),
+ DataTypes.TINYINT(),
+ DataTypes.SMALLINT(),
+ new IntType(false),
+ DataTypes.BIGINT(),
+ DataTypes.FLOAT(),
+ DataTypes.DOUBLE(),
+ DataTypes.DECIMAL(38, 18),
+ DataTypes.CHAR(10),
+ DataTypes.VARCHAR(100),
+ DataTypes.STRING(),
+ DataTypes.DATE(),
+ DataTypes.TIME(),
+ DataTypes.TIMESTAMP(3),
+ DataTypes.TIMESTAMP_LTZ(6)
+ };
+
+ com.alibaba.fluss.types.DataType[] flussDataTypes =
+ new com.alibaba.fluss.types.DataType[] {
+ com.alibaba.fluss.types.DataTypes.BINARY(10),
+ // fluss not support binary, will be mapped to bytes
+ com.alibaba.fluss.types.DataTypes.BYTES(),
+ com.alibaba.fluss.types.DataTypes.BYTES(),
+ com.alibaba.fluss.types.DataTypes.BOOLEAN(),
+ com.alibaba.fluss.types.DataTypes.TINYINT(),
+ com.alibaba.fluss.types.DataTypes.SMALLINT(),
+ new com.alibaba.fluss.types.IntType(false),
+ com.alibaba.fluss.types.DataTypes.BIGINT(),
+ com.alibaba.fluss.types.DataTypes.FLOAT(),
+ com.alibaba.fluss.types.DataTypes.DOUBLE(),
+ com.alibaba.fluss.types.DataTypes.DECIMAL(38, 18),
+ com.alibaba.fluss.types.DataTypes.CHAR(10),
+ // fluss not support varchar, will be mapped to string
+ com.alibaba.fluss.types.DataTypes.STRING(),
+ com.alibaba.fluss.types.DataTypes.STRING(),
+ com.alibaba.fluss.types.DataTypes.DATE(),
+ com.alibaba.fluss.types.DataTypes.TIME(),
+ com.alibaba.fluss.types.DataTypes.TIMESTAMP(3),
+ com.alibaba.fluss.types.DataTypes.TIMESTAMP_LTZ(6)
+ };
+
+ try (FlussMetaDataApplier applier =
+ new FlussMetaDataApplier(
+ FLUSS_CLUSTER_EXTENSION.getClientConfig(),
+ Collections.emptyMap(),
+ Collections.emptyMap(),
+ Collections.emptyMap())) {
+ // apply create-table event
+ Schema.Builder builder = Schema.newBuilder();
+ for (int i = 0; i < fieldNames.length; i++) {
+ builder.physicalColumn(fieldNames[i], cdcDataTypes[i]);
+ }
+ if (primaryKeyTable) {
+ builder.primaryKey("int_col");
+ }
+ Schema schema = builder.build();
+ TableId tableId = TableId.tableId("default_namespace", DATABASE_NAME, "table1");
+ CreateTableEvent createTableEvent = new CreateTableEvent(tableId, schema);
+ applier.applySchemaChange(createTableEvent);
+
+ // check table
+ TableInfo tableInfo = admin.getTableInfo(new TablePath(DATABASE_NAME, "table1")).get();
+ RowType flussRowType = tableInfo.getRowType();
+ for (int i = 0; i < fieldNames.length; i++) {
+ assertThat(flussRowType.getTypeAt(i)).isEqualTo(flussDataTypes[i]);
+ assertThat(flussRowType.getFieldNames().get(i)).isEqualTo(fieldNames[i]);
+ }
+ assertThat(tableInfo.hasPrimaryKey()).isEqualTo(primaryKeyTable);
+ if (primaryKeyTable) {
+ assertThat(tableInfo.getPrimaryKeys()).containsExactly("int_col");
+ }
+ }
+ }
+
+ @Test
+ void testUnsupportedType() throws Exception {
+ String[] fieldNames = new String[] {"timestamp_tz_col", "array_col", "map_col", "row_col"};
+
+ org.apache.flink.cdc.common.types.DataType[] cdcDataTypes =
+ new org.apache.flink.cdc.common.types.DataType[] {
+ DataTypes.ARRAY(DataTypes.STRING()),
+ DataTypes.MAP(DataTypes.STRING(), DataTypes.INT()),
+ DataTypes.ROW(
+ DataTypes.FIELD("name", DataTypes.STRING()),
+ DataTypes.FIELD("age", DataTypes.INT())),
+ DataTypes.TIMESTAMP_TZ()
+ };
+
+ try (FlussMetaDataApplier applier =
+ new FlussMetaDataApplier(
+ FLUSS_CLUSTER_EXTENSION.getClientConfig(),
+ Collections.emptyMap(),
+ Collections.emptyMap(),
+ Collections.emptyMap())) {
+ for (int i = 0; i < fieldNames.length; i++) {
+ // apply create-table event
+ Schema.Builder builder = Schema.newBuilder();
+ builder.physicalColumn(fieldNames[i], cdcDataTypes[i]);
+ Schema schema = builder.build();
+ TableId tableId = TableId.tableId("default_namespace", DATABASE_NAME, "table1");
+ CreateTableEvent createTableEvent = new CreateTableEvent(tableId, schema);
+ assertThatThrownBy(() -> applier.applySchemaChange(createTableEvent))
+ .rootCause()
+ .isExactlyInstanceOf(UnsupportedOperationException.class);
+ }
+ }
+ }
+
+ @Test
+ void testDropTableEvent() throws Exception {
+ TablePath tablePath = new TablePath("fluss", "table1");
+ admin.createTable(
+ tablePath,
+ TableDescriptor.builder()
+ .schema(
+ com.alibaba.fluss.metadata.Schema.newBuilder()
+ .column(
+ "id",
+ com.alibaba.fluss.types.DataTypes.INT())
+ .build())
+ .build(),
+ true)
+ .get();
+ try (FlussMetaDataApplier applier =
+ new FlussMetaDataApplier(
+ FLUSS_CLUSTER_EXTENSION.getClientConfig(),
+ Collections.emptyMap(),
+ Collections.emptyMap(),
+ Collections.emptyMap())) {
+ assertThat(admin.tableExists(tablePath).get()).isTrue();
+ applier.applySchemaChange(
+ new DropTableEvent(
+ TableId.tableId(
+ tablePath.getDatabaseName(), tablePath.getTableName())));
+ assertThat(admin.tableExists(tablePath).get()).isFalse();
+ }
+ }
+
+ @Test
+ void testCreatePartitionedTable() throws Exception {
+ try (FlussMetaDataApplier applier =
+ new FlussMetaDataApplier(
+ FLUSS_CLUSTER_EXTENSION.getClientConfig(),
+ Collections.emptyMap(),
+ Collections.emptyMap(),
+ Collections.emptyMap())) {
+ // apply create-table event
+ Schema schema =
+ Schema.newBuilder()
+ .physicalColumn("id", DataTypes.INT())
+ .physicalColumn("name", DataTypes.STRING())
+ .primaryKey("id", "name")
+ .partitionKey("name")
+ .build();
+ TableId tableId = TableId.tableId("default_namespace", DATABASE_NAME, "table1");
+ CreateTableEvent createTableEvent = new CreateTableEvent(tableId, schema);
+ applier.applySchemaChange(createTableEvent);
+
+ TableInfo tableInfo = admin.getTableInfo(new TablePath(DATABASE_NAME, "table1")).get();
+ assertThat(tableInfo.getPartitionKeys()).containsExactly("name");
+ }
+ }
+
+ @Test
+ void testPartitionKeyIsNotSubsetOfPrimaryKeys() throws Exception {
+ try (FlussMetaDataApplier applier =
+ new FlussMetaDataApplier(
+ FLUSS_CLUSTER_EXTENSION.getClientConfig(),
+ Collections.emptyMap(),
+ Collections.emptyMap(),
+ Collections.emptyMap())) {
+ // apply create-table event
+ Schema schema =
+ Schema.newBuilder()
+ .physicalColumn("id", DataTypes.INT())
+ .physicalColumn("name", DataTypes.STRING())
+ .primaryKey("id")
+ .partitionKey("name")
+ .build();
+ TableId tableId = TableId.tableId("default_namespace", DATABASE_NAME, "table1");
+ CreateTableEvent createTableEvent = new CreateTableEvent(tableId, schema);
+ assertThatThrownBy(() -> applier.applySchemaChange(createTableEvent))
+ .rootCause()
+ .isExactlyInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining(
+ "Partitioned Primary Key Table requires partition key [name] is a subset of the primary key [id]");
+ }
+ }
+
+ @Test
+ void testBuckNumber() throws Exception {
+ try (FlussMetaDataApplier applier =
+ new FlussMetaDataApplier(
+ FLUSS_CLUSTER_EXTENSION.getClientConfig(),
+ Collections.emptyMap(),
+ Collections.emptyMap(),
+ Collections.singletonMap(DATABASE_NAME + ".table1", 3))) {
+ Schema schema =
+ Schema.newBuilder()
+ .physicalColumn("id", DataTypes.INT())
+ .physicalColumn("name", DataTypes.STRING())
+ .build();
+ TableId tableId1 = TableId.tableId("default_namespace", DATABASE_NAME, "table1");
+ TableId tableId2 = TableId.tableId("default_namespace", DATABASE_NAME, "table2");
+ CreateTableEvent createTableEvent1 = new CreateTableEvent(tableId1, schema);
+ CreateTableEvent createTableEvent2 = new CreateTableEvent(tableId2, schema);
+ applier.applySchemaChange(createTableEvent1);
+ applier.applySchemaChange(createTableEvent2);
+ assertThat(
+ admin.getTableInfo(new TablePath(DATABASE_NAME, "table1"))
+ .get()
+ .getNumBuckets())
+ .isEqualTo(3);
+ // if not set bucket.num, will use the default buck number which is 1.
+ assertThat(
+ admin.getTableInfo(new TablePath(DATABASE_NAME, "table2"))
+ .get()
+ .getNumBuckets())
+ .isEqualTo(1);
+ }
+ }
+
+ @ParameterizedTest
+ @ValueSource(booleans = {true, false})
+ void testBucketKey(boolean primaryKeyTable) throws Exception {
+ try (FlussMetaDataApplier applier =
+ new FlussMetaDataApplier(
+ FLUSS_CLUSTER_EXTENSION.getClientConfig(),
+ Collections.emptyMap(),
+ Collections.singletonMap(
+ DATABASE_NAME + ".table1", Collections.singletonList("id")),
+ Collections.emptyMap())) {
+ Schema.Builder builder =
+ Schema.newBuilder()
+ .physicalColumn("id", DataTypes.INT())
+ .physicalColumn("name", DataTypes.STRING());
+ if (primaryKeyTable) {
+ builder.primaryKey("name", "id");
+ }
+ Schema schema = builder.build();
+ TableId tableId1 = TableId.tableId("default_namespace", DATABASE_NAME, "table1");
+ TableId tableId2 = TableId.tableId("default_namespace", DATABASE_NAME, "table2");
+ CreateTableEvent createTableEvent1 = new CreateTableEvent(tableId1, schema);
+ CreateTableEvent createTableEvent2 = new CreateTableEvent(tableId2, schema);
+ applier.applySchemaChange(createTableEvent1);
+ applier.applySchemaChange(createTableEvent2);
+ assertThat(
+ admin.getTableInfo(new TablePath(DATABASE_NAME, "table1"))
+ .get()
+ .getBucketKeys())
+ .containsExactly("id");
+ // if not set bucket.keys, primary key table will use (primary keys - partition keys) as
+ // bucket keys.
+ if (primaryKeyTable) {
+ assertThat(
+ admin.getTableInfo(new TablePath(DATABASE_NAME, "table2"))
+ .get()
+ .getBucketKeys())
+ .containsExactlyInAnyOrder("id", "name");
+ } else {
+ assertThat(
+ admin.getTableInfo(new TablePath(DATABASE_NAME, "table2"))
+ .get()
+ .getBucketKeys())
+ .isEmpty();
+ }
+ }
+ }
+
+ @ParameterizedTest
+ @ValueSource(booleans = {true, false})
+ void testBucketKeyNotSetOfPrimaryKeys(boolean partitionTables) throws Exception {
+ Schema.Builder builder =
+ Schema.newBuilder()
+ .physicalColumn("id", DataTypes.INT())
+ .physicalColumn("name", DataTypes.STRING())
+ .physicalColumn("age", DataTypes.INT())
+ .primaryKey("id", "name");
+ if (partitionTables) {
+ builder.partitionKey("name");
+ }
+ Schema schema = builder.build();
+ Map<String, List<String>> bucketKeys = new HashMap<>();
+ // table1 is designed to test bucket.keys is not allowed to include any column in partition
+ // keys.
+ // table2 and table 3 is designed to test bucket.keys is must be a subset of primary keys
+ // excluding partition keys.
+ // table4 is designed to test primary key table will use (primary keys - partition keys) as
+ // bucket keys if not set bucket.keys.
+ bucketKeys.put(DATABASE_NAME + ".table1", Collections.singletonList("name"));
+ bucketKeys.put(DATABASE_NAME + ".table2", Collections.singletonList("age"));
+ bucketKeys.put(DATABASE_NAME + ".table3", Collections.singletonList("id"));
+ try (FlussMetaDataApplier applier =
+ new FlussMetaDataApplier(
+ FLUSS_CLUSTER_EXTENSION.getClientConfig(),
+ Collections.emptyMap(),
+ bucketKeys,
+ Collections.emptyMap())) {
+ if (partitionTables) {
+ assertThatThrownBy(
+ () ->
+ applier.applySchemaChange(
+ new CreateTableEvent(
+ TableId.tableId(DATABASE_NAME, "table1"),
+ schema)))
+ .hasMessageContaining(
+ "Bucket key [name] shouldn't include any column in partition keys [name]");
+ } else {
+ applier.applySchemaChange(
+ new CreateTableEvent(TableId.tableId(DATABASE_NAME, "table1"), schema));
+ assertThat(
+ admin.getTableInfo(new TablePath(DATABASE_NAME, "table1"))
+ .get()
+ .getBucketKeys())
+ .containsExactly("name");
+ }
+ assertThatThrownBy(
+ () ->
+ applier.applySchemaChange(
+ new CreateTableEvent(
+ TableId.tableId(DATABASE_NAME, "table2"),
+ schema)))
+ .hasMessageContaining(
+ "Bucket keys must be a subset of primary keys excluding partition keys for primary-key tables");
+ applier.applySchemaChange(
+ new CreateTableEvent(TableId.tableId(DATABASE_NAME, "table3"), schema));
+ assertThat(
+ admin.getTableInfo(new TablePath(DATABASE_NAME, "table3"))
+ .get()
+ .getBucketKeys())
+ .containsExactly("id");
+
+ applier.applySchemaChange(
+ new CreateTableEvent(TableId.tableId(DATABASE_NAME, "table4"), schema));
+ // if not set bucket.keys, primary key table will use (primary keys - partition keys) as
+ // bucket keys.
+ if (partitionTables) {
+ assertThat(
+ admin.getTableInfo(new TablePath(DATABASE_NAME, "table4"))
+ .get()
+ .getBucketKeys())
+ .containsExactly("id");
+ } else {
+ assertThat(
+ admin.getTableInfo(new TablePath(DATABASE_NAME, "table4"))
+ .get()
+ .getBucketKeys())
+ .containsExactlyInAnyOrder("name", "id");
+ }
+ }
+ }
+
+ @Test
+ void testRecreateTableWithDifferentSchema() throws Exception {
+ TableId tableId = TableId.tableId("default_namespace", DATABASE_NAME, "table1");
+ Schema initialSchema =
+ Schema.newBuilder()
+ .physicalColumn("id", DataTypes.INT())
+ .physicalColumn("name", DataTypes.STRING())
+ .physicalColumn("age", DataTypes.INT())
+ .primaryKey("id", "name")
+ .partitionKey("name")
+ .build();
+ Schema differentSchema =
+ Schema.newBuilder()
+ .physicalColumn("id", DataTypes.INT())
+ .physicalColumn("age", DataTypes.INT())
+ .build();
+ // Use another object for comparison.
+ Schema sameSchema =
+ Schema.newBuilder()
+ .physicalColumn("id", DataTypes.INT())
+ .physicalColumn("name", DataTypes.STRING())
+ .physicalColumn("age", DataTypes.INT())
+ .primaryKey("id", "name")
+ .partitionKey("name")
+ .build();
+ // create table with schema1
+ try (FlussMetaDataApplier applier =
+ new FlussMetaDataApplier(
+ FLUSS_CLUSTER_EXTENSION.getClientConfig(),
+ Collections.emptyMap(),
+ Collections.emptyMap(),
+ Collections.emptyMap())) {
+ applier.applySchemaChange(new CreateTableEvent(tableId, initialSchema));
+ }
+
+ // recreate table with schema2 again
+ try (FlussMetaDataApplier applier =
+ new FlussMetaDataApplier(
+ FLUSS_CLUSTER_EXTENSION.getClientConfig(),
+ Collections.emptyMap(),
+ Collections.emptyMap(),
+ Collections.emptyMap())) {
+ assertThatThrownBy(
+ () ->
+ applier.applySchemaChange(
+ new CreateTableEvent(tableId, differentSchema)))
+ .hasMessageContaining(
+ "The table schema inffered by Flink CDC is not matched with current Fluss table schema");
+ }
+
+ // recreate table with schema1 again
+ try (FlussMetaDataApplier applier =
+ new FlussMetaDataApplier(
+ FLUSS_CLUSTER_EXTENSION.getClientConfig(),
+ Collections.emptyMap(),
+ Collections.emptyMap(),
+ Collections.emptyMap())) {
+ applier.applySchemaChange(new CreateTableEvent(tableId, sameSchema));
+ }
+ }
+
+ @Test
+ void testCreateTableWithTableOptions() throws Exception {
+ Schema schema =
+ Schema.newBuilder()
+ .physicalColumn("id", DataTypes.INT())
+ .physicalColumn("name", DataTypes.STRING())
+ .primaryKey("id", "name")
+ .partitionKey("name")
+ .build();
+ TableId tableId = TableId.tableId("default_namespace", DATABASE_NAME, "table1");
+ CreateTableEvent createTableEvent = new CreateTableEvent(tableId, schema);
+ try (FlussMetaDataApplier applier =
+ new FlussMetaDataApplier(
+ FLUSS_CLUSTER_EXTENSION.getClientConfig(),
+ Collections.singletonMap("key", "value"),
+ Collections.emptyMap(),
+ Collections.emptyMap())) {
+ assertThatThrownBy(() -> applier.applySchemaChange(createTableEvent))
+ .rootCause()
+ .isExactlyInstanceOf(InvalidConfigException.class)
+ .hasMessageContaining("'key' is not a Fluss table property");
+ }
+ try (FlussMetaDataApplier applier =
+ new FlussMetaDataApplier(
+ FLUSS_CLUSTER_EXTENSION.getClientConfig(),
+ Collections.singletonMap(TABLE_REPLICATION_FACTOR.key(), "5"),
+ Collections.emptyMap(),
+ Collections.emptyMap())) {
+ applier.applySchemaChange(createTableEvent);
+ TableInfo tableInfo = admin.getTableInfo(new TablePath(DATABASE_NAME, "table1")).get();
+ assertThat(tableInfo.getProperties().get(TABLE_REPLICATION_FACTOR)).isEqualTo(5);
+ }
+ }
+}
diff --git a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-fluss/src/test/java/org/apache/flink/cdc/connectors/fluss/sink/v2/FlussSinkITCase.java b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-fluss/src/test/java/org/apache/flink/cdc/connectors/fluss/sink/v2/FlussSinkITCase.java
new file mode 100644
index 000000000..535d90244
--- /dev/null
+++ b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-fluss/src/test/java/org/apache/flink/cdc/connectors/fluss/sink/v2/FlussSinkITCase.java
@@ -0,0 +1,576 @@
+/*
+ * 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.connectors.fluss.sink.v2;
+
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.cdc.common.data.DecimalData;
+import org.apache.flink.cdc.common.data.LocalZonedTimestampData;
+import org.apache.flink.cdc.common.data.TimestampData;
+import org.apache.flink.cdc.common.data.binary.BinaryStringData;
+import org.apache.flink.cdc.common.event.CreateTableEvent;
+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.schema.Schema;
+import org.apache.flink.cdc.common.types.DataType;
+import org.apache.flink.cdc.common.types.DataTypes;
+import org.apache.flink.cdc.connectors.fluss.sink.FlussEventSerializationSchema;
+import org.apache.flink.cdc.runtime.typeutils.BinaryRecordDataGenerator;
+import org.apache.flink.streaming.api.datastream.DataStreamSource;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.table.api.EnvironmentSettings;
+import org.apache.flink.table.api.TableEnvironment;
+import org.apache.flink.table.api.config.ExecutionConfigOptions;
+import org.apache.flink.test.util.AbstractTestBase;
+import org.apache.flink.types.Row;
+import org.apache.flink.util.CloseableIterator;
+
+import com.alibaba.fluss.server.testutils.FlussClusterExtension;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.RegisterExtension;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
+
+import java.math.BigDecimal;
+import java.time.LocalDateTime;
+import java.time.ZoneId;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+
+import static com.alibaba.fluss.config.ConfigOptions.BOOTSTRAP_SERVERS;
+import static com.alibaba.fluss.flink.source.testutils.FlinkRowAssertionsUtils.assertResultsIgnoreOrder;
+import static com.alibaba.fluss.server.testutils.FlussClusterExtension.BUILTIN_DATABASE;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/** Integration tests for FlussSink. */
+public class FlussSinkITCase extends AbstractTestBase {
+
+ @RegisterExtension
+ public static final FlussClusterExtension FLUSS_CLUSTER_EXTENSION =
+ FlussClusterExtension.builder().setNumOfTabletServers(3).build();
+
+ static final String CATALOG_NAME = "test_catalog";
+ static final String DEFAULT_DB = "default_db";
+
+ protected TableEnvironment tBatchEnv;
+
+ @BeforeEach
+ void before() {
+ // open a catalog so that we can get table from the catalog
+ String bootstrapServers = FLUSS_CLUSTER_EXTENSION.getBootstrapServers();
+
+ // create batch table environment
+ tBatchEnv =
+ TableEnvironment.create(EnvironmentSettings.newInstance().inBatchMode().build());
+ tBatchEnv.executeSql(
+ String.format(
+ "create catalog %s with ('type' = 'fluss', '%s' = '%s')",
+ CATALOG_NAME, BOOTSTRAP_SERVERS.key(), bootstrapServers));
+ tBatchEnv.executeSql("use catalog " + CATALOG_NAME);
+ tBatchEnv
+ .getConfig()
+ .set(ExecutionConfigOptions.TABLE_EXEC_RESOURCE_DEFAULT_PARALLELISM, 2);
+ // create database
+ tBatchEnv.executeSql("create database " + DEFAULT_DB);
+ tBatchEnv.useDatabase(DEFAULT_DB);
+ }
+
+ @AfterEach
+ void after() {
+ tBatchEnv.useDatabase(BUILTIN_DATABASE);
+ tBatchEnv.executeSql(String.format("drop database %s cascade", DEFAULT_DB));
+ }
+
+ @ParameterizedTest
+ @ValueSource(booleans = {true, false})
+ void testAllSinkType(boolean primaryKeyTable) throws Exception {
+ TableId tableId = TableId.tableId("default_namespace", DEFAULT_DB, "test_all_sink_table");
+ tBatchEnv
+ .executeSql(
+ String.format(
+ "CREATE TABLE %s.%s (\n"
+ + " char_type CHAR,\n"
+ + " varchar_type VARCHAR,\n"
+ + " string_type STRING,\n"
+ + " boolean_type BOOLEAN,\n"
+ + " decimal_type DECIMAL(6, 2),\n"
+ + " tinyint_type TINYINT,\n"
+ + " smallint_type SMALLINT,\n"
+ + " int_type INT,\n"
+ + " bigint_type BIGINT,\n"
+ + " float_type FLOAT,\n"
+ + " double_type DOUBLE,\n"
+ + " date_type DATE,\n"
+ + " time_type TIME,\n"
+ + " timestamp_type TIMESTAMP,\n"
+ + " timestamp_ltz_type TIMESTAMP_LTZ(8)\n"
+ + (primaryKeyTable
+ ? " ,PRIMARY KEY (int_type) NOT ENFORCED \n"
+ : "")
+ + ");",
+ tableId.getSchemaName(),
+ tableId.getTableName()))
+ .await();
+ String[] fieldNames =
+ new String[] {
+ "char_type",
+ "varchar_type",
+ "string_type",
+ "boolean_type",
+ "decimal_type",
+ "tinyint_type",
+ "smallint_type",
+ "int_type",
+ "bigint_type",
+ "float_type",
+ "double_type",
+ "date_type",
+ "time_type",
+ "timestamp_type",
+ "timestamp_ltz_type"
+ };
+ String[] pkFieldNames = primaryKeyTable ? new String[] {"int_type"} : new String[0];
+
+ DataType[] dataTypes =
+ new DataType[] {
+ DataTypes.CHAR(1),
+ DataTypes.VARCHAR(20),
+ DataTypes.STRING(),
+ DataTypes.BOOLEAN(),
+ DataTypes.DECIMAL(6, 2),
+ DataTypes.TINYINT(),
+ DataTypes.SMALLINT(),
+ DataTypes.INT(),
+ DataTypes.BIGINT(),
+ DataTypes.FLOAT(),
+ DataTypes.DOUBLE(),
+ DataTypes.DATE(),
+ DataTypes.TIME(),
+ DataTypes.TIMESTAMP(),
+ DataTypes.TIMESTAMP_LTZ(8)
+ };
+
+ Object[][] insertedValues =
+ new Object[][] {
+ new Object[] {
+ BinaryStringData.fromString("a"),
+ BinaryStringData.fromString("test character"),
+ BinaryStringData.fromString("test text"),
+ false,
+ DecimalData.fromBigDecimal(new BigDecimal("8119.21"), 6, 2),
+ Byte.valueOf("1"),
+ Short.valueOf("32767"),
+ 32768,
+ 652482L,
+ 20.2007F,
+ 8.58965,
+ // 2023-11-12 - 1970-01-01 = 19673 days
+ 19673,
+ (8 * 3600 + 30 * 60 + 15) * 1000,
+ TimestampData.fromLocalDateTime(
+ LocalDateTime.of(2023, 11, 11, 11, 11, 11, 11)),
+ LocalZonedTimestampData.fromInstant(
+ LocalDateTime.of(2023, 11, 11, 11, 11, 11, 11)
+ .atZone(ZoneId.of("GMT+05:00"))
+ .toInstant())
+ },
+ new Object[] {
+ null, null, null, null, null, null, null, 0, null, null, null, null, null,
+ null, null
+ }
+ };
+ // default timezone is asian/shanghai
+ List<String> expectedRows =
+ Arrays.asList(
+ String.format(
+ "+I[a, test character, test text, false, 8119.21, 1, 32767, 32768, 652482, 20.2007, 8.58965, 2023-11-12, 08:30:15, %s, 2023-11-11T06:11:11.000000011Z]",
+ primaryKeyTable
+ ? "2023-11-11T11:11:11.000000011"
+ : "2023-11-11T11:11:11"),
+ "+I[null, null, null, null, null, null, null, 0, null, null, null, null, null, null, null]");
+
+ testInsertSingleTable(
+ tableId,
+ fieldNames,
+ pkFieldNames,
+ dataTypes,
+ insertedValues,
+ expectedRows,
+ new String[0]);
+ }
+
+ @Test
+ void testPartitionTable() throws Exception {
+ TableId tableId =
+ TableId.tableId("default_namespace", DEFAULT_DB, "test_partition_sink_table");
+ tBatchEnv
+ .executeSql(
+ String.format(
+ "CREATE TABLE %s.%s (\n"
+ + " id INT,\n"
+ + " ds STRING,\n"
+ + " title STRING,\n"
+ + " body STRING, \n"
+ + " PRIMARY KEY (id, ds) NOT ENFORCED "
+ + ");",
+ tableId.getSchemaName(), tableId.getTableName()))
+ .await();
+
+ String[] fieldNames = new String[] {"id", "ds", "title", "body"};
+ String[] pkFieldNames = new String[] {"id", "ds"};
+ String[] partitionKeys = {"ds"};
+ DataType[] dataTypes =
+ new DataType[] {
+ DataTypes.INT(), DataTypes.STRING(), DataTypes.STRING(), DataTypes.STRING()
+ };
+ Object[][] insertedValues =
+ new Object[][] {
+ new Object[] {
+ 1,
+ BinaryStringData.fromString("20220727"),
+ BinaryStringData.fromString("title_01"),
+ BinaryStringData.fromString("partition_test")
+ }
+ };
+ String[] expected = new String[] {"+I[1, 20220727, title_01, partition_test]"};
+ testInsertSingleTable(
+ tableId,
+ fieldNames,
+ pkFieldNames,
+ dataTypes,
+ insertedValues,
+ Arrays.asList(expected),
+ partitionKeys);
+ }
+
+ @Test
+ void testInsertPkRecordIntoLogTable() throws Exception {
+ TableId tableId = TableId.tableId("default_namespace", DEFAULT_DB, "test_pk_to_log_table");
+ tBatchEnv
+ .executeSql(
+ String.format(
+ "CREATE TABLE %s.%s (\n"
+ + " id INT,\n"
+ + " name STRING \n"
+ + ");",
+ tableId.getSchemaName(), tableId.getTableName()))
+ .await();
+ Schema schema =
+ Schema.newBuilder()
+ .physicalColumn("id", DataTypes.INT().notNull())
+ .physicalColumn("name", DataTypes.STRING())
+ .primaryKey("id")
+ .build();
+
+ List<Event> events = new ArrayList<>();
+ BinaryRecordDataGenerator table1dataGenerator =
+ new BinaryRecordDataGenerator(schema.getColumnDataTypes().toArray(new DataType[0]));
+ events.add(new CreateTableEvent(tableId, schema));
+ events.add(
+ DataChangeEvent.insertEvent(
+ tableId,
+ table1dataGenerator.generate(
+ new Object[] {1, BinaryStringData.fromString("Alice")})));
+ assertThatThrownBy(() -> submitJob(events))
+ .rootCause()
+ .isExactlyInstanceOf(UnsupportedOperationException.class)
+ .hasMessageContaining("UPSERT for log table");
+ }
+
+ @Test
+ void testInsertNonPkRecordIntoPrimaryTable() throws Exception {
+ TableId tableId =
+ TableId.tableId("default_namespace", DEFAULT_DB, "test_non_pk_to_primary_table");
+ tBatchEnv
+ .executeSql(
+ String.format(
+ "CREATE TABLE %s.%s (\n"
+ + " id INT,\n"
+ + " name STRING ,\n"
+ + " PRIMARY KEY (id) NOT ENFORCED "
+ + ");",
+ tableId.getSchemaName(), tableId.getTableName()))
+ .await();
+ Schema schema =
+ Schema.newBuilder()
+ .physicalColumn("id", DataTypes.INT().notNull())
+ .physicalColumn("name", DataTypes.STRING())
+ .build();
+
+ List<Event> events = new ArrayList<>();
+ BinaryRecordDataGenerator table1dataGenerator =
+ new BinaryRecordDataGenerator(schema.getColumnDataTypes().toArray(new DataType[0]));
+ events.add(new CreateTableEvent(tableId, schema));
+ events.add(
+ DataChangeEvent.insertEvent(
+ tableId,
+ table1dataGenerator.generate(
+ new Object[] {1, BinaryStringData.fromString("Alice")})));
+ assertThatThrownBy(() -> submitJob(events))
+ .rootCause()
+ .isExactlyInstanceOf(UnsupportedOperationException.class)
+ .hasMessageContaining("APPEND for primary key table");
+ }
+
+ @Test
+ void testDelete() throws Exception {
+ TableId tableId = TableId.tableId("default_namespace", DEFAULT_DB, "test_delete");
+ tBatchEnv
+ .executeSql(
+ String.format(
+ "CREATE TABLE %s.%s (\n"
+ + " id INT,\n"
+ + " name STRING ,\n"
+ + " PRIMARY KEY (id) NOT ENFORCED "
+ + ");",
+ tableId.getSchemaName(), tableId.getTableName()))
+ .await();
+ Schema schema =
+ Schema.newBuilder()
+ .physicalColumn("id", DataTypes.INT().notNull())
+ .physicalColumn("name", DataTypes.STRING())
+ .primaryKey("id")
+ .build();
+
+ List<Event> events = new ArrayList<>();
+ BinaryRecordDataGenerator table1dataGenerator =
+ new BinaryRecordDataGenerator(schema.getColumnDataTypes().toArray(new DataType[0]));
+ events.add(new CreateTableEvent(tableId, schema));
+ events.add(
+ DataChangeEvent.insertEvent(
+ tableId,
+ table1dataGenerator.generate(
+ new Object[] {1, BinaryStringData.fromString("Alice")})));
+ events.add(
+ DataChangeEvent.insertEvent(
+ tableId,
+ table1dataGenerator.generate(
+ new Object[] {2, BinaryStringData.fromString("Bob")})));
+ events.add(
+ DataChangeEvent.deleteEvent(
+ tableId,
+ table1dataGenerator.generate(
+ new Object[] {1, BinaryStringData.fromString("Alice")})));
+
+ submitJob(events);
+ checkResult(tableId, Collections.singletonList("+I[2, Bob]"));
+ }
+
+ @Test
+ void testUpdate() throws Exception {
+ TableId tableId = TableId.tableId("default_namespace", DEFAULT_DB, "test_delete");
+ tBatchEnv
+ .executeSql(
+ String.format(
+ "CREATE TABLE %s.%s (\n"
+ + " id INT,\n"
+ + " name STRING ,\n"
+ + " PRIMARY KEY (id) NOT ENFORCED "
+ + ");",
+ tableId.getSchemaName(), tableId.getTableName()))
+ .await();
+ Schema schema =
+ Schema.newBuilder()
+ .physicalColumn("id", DataTypes.INT().notNull())
+ .physicalColumn("name", DataTypes.STRING())
+ .primaryKey("id")
+ .build();
+
+ List<Event> events = new ArrayList<>();
+ BinaryRecordDataGenerator table1dataGenerator =
+ new BinaryRecordDataGenerator(schema.getColumnDataTypes().toArray(new DataType[0]));
+ events.add(new CreateTableEvent(tableId, schema));
+ events.add(
+ DataChangeEvent.insertEvent(
+ tableId,
+ table1dataGenerator.generate(
+ new Object[] {1, BinaryStringData.fromString("Alice")})));
+ events.add(
+ DataChangeEvent.insertEvent(
+ tableId,
+ table1dataGenerator.generate(
+ new Object[] {2, BinaryStringData.fromString("Bob")})));
+
+ events.add(
+ DataChangeEvent.updateEvent(
+ tableId,
+ table1dataGenerator.generate(
+ new Object[] {1, BinaryStringData.fromString("Alice")}),
+ table1dataGenerator.generate(
+ new Object[] {1, BinaryStringData.fromString("Mike")})));
+
+ submitJob(events);
+ checkResult(tableId, Arrays.asList("+I[1, Mike]", "+I[2, Bob]"));
+ }
+
+ @Test
+ void testMultipleTables() throws Exception {
+ TableId tableId1 = TableId.tableId("default_namespace", DEFAULT_DB, "test_multiple_1");
+ TableId tableId2 = TableId.tableId("default_namespace", DEFAULT_DB, "test_multiple_2");
+ String prepareCreateTableSql =
+ "CREATE TABLE %s.%s (\n"
+ + " id INT,\n"
+ + " ds STRING,\n"
+ + " title STRING,\n"
+ + " body STRING, \n"
+ + " PRIMARY KEY (id, ds) NOT ENFORCED );";
+ tBatchEnv
+ .executeSql(
+ String.format(
+ prepareCreateTableSql,
+ tableId1.getSchemaName(),
+ tableId1.getTableName()))
+ .await();
+ tBatchEnv
+ .executeSql(
+ String.format(
+ prepareCreateTableSql,
+ tableId2.getSchemaName(),
+ tableId2.getTableName()))
+ .await();
+
+ String[] fieldNames = new String[] {"id", "ds", "title", "body"};
+
+ String[] pkFieldNames = new String[] {"id", "ds"};
+ DataType[] dataTypes =
+ new DataType[] {
+ DataTypes.INT(), DataTypes.STRING(), DataTypes.STRING(), DataTypes.STRING()
+ };
+ Object[][] insertedValues =
+ new Object[][] {
+ new Object[] {
+ 1,
+ BinaryStringData.fromString("20220727"),
+ BinaryStringData.fromString("title_01"),
+ BinaryStringData.fromString("partition_test")
+ }
+ };
+
+ List<Event> events = new ArrayList<>();
+ events.addAll(
+ constructInsertEvents(
+ tableId1,
+ fieldNames,
+ pkFieldNames,
+ dataTypes,
+ insertedValues,
+ new String[0]));
+ events.addAll(
+ constructInsertEvents(
+ tableId2,
+ fieldNames,
+ pkFieldNames,
+ dataTypes,
+ insertedValues,
+ new String[0]));
+
+ submitJob(events);
+ checkResult(
+ tableId1, Collections.singletonList("+I[1, 20220727, title_01, partition_test]"));
+ checkResult(
+ tableId2, Collections.singletonList("+I[1, 20220727, title_01, partition_test]"));
+ }
+
+ private void testInsertSingleTable(
+ TableId tableId,
+ String[] fieldNames,
+ String[] pkFieldNames,
+ DataType[] dataTypes,
+ Object[][] insertedValues,
+ List<String> expectedRows,
+ String[] partitionKeys)
+ throws Exception {
+
+ List<Event> events =
+ constructInsertEvents(
+ tableId,
+ fieldNames,
+ pkFieldNames,
+ dataTypes,
+ insertedValues,
+ partitionKeys);
+ submitJob(events);
+ checkResult(tableId, expectedRows);
+ }
+
+ private List<Event> constructInsertEvents(
+ TableId tableId,
+ String[] fieldNames,
+ String[] pkFieldNames,
+ DataType[] dataTypes,
+ Object[][] insertedValues,
+ String[] partitionKeys) {
+ List<Event> events = new ArrayList<>();
+
+ // 1. create table
+ Schema.Builder builder = Schema.newBuilder();
+ for (int i = 0; i < fieldNames.length; i++) {
+ builder.physicalColumn(fieldNames[i], dataTypes[i]);
+ }
+
+ org.apache.flink.cdc.common.schema.Schema table1Schema =
+ builder.primaryKey(pkFieldNames).partitionKey(partitionKeys).build();
+ events.add(new CreateTableEvent(tableId, table1Schema));
+
+ // table event
+ BinaryRecordDataGenerator table1dataGenerator =
+ new BinaryRecordDataGenerator(
+ table1Schema.getColumnDataTypes().toArray(new DataType[0]));
+
+ for (int i = 0; i < insertedValues.length; i++) {
+ events.add(
+ DataChangeEvent.insertEvent(
+ tableId, table1dataGenerator.generate(insertedValues[i])));
+ }
+
+ return events;
+ }
+
+ private void submitJob(List<Event> events) throws Exception {
+
+ StreamExecutionEnvironment environment =
+ StreamExecutionEnvironment.getExecutionEnvironment();
+ environment.setParallelism(1);
+
+ DataStreamSource<Event> source =
+ environment.fromData(events, TypeInformation.of(Event.class));
+
+ FlussEventSerializationSchema flussRecordSerializer = new FlussEventSerializationSchema();
+ FlussSink<Event> flussSink =
+ new FlussSink<>(FLUSS_CLUSTER_EXTENSION.getClientConfig(), flussRecordSerializer);
+ source.sinkTo(flussSink);
+ environment.execute();
+ }
+
+ private void checkResult(TableId tableId, List<String> expectedRows) {
+ CloseableIterator<Row> rowIter =
+ tBatchEnv
+ .executeSql(
+ String.format(
+ "select * from %s.%s limit %d",
+ tableId.getSchemaName(),
+ tableId.getTableName(),
+ expectedRows.size()))
+ .collect();
+ assertResultsIgnoreOrder(rowIter, expectedRows, true);
+ }
+}
diff --git a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-fluss/src/test/resources/log4j2-test.properties b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-fluss/src/test/resources/log4j2-test.properties
new file mode 100644
index 000000000..3316848d4
--- /dev/null
+++ b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-fluss/src/test/resources/log4j2-test.properties
@@ -0,0 +1,25 @@
+################################################################################
+# 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.
+################################################################################
+
+# Set root logger level to ERROR to not flood build logs
+# set manually to INFO for debugging purposes
+rootLogger.level=OFF
+rootLogger.appenderRef.test.ref = TestLogger
+
+appender.testlogger.name = TestLogger
+appender.testlogger.type = CONSOLE
+appender.testlogger.target = SYSTEM_ERR
+appender.testlogger.layout.type = PatternLayout
+appender.testlogger.layout.pattern = %-4r [%t] %-5p %c - %m%n
diff --git a/flink-cdc-connect/flink-cdc-pipeline-connectors/pom.xml b/flink-cdc-connect/flink-cdc-pipeline-connectors/pom.xml
index c384c240d..bb22bf7ae 100644
--- a/flink-cdc-connect/flink-cdc-pipeline-connectors/pom.xml
+++ b/flink-cdc-connect/flink-cdc-pipeline-connectors/pom.xml
@@ -38,6 +38,7 @@ limitations under the License.
<module>flink-cdc-pipeline-connector-oceanbase</module>
<module>flink-cdc-pipeline-connector-maxcompute</module>
<module>flink-cdc-pipeline-connector-iceberg</module>
+ <module>flink-cdc-pipeline-connector-fluss</module>
</modules>
<dependencies>