This is an automated email from the ASF dual-hosted git repository.
renqs 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 253ef9256 [pipeline-connector][kafka] Add kafka pipeline data sink
connector. (#2938)
253ef9256 is described below
commit 253ef9256bfef23a76d6c63d94b67685e326043e
Author: Kunni <[email protected]>
AuthorDate: Tue Apr 23 18:00:36 2024 +0800
[pipeline-connector][kafka] Add kafka pipeline data sink connector. (#2938)
---
.github/workflows/flink_cdc.yml | 3 +-
docs/content.zh/docs/connectors/kafka.md | 224 ++++++++++
docs/content/docs/connectors/kafka.md | 222 +++++++++
.../flink-cdc-pipeline-connector-kafka/pom.xml | 99 +++++
.../kafka/json/ChangeLogJsonFormatFactory.java | 82 ++++
.../kafka/json/JsonSerializationType.java | 44 ++
.../cdc/connectors/kafka/json/TableSchemaInfo.java | 163 +++++++
.../json/canal/CanalJsonSerializationSchema.java | 212 +++++++++
.../debezium/DebeziumJsonSerializationSchema.java | 197 ++++++++
.../cdc/connectors/kafka/sink/KafkaDataSink.java | 100 +++++
.../kafka/sink/KafkaDataSinkFactory.java | 111 +++++
.../kafka/sink/KafkaDataSinkOptions.java | 66 +++
.../PipelineKafkaRecordSerializationSchema.java | 157 +++++++
.../org.apache.flink.cdc.common.factories.Factory | 16 +
.../kafka/json/MockInitializationContext.java | 38 ++
.../connectors/kafka/json/TableSchemaInfoTest.java | 164 +++++++
.../canal/CanalJsonSerializationSchemaTest.java | 133 ++++++
.../DebeziumJsonSerializationSchemaTest.java | 128 ++++++
.../kafka/sink/KafkaDataSinkFactoryTest.java | 47 ++
.../connectors/kafka/sink/KafkaDataSinkITCase.java | 495 +++++++++++++++++++++
.../flink/cdc/connectors/kafka/sink/KafkaUtil.java | 188 ++++++++
.../src/test/resources/log4j2-test.properties | 26 ++
.../flink-cdc-pipeline-connectors/pom.xml | 1 +
23 files changed, 2915 insertions(+), 1 deletion(-)
diff --git a/.github/workflows/flink_cdc.yml b/.github/workflows/flink_cdc.yml
index 9c72e7d5d..ad842d7b4 100644
--- a/.github/workflows/flink_cdc.yml
+++ b/.github/workflows/flink_cdc.yml
@@ -45,7 +45,8 @@ env:
flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-values,\
flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-mysql,\
flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-doris,\
-
flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-starrocks"
+
flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-starrocks,\
+
flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-kafka"
MODULES_MYSQL: "\
flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc,\
diff --git a/docs/content.zh/docs/connectors/kafka.md
b/docs/content.zh/docs/connectors/kafka.md
new file mode 100644
index 000000000..5402c406d
--- /dev/null
+++ b/docs/content.zh/docs/connectors/kafka.md
@@ -0,0 +1,224 @@
+---
+title: "Kafka"
+weight: 5
+type: docs
+aliases:
+- /connectors/kafka
+---
+<!--
+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.
+-->
+
+# Kafka Pipeline 连接器
+
+Kafka Pipeline 连接器可以用作 Pipeline 的 *Data
Sink*，将数据写入[Kafka](https://kafka.apache.org)。 本文档介绍如何设置 Kafka Pipeline 连接器。
+
+## 连接器的功能
+* 自动建表
+* 表结构变更同步
+* 数据实时同步
+
+如何创建 Pipeline
+----------------
+
+从 MySQL 读取数据同步到 Kafka 的 Pipeline 可以定义如下：
+
+```yaml
+source:
+ type: mysql
+ name: MySQL Source
+ hostname: 127.0.0.1
+ port: 3306
+ username: admin
+ password: pass
+ tables: adb.\.*, bdb.user_table_[0-9]+, [app|web].order_\.*
+ server-id: 5401-5404
+
+sink:
+ type: kafka
+ name: Kafka Sink
+ properties.bootstrap.servers: PLAINTEXT://localhost:62510
+
+pipeline:
+ name: MySQL to Kafka Pipeline
+ parallelism: 2
+```
+
+Pipeline 连接器配置项
+----------------
+<div class="highlight">
+<table class="colwidths-auto docutils">
+ <thead>
+ <tr>
+ <th class="text-left" style="width: 25%">Option</th>
+ <th class="text-left" style="width: 8%">Required</th>
+ <th class="text-left" style="width: 7%">Default</th>
+ <th class="text-left" style="width: 10%">Type</th>
+ <th class="text-left" style="width: 50%">Description</th>
+ </tr>
+ </thead>
+ <tbody>
+ <tr>
+ <td>type</td>
+ <td>required</td>
+ <td style="word-wrap: break-word;">(none)</td>
+ <td>String</td>
+ <td>指定要使用的连接器, 这里需要设置成 <code>'kafka'</code>。 </td>
+ </tr>
+ <tr>
+ <td>name</td>
+ <td>optional</td>
+ <td style="word-wrap: break-word;">(none)</td>
+ <td>String</td>
+ <td>Sink 的名称。 </td>
+ </tr>
+ <tr>
+ <td>value.format</td>
+ <td>optional</td>
+ <td style="word-wrap: break-word;">(none)</td>
+ <td>String</td>
+ <td>用于序列化 Kafka 消息的值部分数据的格式。可选的填写值包括 <a
href="https://debezium.io/documentation/reference/stable/integrations/serdes.html";>debezium-json</a>
和 <a href="https://github.com/alibaba/canal/wiki";>canal-json</a>, 默认值为
`debezium-json`，并且目前不支持用户自定义输出格式。 </td>
+ </tr>
+ <tr>
+ <td>properties.bootstrap.servers</td>
+ <td>required</td>
+ <td style="word-wrap: break-word;">(none)</td>
+ <td>String</td>
+ <td>用于建立与 Kafka 集群初始连接的主机/端口对列表。</td>
+ </tr>
+ <tr>
+ <td>topic</td>
+ <td>optional</td>
+ <td style="word-wrap: break-word;">(none)</td>
+ <td>String</td>
+ <td>如果配置了这个参数，所有的消息都会发送到这一个主题。</td>
+ </tr>
+ <tr>
+ <td>sink.add-tableId-to-header-enabled</td>
+ <td>optional</td>
+ <td style="word-wrap: break-word;">(none)</td>
+ <td>Boolean</td>
+ <td>如果配置了这个参数，所有的消息都会带上键为 `namespace`, 'schemaName', 'tableName'，值为事件
TableId 里对应的 字符串的 header。</td>
+ </tr>
+ <tr>
+ <td>properties.*</td>
+ <td>optional</td>
+ <td style="word-wrap: break-word;">(none)</td>
+ <td>String</td>
+ <td>将 Kafka 支持的参数传递给 pipeline，参考 <a
href="https://kafka.apache.org/28/documentation.html#consumerconfigs";>Kafka
consume options</a>。 </td>
+ </tr>
+ <tr>
+ <td>sink.custom-header</td>
+ <td>optional</td>
+ <td style="word-wrap: break-word;">(none)</td>
+ <td>String</td>
+ <td>Kafka 记录自定义的 Header。每个 Header 使用 ','分割， 键值使用 ':' 分割。举例来说，可以使用这种方式
'key1:value1,key2:value2'。 </td>
+ </tr>
+ </tbody>
+</table>
+</div>
+
+使用说明
+--------
+
+* 写入 Kafka 的 topic 默认会是上游表 `namespace.schemaName.tableName` 对应的字符串，可以通过
pipeline 的 route 功能进行修改。
+* 如果配置了 `topic` 参数，所有的消息都会发送到这一个主题。
+* 写入 Kafka 的 topic 如果不存在，则会默认创建。
+
+数据类型映射
+----------------
+<div class="wy-table-responsive">
+<table class="colwidths-auto docutils">
+ <thead>
+ <tr>
+ <th class="text-left">CDC type</th>
+ <th class="text-left">JSON type</th>
+ <th class="text-left" style="width:60%;">NOTE</th>
+ </tr>
+ </thead>
+ <tbody>
+ <tr>
+ <td>TINYINT</td>
+ <td>TINYINT</td>
+ <td></td>
+ </tr>
+ <tr>
+ <td>SMALLINT</td>
+ <td>SMALLINT</td>
+ <td></td>
+ </tr>
+ <tr>
+ <td>INT</td>
+ <td>INT</td>
+ <td></td>
+ </tr>
+ <tr>
+ <td>BIGINT</td>
+ <td>BIGINT</td>
+ <td></td>
+ </tr>
+ <tr>
+ <td>FLOAT</td>
+ <td>FLOAT</td>
+ <td></td>
+ </tr>
+ <tr>
+ <td>DOUBLE</td>
+ <td>DOUBLE</td>
+ <td></td>
+ </tr>
+ <tr>
+ <td>DECIMAL(p, s)</td>
+ <td>DECIMAL(p, s)</td>
+ <td></td>
+ </tr>
+ <tr>
+ <td>BOOLEAN</td>
+ <td>BOOLEAN</td>
+ <td></td>
+ </tr>
+ <tr>
+ <td>DATE</td>
+ <td>DATE</td>
+ <td></td>
+ </tr>
+ <tr>
+ <td>TIMESTAMP</td>
+ <td>DATETIME</td>
+ <td></td>
+ </tr>
+ <tr>
+ <td>TIMESTAMP_LTZ</td>
+ <td>TIMESTAMP_LTZ</td>
+ <td></td>
+ </tr>
+ <tr>
+ <td>CHAR(n)</td>
+ <td>CHAR(n)</td>
+ <td></td>
+ </tr>
+ <tr>
+ <td>VARCHAR(n)</td>
+ <td>VARCHAR(n)</td>
+ <td></td>
+ </tr>
+ </tbody>
+</table>
+</div>
+
+{{< top >}}
\ No newline at end of file
diff --git a/docs/content/docs/connectors/kafka.md
b/docs/content/docs/connectors/kafka.md
new file mode 100644
index 000000000..83a735a2d
--- /dev/null
+++ b/docs/content/docs/connectors/kafka.md
@@ -0,0 +1,222 @@
+---
+title: "Kafka"
+weight: 5
+type: docs
+aliases:
+- /connectors/kafka
+---
+<!--
+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.
+-->
+
+# Kafka Pipeline Connector
+
+The Kafka Pipeline connector can be used as the *Data Sink* of the pipeline,
and write data to [Kafka](https://kafka.apache.org). This document describes
how to set up the Kafka Pipeline connector.
+
+## What can the connector do?
+* Data synchronization
+
+How to create Pipeline
+----------------
+
+The pipeline for reading data from MySQL and sink to Kafka can be defined as
follows:
+
+```yaml
+source:
+ type: mysql
+ name: MySQL Source
+ hostname: 127.0.0.1
+ port: 3306
+ username: admin
+ password: pass
+ tables: adb.\.*, bdb.user_table_[0-9]+, [app|web].order_\.*
+ server-id: 5401-5404
+
+sink:
+ type: kafka
+ name: Kafka Sink
+ properties.bootstrap.servers: PLAINTEXT://localhost:62510
+
+pipeline:
+ name: MySQL to Kafka Pipeline
+ parallelism: 2
+```
+
+Pipeline Connector Options
+----------------
+<div class="highlight">
+<table class="colwidths-auto docutils">
+ <thead>
+ <tr>
+ <th class="text-left" style="width: 25%">Option</th>
+ <th class="text-left" style="width: 8%">Required</th>
+ <th class="text-left" style="width: 7%">Default</th>
+ <th class="text-left" style="width: 10%">Type</th>
+ <th class="text-left" style="width: 50%">Description</th>
+ </tr>
+ </thead>
+ <tbody>
+ <tr>
+ <td>type</td>
+ <td>required</td>
+ <td style="word-wrap: break-word;">(none)</td>
+ <td>String</td>
+ <td>Specify what connector to use, here should be
<code>'kafka'</code>.</td>
+ </tr>
+ <tr>
+ <td>name</td>
+ <td>optional</td>
+ <td style="word-wrap: break-word;">(none)</td>
+ <td>String</td>
+ <td>The name of the sink.</td>
+ </tr>
+ <tr>
+ <td>value.format</td>
+ <td>optional</td>
+ <td style="word-wrap: break-word;">(none)</td>
+ <td>String</td>
+ <td>The format used to serialize the value part of Kafka messages.
Available options are <a
href="https://debezium.io/documentation/reference/stable/integrations/serdes.html";>debezium-json</a>
and <a href="https://github.com/alibaba/canal/wiki";>canal-json</a>, default
option is `debezium-json`, and do not support user-defined format now. </td>
+ </tr>
+ <tr>
+ <td>properties.bootstrap.servers</td>
+ <td>required</td>
+ <td style="word-wrap: break-word;">(none)</td>
+ <td>String</td>
+ <td>A list of host/port pairs to use for establishing the initial
connection to the Kafka cluster.</td>
+ </tr>
+ <tr>
+ <td>topic</td>
+ <td>optional</td>
+ <td style="word-wrap: break-word;">(none)</td>
+ <td>String</td>
+ <td>If this parameter is configured, all events will be sent to this
topic.</td>
+ </tr>
+ <tr>
+ <td>sink.add-tableId-to-header-enabled</td>
+ <td>optional</td>
+ <td style="word-wrap: break-word;">(none)</td>
+ <td>Boolean</td>
+ <td>If this parameter is true, a header with key of
'namespace','schemaName','tableName' will be added for each Kafka record.
Default value is false.</td>
+ </tr>
+ <tr>
+ <td>properties.*</td>
+ <td>optional</td>
+ <td style="word-wrap: break-word;">(none)</td>
+ <td>String</td>
+ <td>Pass options of Kafka table to pipeline，See <a
href="https://kafka.apache.org/28/documentation.html#consumerconfigs";>Kafka
consume options</a>. </td>
+ </tr>
+ <tr>
+ <td>sink.custom-header</td>
+ <td>optional</td>
+ <td style="word-wrap: break-word;">(none)</td>
+ <td>String</td>
+ <td>custom headers for each kafka record. Each header are separated by
',', separate key and value by ':'. For example, we can set headers like
'key1:value1,key2:value2'. </td>
+ </tr>
+ </tbody>
+</table>
+</div>
+
+Usage Notes
+--------
+
+* The written topic of Kafka will be `namespace.schemaName.tableName` string
of TableId，this can be changed using route function of pipeline.
+
+* If the written topic of Kafka is not existed, we will create one
automatically.
+
+Data Type Mapping
+----------------
+<div class="wy-table-responsive">
+<table class="colwidths-auto docutils">
+ <thead>
+ <tr>
+ <th class="text-left">CDC type</th>
+ <th class="text-left">JSON type</th>
+ <th class="text-left" style="width:60%;">NOTE</th>
+ </tr>
+ </thead>
+ <tbody>
+ <tr>
+ <td>TINYINT</td>
+ <td>TINYINT</td>
+ <td></td>
+ </tr>
+ <tr>
+ <td>SMALLINT</td>
+ <td>SMALLINT</td>
+ <td></td>
+ </tr>
+ <tr>
+ <td>INT</td>
+ <td>INT</td>
+ <td></td>
+ </tr>
+ <tr>
+ <td>BIGINT</td>
+ <td>BIGINT</td>
+ <td></td>
+ </tr>
+ <tr>
+ <td>FLOAT</td>
+ <td>FLOAT</td>
+ <td></td>
+ </tr>
+ <tr>
+ <td>DOUBLE</td>
+ <td>DOUBLE</td>
+ <td></td>
+ </tr>
+ <tr>
+ <td>DECIMAL(p, s)</td>
+ <td>DECIMAL(p, s)</td>
+ <td></td>
+ </tr>
+ <tr>
+ <td>BOOLEAN</td>
+ <td>BOOLEAN</td>
+ <td></td>
+ </tr>
+ <tr>
+ <td>DATE</td>
+ <td>DATE</td>
+ <td></td>
+ </tr>
+ <tr>
+ <td>TIMESTAMP</td>
+ <td>TIMESTAMP</td>
+ <td></td>
+ </tr>
+ <tr>
+ <td>TIMESTAMP_LTZ</td>
+ <td>TIMESTAMP_LTZ</td>
+ <td></td>
+ </tr>
+ <tr>
+ <td>CHAR(n)</td>
+ <td>CHAR(n)</td>
+ <td></td>
+ </tr>
+ <tr>
+ <td>VARCHAR(n)</td>
+ <td>VARCHAR(n)</td>
+ <td></td>
+ </tr>
+ </tbody>
+</table>
+</div>
+
+{{< top >}}
\ No newline at end of file
diff --git
a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-kafka/pom.xml
b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-kafka/pom.xml
new file mode 100644
index 000000000..0b0c6f687
--- /dev/null
+++
b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-kafka/pom.xml
@@ -0,0 +1,99 @@
+<?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";>
+ <parent>
+ <artifactId>flink-cdc-pipeline-connectors</artifactId>
+ <groupId>org.apache.flink</groupId>
+ <version>${revision}</version>
+ </parent>
+ <modelVersion>4.0.0</modelVersion>
+
+ <artifactId>flink-cdc-pipeline-connector-kafka</artifactId>
+
+ <properties>
+ </properties>
+
+ <dependencies>
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-connector-kafka</artifactId>
+ <version>3.0.2-${flink.major.version}</version>
+ </dependency>
+
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-json</artifactId>
+ <version>${flink.version}</version>
+ <scope>provided</scope>
+ </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.testcontainers</groupId>
+ <artifactId>kafka</artifactId>
+ <version>${testcontainers.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>
+ </dependencies>
+
+ <build>
+ <plugins>
+ <plugin>
+ <groupId>org.apache.maven.plugins</groupId>
+ <artifactId>maven-shade-plugin</artifactId>
+ <version>3.1.1</version>
+ <executions>
+ <execution>
+ <id>shade-flink</id>
+ <phase>package</phase>
+ <goals>
+ <goal>shade</goal>
+ </goals>
+ <configuration>
+ <shadeTestJar>false</shadeTestJar>
+ <relocations>
+ <relocation>
+ <pattern>org.apache.kafka</pattern>
+
<shadedPattern>org.apache.flink.cdc.connectors.kafka.shaded.org.apache.kafka</shadedPattern>
+ </relocation>
+ </relocations>
+ </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-kafka/src/main/java/org/apache/flink/cdc/connectors/kafka/json/ChangeLogJsonFormatFactory.java
b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-kafka/src/main/java/org/apache/flink/cdc/connectors/kafka/json/ChangeLogJsonFormatFactory.java
new file mode 100644
index 000000000..81c401804
--- /dev/null
+++
b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-kafka/src/main/java/org/apache/flink/cdc/connectors/kafka/json/ChangeLogJsonFormatFactory.java
@@ -0,0 +1,82 @@
+/*
+ * 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.kafka.json;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.api.common.serialization.SerializationSchema;
+import org.apache.flink.cdc.common.event.Event;
+import
org.apache.flink.cdc.connectors.kafka.json.canal.CanalJsonSerializationSchema;
+import
org.apache.flink.cdc.connectors.kafka.json.debezium.DebeziumJsonSerializationSchema;
+import org.apache.flink.configuration.ReadableConfig;
+import org.apache.flink.formats.common.TimestampFormat;
+import org.apache.flink.formats.json.JsonFormatOptions;
+import org.apache.flink.formats.json.JsonFormatOptionsUtil;
+
+import static
org.apache.flink.formats.json.JsonFormatOptions.ENCODE_DECIMAL_AS_PLAIN_NUMBER;
+import static
org.apache.flink.formats.json.debezium.DebeziumJsonFormatOptions.JSON_MAP_NULL_KEY_LITERAL;
+
+/**
+ * Format factory for providing configured instances of {@link
SerializationSchema} to convert
+ * {@link Event} to json.
+ */
+@Internal
+public class ChangeLogJsonFormatFactory {
+
+ /**
+ * Creates a configured instance of {@link SerializationSchema} to convert
{@link Event} to
+ * json.
+ *
+ * @param formatOptions The format options.
+ * @param type The type of json serialization.
+ * @return The configured instance of {@link SerializationSchema}.
+ */
+ public static SerializationSchema<Event> createSerializationSchema(
+ ReadableConfig formatOptions, JsonSerializationType type) {
+ TimestampFormat timestampFormat =
JsonFormatOptionsUtil.getTimestampFormat(formatOptions);
+ JsonFormatOptions.MapNullKeyMode mapNullKeyMode =
+ JsonFormatOptionsUtil.getMapNullKeyMode(formatOptions);
+ String mapNullKeyLiteral =
formatOptions.get(JSON_MAP_NULL_KEY_LITERAL);
+
+ final boolean encodeDecimalAsPlainNumber =
+ formatOptions.get(ENCODE_DECIMAL_AS_PLAIN_NUMBER);
+
+ switch (type) {
+ case DEBEZIUM_JSON:
+ {
+ return new DebeziumJsonSerializationSchema(
+ timestampFormat,
+ mapNullKeyMode,
+ mapNullKeyLiteral,
+ encodeDecimalAsPlainNumber);
+ }
+ case CANAL_JSON:
+ {
+ return new CanalJsonSerializationSchema(
+ timestampFormat,
+ mapNullKeyMode,
+ mapNullKeyLiteral,
+ encodeDecimalAsPlainNumber);
+ }
+ default:
+ {
+ throw new IllegalArgumentException(
+ "unSupport JsonSerializationType of " + type);
+ }
+ }
+ }
+}
diff --git
a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-kafka/src/main/java/org/apache/flink/cdc/connectors/kafka/json/JsonSerializationType.java
b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-kafka/src/main/java/org/apache/flink/cdc/connectors/kafka/json/JsonSerializationType.java
new file mode 100644
index 000000000..f7b5e0ffc
--- /dev/null
+++
b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-kafka/src/main/java/org/apache/flink/cdc/connectors/kafka/json/JsonSerializationType.java
@@ -0,0 +1,44 @@
+/*
+ * 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.kafka.json;
+
+import org.apache.flink.api.common.serialization.SerializationSchema;
+import org.apache.flink.cdc.common.event.Event;
+import
org.apache.flink.cdc.connectors.kafka.json.canal.CanalJsonSerializationSchema;
+import
org.apache.flink.cdc.connectors.kafka.json.debezium.DebeziumJsonSerializationSchema;
+
+/** type of {@link SerializationSchema} to serialize {@link Event} for kafka.
*/
+public enum JsonSerializationType {
+
+ /** Use {@link DebeziumJsonSerializationSchema} to serialize. */
+ DEBEZIUM_JSON("debezium-json"),
+
+ /** Use {@link CanalJsonSerializationSchema} to serialize. */
+ CANAL_JSON("canal-json");
+
+ private final String value;
+
+ JsonSerializationType(String value) {
+ this.value = value;
+ }
+
+ @Override
+ public String toString() {
+ return value;
+ }
+}
diff --git
a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-kafka/src/main/java/org/apache/flink/cdc/connectors/kafka/json/TableSchemaInfo.java
b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-kafka/src/main/java/org/apache/flink/cdc/connectors/kafka/json/TableSchemaInfo.java
new file mode 100644
index 000000000..a8480d77d
--- /dev/null
+++
b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-kafka/src/main/java/org/apache/flink/cdc/connectors/kafka/json/TableSchemaInfo.java
@@ -0,0 +1,163 @@
+/*
+ * 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.kafka.json;
+
+import org.apache.flink.api.common.serialization.SerializationSchema;
+import org.apache.flink.cdc.common.data.RecordData;
+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.table.data.DecimalData;
+import org.apache.flink.table.data.GenericRowData;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.data.TimestampData;
+import org.apache.flink.table.data.binary.BinaryStringData;
+
+import java.time.ZoneId;
+import java.time.ZonedDateTime;
+import java.util.ArrayList;
+import java.util.List;
+
+import static org.apache.flink.cdc.common.types.DataTypeChecks.getPrecision;
+import static org.apache.flink.cdc.common.types.DataTypeChecks.getScale;
+
+/** maintain the {@link SerializationSchema} of a specific {@link TableId}. */
+public class TableSchemaInfo {
+
+ private final Schema schema;
+
+ private final List<RecordData.FieldGetter> fieldGetters;
+
+ private final SerializationSchema<RowData> serializationSchema;
+
+ public TableSchemaInfo(
+ Schema schema, SerializationSchema<RowData> serializationSchema,
ZoneId zoneId) {
+ this.schema = schema;
+ this.serializationSchema = serializationSchema;
+ this.fieldGetters = createFieldGetters(schema, zoneId);
+ }
+
+ /** convert to {@link RowData}, which will be pass to serializationSchema.
*/
+ public RowData getRowDataFromRecordData(RecordData recordData) {
+ GenericRowData genericRowData = new
GenericRowData(recordData.getArity());
+ for (int i = 0; i < recordData.getArity(); i++) {
+ genericRowData.setField(i,
fieldGetters.get(i).getFieldOrNull(recordData));
+ }
+ return genericRowData;
+ }
+
+ private static List<RecordData.FieldGetter> createFieldGetters(Schema
schema, ZoneId zoneId) {
+ List<RecordData.FieldGetter> fieldGetters = new
ArrayList<>(schema.getColumns().size());
+ for (int i = 0; i < schema.getColumns().size(); i++) {
+
fieldGetters.add(createFieldGetter(schema.getColumns().get(i).getType(), i,
zoneId));
+ }
+ return fieldGetters;
+ }
+
+ private static RecordData.FieldGetter createFieldGetter(
+ DataType fieldType, int fieldPos, ZoneId zoneId) {
+ final RecordData.FieldGetter fieldGetter;
+ // ordered by type root definition
+ switch (fieldType.getTypeRoot()) {
+ case CHAR:
+ case VARCHAR:
+ fieldGetter =
+ record ->
+
BinaryStringData.fromString(record.getString(fieldPos).toString());
+ break;
+ case BOOLEAN:
+ fieldGetter = record -> record.getBoolean(fieldPos);
+ break;
+ case BINARY:
+ case VARBINARY:
+ fieldGetter = record -> record.getBinary(fieldPos);
+ break;
+ case DECIMAL:
+ final int decimalPrecision = getPrecision(fieldType);
+ final int decimalScale = getScale(fieldType);
+ fieldGetter =
+ record ->
+ DecimalData.fromBigDecimal(
+ record.getDecimal(fieldPos,
decimalPrecision, decimalScale)
+ .toBigDecimal(),
+ decimalPrecision,
+ decimalScale);
+ break;
+ case TINYINT:
+ fieldGetter = record -> record.getByte(fieldPos);
+ break;
+ case SMALLINT:
+ fieldGetter = record -> record.getShort(fieldPos);
+ break;
+ case INTEGER:
+ case DATE:
+ case TIME_WITHOUT_TIME_ZONE:
+ fieldGetter = record -> record.getInt(fieldPos);
+ break;
+ case BIGINT:
+ fieldGetter = record -> record.getLong(fieldPos);
+ break;
+ case FLOAT:
+ fieldGetter = record -> record.getFloat(fieldPos);
+ break;
+ case DOUBLE:
+ fieldGetter = record -> record.getDouble(fieldPos);
+ break;
+ case TIMESTAMP_WITHOUT_TIME_ZONE:
+ fieldGetter =
+ record ->
+ TimestampData.fromTimestamp(
+ record.getTimestamp(fieldPos,
getPrecision(fieldType))
+ .toTimestamp());
+ break;
+ case TIMESTAMP_WITH_TIME_ZONE:
+ case TIMESTAMP_WITH_LOCAL_TIME_ZONE:
+ fieldGetter =
+ record ->
+ TimestampData.fromLocalDateTime(
+ ZonedDateTime.ofInstant(
+
record.getLocalZonedTimestampData(
+
fieldPos,
+
getPrecision(fieldType))
+ .toInstant(),
+ zoneId)
+ .toLocalDateTime());
+ break;
+ default:
+ throw new IllegalArgumentException(
+ "don't support type of " + fieldType.getTypeRoot());
+ }
+ if (!fieldType.isNullable()) {
+ return fieldGetter;
+ }
+ return row -> {
+ if (row.isNullAt(fieldPos)) {
+ return null;
+ }
+ return fieldGetter.getFieldOrNull(row);
+ };
+ }
+
+ public Schema getSchema() {
+ return schema;
+ }
+
+ public SerializationSchema<RowData> getSerializationSchema() {
+ return serializationSchema;
+ }
+}
diff --git
a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-kafka/src/main/java/org/apache/flink/cdc/connectors/kafka/json/canal/CanalJsonSerializationSchema.java
b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-kafka/src/main/java/org/apache/flink/cdc/connectors/kafka/json/canal/CanalJsonSerializationSchema.java
new file mode 100644
index 000000000..eecd2b801
--- /dev/null
+++
b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-kafka/src/main/java/org/apache/flink/cdc/connectors/kafka/json/canal/CanalJsonSerializationSchema.java
@@ -0,0 +1,212 @@
+/*
+ * 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.kafka.json.canal;
+
+import org.apache.flink.api.common.serialization.SerializationSchema;
+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.SchemaChangeEvent;
+import org.apache.flink.cdc.common.event.TableId;
+import org.apache.flink.cdc.common.schema.Schema;
+import org.apache.flink.cdc.common.types.utils.DataTypeUtils;
+import org.apache.flink.cdc.common.utils.SchemaUtils;
+import org.apache.flink.cdc.connectors.kafka.json.TableSchemaInfo;
+import org.apache.flink.formats.common.TimestampFormat;
+import org.apache.flink.formats.json.JsonFormatOptions;
+import org.apache.flink.formats.json.JsonRowDataSerializationSchema;
+import org.apache.flink.table.api.DataTypes;
+import org.apache.flink.table.data.GenericArrayData;
+import org.apache.flink.table.data.GenericRowData;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.data.StringData;
+import org.apache.flink.table.types.DataType;
+import org.apache.flink.table.types.logical.LogicalType;
+import org.apache.flink.table.types.logical.RowType;
+
+import java.time.ZoneId;
+import java.util.HashMap;
+import java.util.Map;
+
+import static java.lang.String.format;
+import static
org.apache.flink.table.types.utils.TypeConversions.fromLogicalToDataType;
+
+/**
+ * Serialization schema that serializes an object of FlinkCDC pipeline
internal data structure
+ * {@link Event} into a Canal JSON bytes.
+ *
+ * @see <a href="https://github.com/alibaba/canal";>Alibaba Canal</a>
+ */
+public class CanalJsonSerializationSchema implements
SerializationSchema<Event> {
+
+ private static final long serialVersionUID = 1L;
+
+ private static final StringData OP_INSERT =
StringData.fromString("INSERT");
+ private static final StringData OP_DELETE =
StringData.fromString("DELETE");
+ private static final StringData OP_UPDATE =
StringData.fromString("UPDATE");
+
+ private transient GenericRowData reuseGenericRowData;
+
+ /** The serializer to serialize Canal JSON data. */
+ private final Map<TableId, TableSchemaInfo> jsonSerializers;
+
+ private final TimestampFormat timestampFormat;
+
+ private final JsonFormatOptions.MapNullKeyMode mapNullKeyMode;
+
+ private final String mapNullKeyLiteral;
+
+ private final boolean encodeDecimalAsPlainNumber;
+
+ private final ZoneId zoneId;
+
+ private InitializationContext context;
+
+ public CanalJsonSerializationSchema(
+ TimestampFormat timestampFormat,
+ JsonFormatOptions.MapNullKeyMode mapNullKeyMode,
+ String mapNullKeyLiteral,
+ boolean encodeDecimalAsPlainNumber) {
+ this.timestampFormat = timestampFormat;
+ this.mapNullKeyMode = mapNullKeyMode;
+ this.mapNullKeyLiteral = mapNullKeyLiteral;
+ this.encodeDecimalAsPlainNumber = encodeDecimalAsPlainNumber;
+ this.zoneId = ZoneId.systemDefault();
+ jsonSerializers = new HashMap<>();
+ }
+
+ @Override
+ public void open(InitializationContext context) {
+ this.context = context;
+ reuseGenericRowData = new GenericRowData(3);
+ }
+
+ @Override
+ public byte[] serialize(Event event) {
+ if (event instanceof SchemaChangeEvent) {
+ Schema schema;
+ SchemaChangeEvent schemaChangeEvent = (SchemaChangeEvent) event;
+ if (event instanceof CreateTableEvent) {
+ CreateTableEvent createTableEvent = (CreateTableEvent) event;
+ schema = createTableEvent.getSchema();
+ } else {
+ schema =
+ SchemaUtils.applySchemaChangeEvent(
+
jsonSerializers.get(schemaChangeEvent.tableId()).getSchema(),
+ schemaChangeEvent);
+ }
+ LogicalType rowType =
+
DataTypeUtils.toFlinkDataType(schema.toRowDataType()).getLogicalType();
+ JsonRowDataSerializationSchema jsonSerializer =
+ new JsonRowDataSerializationSchema(
+ createJsonRowType(fromLogicalToDataType(rowType)),
+ timestampFormat,
+ mapNullKeyMode,
+ mapNullKeyLiteral,
+ encodeDecimalAsPlainNumber);
+ try {
+ jsonSerializer.open(context);
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+ jsonSerializers.put(
+ schemaChangeEvent.tableId(),
+ new TableSchemaInfo(schema, jsonSerializer, zoneId));
+ return null;
+ }
+
+ DataChangeEvent dataChangeEvent = (DataChangeEvent) event;
+ try {
+ switch (dataChangeEvent.op()) {
+ case INSERT:
+ reuseGenericRowData.setField(0, null);
+ reuseGenericRowData.setField(
+ 1,
+ new GenericArrayData(
+ new RowData[] {
+ jsonSerializers
+ .get(dataChangeEvent.tableId())
+
.getRowDataFromRecordData((dataChangeEvent.after()))
+ }));
+ reuseGenericRowData.setField(2, OP_INSERT);
+ return jsonSerializers
+ .get(dataChangeEvent.tableId())
+ .getSerializationSchema()
+ .serialize(reuseGenericRowData);
+ case DELETE:
+ reuseGenericRowData.setField(
+ 0,
+ new GenericArrayData(
+ new RowData[] {
+ jsonSerializers
+ .get(dataChangeEvent.tableId())
+ .getRowDataFromRecordData(
+
(dataChangeEvent.before()))
+ }));
+ reuseGenericRowData.setField(1, null);
+ reuseGenericRowData.setField(2, OP_DELETE);
+ return jsonSerializers
+ .get(dataChangeEvent.tableId())
+ .getSerializationSchema()
+ .serialize(reuseGenericRowData);
+ case UPDATE:
+ case REPLACE:
+ reuseGenericRowData.setField(
+ 0,
+ new GenericArrayData(
+ new RowData[] {
+ jsonSerializers
+ .get(dataChangeEvent.tableId())
+ .getRowDataFromRecordData(
+
(dataChangeEvent.before()))
+ }));
+ reuseGenericRowData.setField(
+ 1,
+ new GenericArrayData(
+ new RowData[] {
+ jsonSerializers
+ .get(dataChangeEvent.tableId())
+
.getRowDataFromRecordData((dataChangeEvent.after()))
+ }));
+ reuseGenericRowData.setField(2, OP_UPDATE);
+ return jsonSerializers
+ .get(dataChangeEvent.tableId())
+ .getSerializationSchema()
+ .serialize(reuseGenericRowData);
+ default:
+ throw new UnsupportedOperationException(
+ format(
+ "Unsupported operation '%s' for
OperationType.",
+ dataChangeEvent.op()));
+ }
+ } catch (Throwable t) {
+ throw new RuntimeException(format("Could not serialize event
'%s'.", event), t);
+ }
+ }
+
+ private static RowType createJsonRowType(DataType databaseSchema) {
+ // Canal JSON contains other information, e.g. "database", "ts"
+ // but we don't need them
+ return (RowType)
+ DataTypes.ROW(
+ DataTypes.FIELD("old",
DataTypes.ARRAY(databaseSchema)),
+ DataTypes.FIELD("data",
DataTypes.ARRAY(databaseSchema)),
+ DataTypes.FIELD("type", DataTypes.STRING()))
+ .getLogicalType();
+ }
+}
diff --git
a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-kafka/src/main/java/org/apache/flink/cdc/connectors/kafka/json/debezium/DebeziumJsonSerializationSchema.java
b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-kafka/src/main/java/org/apache/flink/cdc/connectors/kafka/json/debezium/DebeziumJsonSerializationSchema.java
new file mode 100644
index 000000000..3fa1d4ce1
--- /dev/null
+++
b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-kafka/src/main/java/org/apache/flink/cdc/connectors/kafka/json/debezium/DebeziumJsonSerializationSchema.java
@@ -0,0 +1,197 @@
+/*
+ * 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.kafka.json.debezium;
+
+import org.apache.flink.api.common.serialization.SerializationSchema;
+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.SchemaChangeEvent;
+import org.apache.flink.cdc.common.event.TableId;
+import org.apache.flink.cdc.common.schema.Schema;
+import org.apache.flink.cdc.common.types.utils.DataTypeUtils;
+import org.apache.flink.cdc.common.utils.SchemaUtils;
+import org.apache.flink.cdc.connectors.kafka.json.TableSchemaInfo;
+import org.apache.flink.formats.common.TimestampFormat;
+import org.apache.flink.formats.json.JsonFormatOptions;
+import org.apache.flink.formats.json.JsonRowDataSerializationSchema;
+import org.apache.flink.table.api.DataTypes;
+import org.apache.flink.table.data.GenericRowData;
+import org.apache.flink.table.data.StringData;
+import org.apache.flink.table.types.DataType;
+import org.apache.flink.table.types.logical.LogicalType;
+import org.apache.flink.table.types.logical.RowType;
+
+import java.time.ZoneId;
+import java.util.HashMap;
+import java.util.Map;
+
+import static java.lang.String.format;
+import static
org.apache.flink.table.types.utils.TypeConversions.fromLogicalToDataType;
+
+/**
+ * Serialization schema from FlinkCDC pipeline internal data structure {@link
Event} to Debezium
+ * JSON.
+ *
+ * @see <a href="https://debezium.io/";>Debezium</a>
+ */
+public class DebeziumJsonSerializationSchema implements
SerializationSchema<Event> {
+ private static final long serialVersionUID = 1L;
+
+ private static final StringData OP_INSERT = StringData.fromString("c"); //
insert
+ private static final StringData OP_DELETE = StringData.fromString("d"); //
delete
+ private static final StringData OP_UPDATE = StringData.fromString("u"); //
update
+
+ /**
+ * A map of {@link TableId} and its {@link SerializationSchema} to
serialize Debezium JSON data.
+ */
+ private final Map<TableId, TableSchemaInfo> jsonSerializers;
+
+ private transient GenericRowData reuseGenericRowData;
+
+ private final TimestampFormat timestampFormat;
+
+ private final JsonFormatOptions.MapNullKeyMode mapNullKeyMode;
+
+ private final String mapNullKeyLiteral;
+
+ private final boolean encodeDecimalAsPlainNumber;
+
+ private final ZoneId zoneId;
+
+ private InitializationContext context;
+
+ public DebeziumJsonSerializationSchema(
+ TimestampFormat timestampFormat,
+ JsonFormatOptions.MapNullKeyMode mapNullKeyMode,
+ String mapNullKeyLiteral,
+ boolean encodeDecimalAsPlainNumber) {
+ this.timestampFormat = timestampFormat;
+ this.mapNullKeyMode = mapNullKeyMode;
+ this.mapNullKeyLiteral = mapNullKeyLiteral;
+ this.encodeDecimalAsPlainNumber = encodeDecimalAsPlainNumber;
+ this.zoneId = ZoneId.systemDefault();
+ jsonSerializers = new HashMap<>();
+ }
+
+ @Override
+ public void open(InitializationContext context) {
+ reuseGenericRowData = new GenericRowData(3);
+ this.context = context;
+ }
+
+ @Override
+ public byte[] serialize(Event event) {
+ if (event instanceof SchemaChangeEvent) {
+ Schema schema;
+ SchemaChangeEvent schemaChangeEvent = (SchemaChangeEvent) event;
+ if (event instanceof CreateTableEvent) {
+ CreateTableEvent createTableEvent = (CreateTableEvent) event;
+ schema = createTableEvent.getSchema();
+ } else {
+ schema =
+ SchemaUtils.applySchemaChangeEvent(
+
jsonSerializers.get(schemaChangeEvent.tableId()).getSchema(),
+ schemaChangeEvent);
+ }
+ LogicalType rowType =
+
DataTypeUtils.toFlinkDataType(schema.toRowDataType()).getLogicalType();
+ JsonRowDataSerializationSchema jsonSerializer =
+ new JsonRowDataSerializationSchema(
+ createJsonRowType(fromLogicalToDataType(rowType)),
+ timestampFormat,
+ mapNullKeyMode,
+ mapNullKeyLiteral,
+ encodeDecimalAsPlainNumber);
+ try {
+ jsonSerializer.open(context);
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+ jsonSerializers.put(
+ schemaChangeEvent.tableId(),
+ new TableSchemaInfo(schema, jsonSerializer, zoneId));
+ return null;
+ }
+
+ DataChangeEvent dataChangeEvent = (DataChangeEvent) event;
+ try {
+ switch (dataChangeEvent.op()) {
+ case INSERT:
+ reuseGenericRowData.setField(0, null);
+ reuseGenericRowData.setField(
+ 1,
+ jsonSerializers
+ .get(dataChangeEvent.tableId())
+
.getRowDataFromRecordData(dataChangeEvent.after()));
+ reuseGenericRowData.setField(2, OP_INSERT);
+ return jsonSerializers
+ .get(dataChangeEvent.tableId())
+ .getSerializationSchema()
+ .serialize(reuseGenericRowData);
+ case DELETE:
+ reuseGenericRowData.setField(
+ 0,
+ jsonSerializers
+ .get(dataChangeEvent.tableId())
+
.getRowDataFromRecordData(dataChangeEvent.before()));
+ reuseGenericRowData.setField(1, null);
+ reuseGenericRowData.setField(2, OP_DELETE);
+ return jsonSerializers
+ .get(dataChangeEvent.tableId())
+ .getSerializationSchema()
+ .serialize(reuseGenericRowData);
+ case UPDATE:
+ case REPLACE:
+ reuseGenericRowData.setField(
+ 0,
+ jsonSerializers
+ .get(dataChangeEvent.tableId())
+
.getRowDataFromRecordData(dataChangeEvent.before()));
+ reuseGenericRowData.setField(
+ 1,
+ jsonSerializers
+ .get(dataChangeEvent.tableId())
+
.getRowDataFromRecordData(dataChangeEvent.after()));
+ reuseGenericRowData.setField(2, OP_UPDATE);
+ return jsonSerializers
+ .get(dataChangeEvent.tableId())
+ .getSerializationSchema()
+ .serialize(reuseGenericRowData);
+ default:
+ throw new UnsupportedOperationException(
+ format(
+ "Unsupported operation '%s' for
OperationType.",
+ dataChangeEvent.op()));
+ }
+ } catch (Throwable t) {
+ throw new RuntimeException(format("Could not serialize event
'%s'.", event), t);
+ }
+ }
+
+ private static RowType createJsonRowType(DataType databaseSchema) {
+ // Debezium JSON contains some other information, e.g. "source",
"ts_ms"
+ // but we don't need them.
+ return (RowType)
+ DataTypes.ROW(
+ DataTypes.FIELD("before", databaseSchema),
+ DataTypes.FIELD("after", databaseSchema),
+ DataTypes.FIELD("op", DataTypes.STRING()))
+ .getLogicalType();
+ }
+}
diff --git
a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-kafka/src/main/java/org/apache/flink/cdc/connectors/kafka/sink/KafkaDataSink.java
b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-kafka/src/main/java/org/apache/flink/cdc/connectors/kafka/sink/KafkaDataSink.java
new file mode 100644
index 000000000..947bde1a8
--- /dev/null
+++
b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-kafka/src/main/java/org/apache/flink/cdc/connectors/kafka/sink/KafkaDataSink.java
@@ -0,0 +1,100 @@
+/*
+ * 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.kafka.sink;
+
+import org.apache.flink.api.common.serialization.SerializationSchema;
+import org.apache.flink.cdc.common.event.Event;
+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.connector.base.DeliveryGuarantee;
+import org.apache.flink.connector.kafka.sink.KafkaSink;
+import org.apache.flink.connector.kafka.sink.KafkaSinkBuilder;
+import
org.apache.flink.streaming.connectors.kafka.partitioner.FlinkKafkaPartitioner;
+
+import org.apache.kafka.clients.producer.ProducerConfig;
+
+import java.time.ZoneId;
+import java.util.Properties;
+
+/** A {@link DataSink} for "Kafka" connector. */
+public class KafkaDataSink implements DataSink {
+
+ final Properties kafkaProperties;
+
+ final DeliveryGuarantee deliveryGuarantee;
+
+ final FlinkKafkaPartitioner<Event> partitioner;
+
+ final ZoneId zoneId;
+
+ final SerializationSchema<Event> valueSerialization;
+
+ final String topic;
+
+ final boolean addTableToHeaderEnabled;
+
+ final String customHeaders;
+
+ public KafkaDataSink(
+ DeliveryGuarantee deliveryGuarantee,
+ Properties kafkaProperties,
+ FlinkKafkaPartitioner<Event> partitioner,
+ ZoneId zoneId,
+ SerializationSchema<Event> valueSerialization,
+ String topic,
+ boolean addTableToHeaderEnabled,
+ String customHeaders) {
+ this.deliveryGuarantee = deliveryGuarantee;
+ this.kafkaProperties = kafkaProperties;
+ this.partitioner = partitioner;
+ this.zoneId = zoneId;
+ this.valueSerialization = valueSerialization;
+ this.topic = topic;
+ this.addTableToHeaderEnabled = addTableToHeaderEnabled;
+ this.customHeaders = customHeaders;
+ }
+
+ @Override
+ public EventSinkProvider getEventSinkProvider() {
+ final KafkaSinkBuilder<Event> sinkBuilder = KafkaSink.builder();
+ return FlinkSinkProvider.of(
+ sinkBuilder
+ .setDeliveryGuarantee(deliveryGuarantee)
+ .setBootstrapServers(
+ kafkaProperties
+
.get(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG)
+ .toString())
+ .setKafkaProducerConfig(kafkaProperties)
+ .setRecordSerializer(
+ new PipelineKafkaRecordSerializationSchema(
+ partitioner,
+ valueSerialization,
+ topic,
+ addTableToHeaderEnabled,
+ customHeaders))
+ .build());
+ }
+
+ @Override
+ public MetadataApplier getMetadataApplier() {
+ // simply do nothing here because Kafka do not maintain the schemas.
+ return schemaChangeEvent -> {};
+ }
+}
diff --git
a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-kafka/src/main/java/org/apache/flink/cdc/connectors/kafka/sink/KafkaDataSinkFactory.java
b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-kafka/src/main/java/org/apache/flink/cdc/connectors/kafka/sink/KafkaDataSinkFactory.java
new file mode 100644
index 000000000..9b117243f
--- /dev/null
+++
b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-kafka/src/main/java/org/apache/flink/cdc/connectors/kafka/sink/KafkaDataSinkFactory.java
@@ -0,0 +1,111 @@
+/*
+ * 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.kafka.sink;
+
+import org.apache.flink.api.common.serialization.SerializationSchema;
+import org.apache.flink.cdc.common.configuration.ConfigOption;
+import org.apache.flink.cdc.common.event.Event;
+import org.apache.flink.cdc.common.factories.DataSinkFactory;
+import org.apache.flink.cdc.common.pipeline.PipelineOptions;
+import org.apache.flink.cdc.common.sink.DataSink;
+import org.apache.flink.cdc.connectors.kafka.json.ChangeLogJsonFormatFactory;
+import org.apache.flink.cdc.connectors.kafka.json.JsonSerializationType;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.connector.base.DeliveryGuarantee;
+import
org.apache.flink.streaming.connectors.kafka.partitioner.FlinkFixedPartitioner;
+
+import java.time.ZoneId;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Properties;
+import java.util.Set;
+
+import static
org.apache.flink.cdc.connectors.kafka.sink.KafkaDataSinkOptions.PROPERTIES_PREFIX;
+
+/** A dummy {@link DataSinkFactory} to create {@link KafkaDataSink}. */
+public class KafkaDataSinkFactory implements DataSinkFactory {
+
+ public static final String IDENTIFIER = "kafka";
+
+ @Override
+ public DataSink createDataSink(Context context) {
+ Configuration configuration =
+
Configuration.fromMap(context.getFactoryConfiguration().toMap());
+ DeliveryGuarantee deliveryGuarantee =
+
context.getFactoryConfiguration().get(KafkaDataSinkOptions.DELIVERY_GUARANTEE);
+ ZoneId zoneId = ZoneId.systemDefault();
+ if (!Objects.equals(
+
context.getPipelineConfiguration().get(PipelineOptions.PIPELINE_LOCAL_TIME_ZONE),
+ PipelineOptions.PIPELINE_LOCAL_TIME_ZONE.defaultValue())) {
+ zoneId =
+ ZoneId.of(
+ context.getPipelineConfiguration()
+
.get(PipelineOptions.PIPELINE_LOCAL_TIME_ZONE));
+ }
+ JsonSerializationType jsonSerializationType =
+
context.getFactoryConfiguration().get(KafkaDataSinkOptions.VALUE_FORMAT);
+ SerializationSchema<Event> valueSerialization =
+ ChangeLogJsonFormatFactory.createSerializationSchema(
+ configuration, jsonSerializationType);
+ final Properties kafkaProperties = new Properties();
+ Map<String, String> allOptions =
context.getFactoryConfiguration().toMap();
+ allOptions.keySet().stream()
+ .filter(key -> key.startsWith(PROPERTIES_PREFIX))
+ .forEach(
+ key -> {
+ final String value = allOptions.get(key);
+ final String subKey =
key.substring((PROPERTIES_PREFIX).length());
+ kafkaProperties.put(subKey, value);
+ });
+ String topic =
context.getFactoryConfiguration().get(KafkaDataSinkOptions.TOPIC);
+ boolean addTableToHeaderEnabled =
+ context.getFactoryConfiguration()
+
.get(KafkaDataSinkOptions.SINK_ADD_TABLEID_TO_HEADER_ENABLED);
+ String customHeaders =
+
context.getFactoryConfiguration().get(KafkaDataSinkOptions.SINK_CUSTOM_HEADER);
+ return new KafkaDataSink(
+ deliveryGuarantee,
+ kafkaProperties,
+ new FlinkFixedPartitioner<>(),
+ zoneId,
+ valueSerialization,
+ topic,
+ addTableToHeaderEnabled,
+ customHeaders);
+ }
+
+ @Override
+ public String identifier() {
+ return IDENTIFIER;
+ }
+
+ @Override
+ public Set<ConfigOption<?>> requiredOptions() {
+ return null;
+ }
+
+ @Override
+ public Set<ConfigOption<?>> optionalOptions() {
+ Set<ConfigOption<?>> options = new HashSet<>();
+ options.add(KafkaDataSinkOptions.DELIVERY_GUARANTEE);
+ options.add(KafkaDataSinkOptions.TOPIC);
+ options.add(KafkaDataSinkOptions.SINK_ADD_TABLEID_TO_HEADER_ENABLED);
+ return options;
+ }
+}
diff --git
a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-kafka/src/main/java/org/apache/flink/cdc/connectors/kafka/sink/KafkaDataSinkOptions.java
b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-kafka/src/main/java/org/apache/flink/cdc/connectors/kafka/sink/KafkaDataSinkOptions.java
new file mode 100644
index 000000000..e55e149a9
--- /dev/null
+++
b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-kafka/src/main/java/org/apache/flink/cdc/connectors/kafka/sink/KafkaDataSinkOptions.java
@@ -0,0 +1,66 @@
+/*
+ * 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.kafka.sink;
+
+import org.apache.flink.cdc.common.configuration.ConfigOption;
+import org.apache.flink.cdc.connectors.kafka.json.JsonSerializationType;
+import org.apache.flink.connector.base.DeliveryGuarantee;
+
+import static org.apache.flink.cdc.common.configuration.ConfigOptions.key;
+
+/** Options for {@link KafkaDataSinkOptions}. */
+public class KafkaDataSinkOptions {
+
+ // Prefix for Kafka specific properties.
+ public static final String PROPERTIES_PREFIX = "properties.";
+
+ public static final ConfigOption<DeliveryGuarantee> DELIVERY_GUARANTEE =
+ key("sink.delivery-guarantee")
+ .enumType(DeliveryGuarantee.class)
+ .defaultValue(DeliveryGuarantee.AT_LEAST_ONCE)
+ .withDescription("Optional delivery guarantee when
committing.");
+
+ public static final ConfigOption<JsonSerializationType> VALUE_FORMAT =
+ key("value.format")
+ .enumType(JsonSerializationType.class)
+ .defaultValue(JsonSerializationType.DEBEZIUM_JSON)
+ .withDescription(
+ "Defines the format identifier for encoding value
data, "
+ + "available options are `debezium-json`
and `canal-json`, default option is `debezium-json`.");
+
+ public static final ConfigOption<String> TOPIC =
+ key("topic")
+ .stringType()
+ .noDefaultValue()
+ .withDescription(
+ "Optional. If this parameter is configured, all
events will be sent to this topic.");
+
+ public static final ConfigOption<Boolean>
SINK_ADD_TABLEID_TO_HEADER_ENABLED =
+ key("sink.add-tableId-to-header-enabled")
+ .booleanType()
+ .defaultValue(false)
+ .withDescription(
+ "Optional. If this parameter is configured, a
header with key of 'namespace','schemaName','tableName' will be added for each
Kafka record.");
+
+ public static final ConfigOption<String> SINK_CUSTOM_HEADER =
+ key("sink.custom-header")
+ .stringType()
+ .defaultValue("")
+ .withDescription(
+ "custom headers for each kafka record. Each header
are separated by ',', separate key and value by ':'. For example, we can set
headers like 'key1:value1,key2:value2'.");
+}
diff --git
a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-kafka/src/main/java/org/apache/flink/cdc/connectors/kafka/sink/PipelineKafkaRecordSerializationSchema.java
b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-kafka/src/main/java/org/apache/flink/cdc/connectors/kafka/sink/PipelineKafkaRecordSerializationSchema.java
new file mode 100644
index 000000000..bb27753d9
--- /dev/null
+++
b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-kafka/src/main/java/org/apache/flink/cdc/connectors/kafka/sink/PipelineKafkaRecordSerializationSchema.java
@@ -0,0 +1,157 @@
+/*
+ * 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.kafka.sink;
+
+import org.apache.flink.api.common.serialization.SerializationSchema;
+import org.apache.flink.cdc.common.event.ChangeEvent;
+import org.apache.flink.cdc.common.event.Event;
+import org.apache.flink.cdc.common.event.SchemaChangeEvent;
+import org.apache.flink.cdc.common.event.TableId;
+import org.apache.flink.connector.kafka.sink.KafkaRecordSerializationSchema;
+import
org.apache.flink.streaming.connectors.kafka.partitioner.FlinkKafkaPartitioner;
+
+import org.apache.kafka.clients.producer.ProducerRecord;
+import org.apache.kafka.common.header.internals.RecordHeader;
+import org.apache.kafka.common.header.internals.RecordHeaders;
+
+import javax.annotation.Nullable;
+
+import java.util.HashMap;
+import java.util.Map;
+
+import static java.nio.charset.StandardCharsets.UTF_8;
+import static org.apache.flink.util.Preconditions.checkNotNull;
+
+/**
+ * A {@link KafkaRecordSerializationSchema} to serialize {@link Event}.
+ *
+ * <p>The topic to be sent is the string value of {@link TableId}.
+ *
+ * <p>the key of {@link ProducerRecord} is null as we don't need to upsert
Kafka.
+ */
+public class PipelineKafkaRecordSerializationSchema
+ implements KafkaRecordSerializationSchema<Event> {
+ private final FlinkKafkaPartitioner<Event> partitioner;
+ private final SerializationSchema<Event> valueSerialization;
+
+ private final String unifiedTopic;
+
+ private final boolean addTableToHeaderEnabled;
+
+ // key value pairs to be put into Kafka Record Header.
+ public final Map<String, String> customHeaders;
+
+ public static final String NAMESPACE_HEADER_KEY = "namespace";
+
+ public static final String SCHEMA_NAME_HEADER_KEY = "schemaName";
+
+ public static final String TABLE_NAME_HEADER_KEY = "tableName";
+
+ PipelineKafkaRecordSerializationSchema(
+ @Nullable FlinkKafkaPartitioner<Event> partitioner,
+ SerializationSchema<Event> valueSerialization,
+ String unifiedTopic,
+ boolean addTableToHeaderEnabled,
+ String customHeaderString) {
+ this.partitioner = partitioner;
+ this.valueSerialization = checkNotNull(valueSerialization);
+ this.unifiedTopic = unifiedTopic;
+ this.addTableToHeaderEnabled = addTableToHeaderEnabled;
+ customHeaders = new HashMap<>();
+ if (!customHeaderString.isEmpty()) {
+ for (String tables : customHeaderString.split(";")) {
+ String[] splits = tables.split(":");
+ if (splits.length == 2) {
+ String headerKey = splits[0].trim();
+ String headerValue = splits[1].trim();
+ customHeaders.put(headerKey, headerValue);
+ } else {
+ throw new IllegalArgumentException(
+ KafkaDataSinkOptions.SINK_CUSTOM_HEADER
+ + " is malformed, please refer to the
documents");
+ }
+ }
+ }
+ }
+
+ @Override
+ public ProducerRecord<byte[], byte[]> serialize(
+ Event event, KafkaSinkContext context, Long timestamp) {
+ ChangeEvent changeEvent = (ChangeEvent) event;
+ final byte[] valueSerialized = valueSerialization.serialize(event);
+ if (event instanceof SchemaChangeEvent) {
+ // skip sending SchemaChangeEvent.
+ return null;
+ }
+ String topic = unifiedTopic == null ? changeEvent.tableId().toString()
: unifiedTopic;
+ RecordHeaders recordHeaders = new RecordHeaders();
+ if (addTableToHeaderEnabled) {
+ String namespace =
+ changeEvent.tableId().getNamespace() == null
+ ? ""
+ : changeEvent.tableId().getNamespace();
+ recordHeaders.add(new RecordHeader(NAMESPACE_HEADER_KEY,
namespace.getBytes(UTF_8)));
+ String schemaName =
+ changeEvent.tableId().getSchemaName() == null
+ ? ""
+ : changeEvent.tableId().getSchemaName();
+ recordHeaders.add(new RecordHeader(SCHEMA_NAME_HEADER_KEY,
schemaName.getBytes(UTF_8)));
+ String tableName = changeEvent.tableId().getTableName();
+ recordHeaders.add(new RecordHeader(TABLE_NAME_HEADER_KEY,
tableName.getBytes(UTF_8)));
+ }
+ if (!this.customHeaders.isEmpty()) {
+ for (Map.Entry<String, String> entry : customHeaders.entrySet()) {
+ recordHeaders.add(
+ new RecordHeader(entry.getKey(),
entry.getValue().getBytes(UTF_8)));
+ }
+ }
+ return new ProducerRecord<>(
+ topic,
+ extractPartition(
+ changeEvent, valueSerialized,
context.getPartitionsForTopic(topic)),
+ null,
+ null,
+ valueSerialized,
+ recordHeaders);
+ }
+
+ @Override
+ public void open(
+ SerializationSchema.InitializationContext context,
KafkaSinkContext sinkContext)
+ throws Exception {
+ if (partitioner != null) {
+ partitioner.open(
+ sinkContext.getParallelInstanceId(),
+ sinkContext.getNumberOfParallelInstances());
+ }
+ valueSerialization.open(context);
+ }
+
+ private Integer extractPartition(
+ ChangeEvent changeEvent, byte[] valueSerialized, int[] partitions)
{
+ if (partitioner != null) {
+ return partitioner.partition(
+ changeEvent,
+ null,
+ valueSerialized,
+ changeEvent.tableId().toString(),
+ partitions);
+ }
+ return null;
+ }
+}
diff --git
a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-kafka/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-kafka/src/main/resources/META-INF/services/org.apache.flink.cdc.common.factories.Factory
new file mode 100644
index 000000000..fa13cd600
--- /dev/null
+++
b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-kafka/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.kafka.sink.KafkaDataSinkFactory
diff --git
a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-kafka/src/test/java/org/apache/flink/cdc/connectors/kafka/json/MockInitializationContext.java
b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-kafka/src/test/java/org/apache/flink/cdc/connectors/kafka/json/MockInitializationContext.java
new file mode 100644
index 000000000..deb60064a
--- /dev/null
+++
b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-kafka/src/test/java/org/apache/flink/cdc/connectors/kafka/json/MockInitializationContext.java
@@ -0,0 +1,38 @@
+/*
+ * 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.kafka.json;
+
+import org.apache.flink.api.common.serialization.SerializationSchema;
+import org.apache.flink.metrics.MetricGroup;
+import org.apache.flink.metrics.groups.UnregisteredMetricsGroup;
+import org.apache.flink.util.SimpleUserCodeClassLoader;
+import org.apache.flink.util.UserCodeClassLoader;
+
+/** A mock context for serialization schema testing. */
+public class MockInitializationContext implements
SerializationSchema.InitializationContext {
+
+ @Override
+ public MetricGroup getMetricGroup() {
+ return new UnregisteredMetricsGroup();
+ }
+
+ @Override
+ public UserCodeClassLoader getUserCodeClassLoader() {
+ return
SimpleUserCodeClassLoader.create(MockInitializationContext.class.getClassLoader());
+ }
+}
diff --git
a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-kafka/src/test/java/org/apache/flink/cdc/connectors/kafka/json/TableSchemaInfoTest.java
b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-kafka/src/test/java/org/apache/flink/cdc/connectors/kafka/json/TableSchemaInfoTest.java
new file mode 100644
index 000000000..73f8ad64d
--- /dev/null
+++
b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-kafka/src/test/java/org/apache/flink/cdc/connectors/kafka/json/TableSchemaInfoTest.java
@@ -0,0 +1,164 @@
+/*
+ * 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.kafka.json;
+
+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.BinaryRecordData;
+import org.apache.flink.cdc.common.data.binary.BinaryStringData;
+import org.apache.flink.cdc.common.schema.Schema;
+import org.apache.flink.cdc.common.types.DataType;
+import org.apache.flink.cdc.runtime.typeutils.BinaryRecordDataGenerator;
+import org.apache.flink.table.data.GenericRowData;
+import org.apache.flink.types.RowKind;
+
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+import java.math.BigDecimal;
+import java.sql.Timestamp;
+import java.time.Instant;
+import java.time.ZoneId;
+
+/** Tests for {@link TableSchemaInfo}. */
+public class TableSchemaInfoTest {
+
+ @Test
+ public void testGetRowDataFromRecordData() {
+ Schema schema =
+ Schema.newBuilder()
+ .physicalColumn(
+ "col1",
+
org.apache.flink.cdc.common.types.DataTypes.STRING().notNull())
+ .physicalColumn(
+ "boolean",
org.apache.flink.cdc.common.types.DataTypes.BOOLEAN())
+ .physicalColumn(
+ "binary",
org.apache.flink.cdc.common.types.DataTypes.BINARY(3))
+ .physicalColumn(
+ "varbinary",
+
org.apache.flink.cdc.common.types.DataTypes.VARBINARY(10))
+ .physicalColumn(
+ "bytes",
org.apache.flink.cdc.common.types.DataTypes.BYTES())
+ .physicalColumn(
+ "tinyint",
org.apache.flink.cdc.common.types.DataTypes.TINYINT())
+ .physicalColumn(
+ "smallint",
org.apache.flink.cdc.common.types.DataTypes.SMALLINT())
+ .physicalColumn("int",
org.apache.flink.cdc.common.types.DataTypes.INT())
+ .physicalColumn(
+ "big_int",
org.apache.flink.cdc.common.types.DataTypes.BIGINT())
+ .physicalColumn(
+ "float",
org.apache.flink.cdc.common.types.DataTypes.FLOAT())
+ .physicalColumn(
+ "double",
org.apache.flink.cdc.common.types.DataTypes.DOUBLE())
+ .physicalColumn(
+ "decimal",
+
org.apache.flink.cdc.common.types.DataTypes.DECIMAL(6, 3))
+ .physicalColumn("char",
org.apache.flink.cdc.common.types.DataTypes.CHAR(5))
+ .physicalColumn(
+ "varchar",
org.apache.flink.cdc.common.types.DataTypes.VARCHAR(10))
+ .physicalColumn(
+ "string",
org.apache.flink.cdc.common.types.DataTypes.STRING())
+ .physicalColumn("date",
org.apache.flink.cdc.common.types.DataTypes.DATE())
+ .physicalColumn("time",
org.apache.flink.cdc.common.types.DataTypes.TIME())
+ .physicalColumn(
+ "time_with_precision",
+
org.apache.flink.cdc.common.types.DataTypes.TIME(6))
+ .physicalColumn(
+ "timestamp",
+
org.apache.flink.cdc.common.types.DataTypes.TIMESTAMP())
+ .physicalColumn(
+ "timestamp_with_precision",
+
org.apache.flink.cdc.common.types.DataTypes.TIMESTAMP(3))
+ .physicalColumn(
+ "timestamp_ltz",
+
org.apache.flink.cdc.common.types.DataTypes.TIMESTAMP_LTZ())
+ .physicalColumn(
+ "timestamp_ltz_with_precision",
+
org.apache.flink.cdc.common.types.DataTypes.TIMESTAMP_LTZ(3))
+ .physicalColumn(
+ "null_string",
org.apache.flink.cdc.common.types.DataTypes.STRING())
+ .primaryKey("col1")
+ .build();
+ TableSchemaInfo tableSchemaInfo = new TableSchemaInfo(schema, null,
ZoneId.of("UTC+8"));
+ Object[] testData =
+ new Object[] {
+ BinaryStringData.fromString("pk"),
+ true,
+ new byte[] {1, 2},
+ new byte[] {3, 4},
+ new byte[] {5, 6, 7},
+ (byte) 1,
+ (short) 2,
+ 3,
+ 4L,
+ 5.1f,
+ 6.2,
+ DecimalData.fromBigDecimal(new BigDecimal("7.123"), 6, 3),
+ BinaryStringData.fromString("test1"),
+ BinaryStringData.fromString("test2"),
+ BinaryStringData.fromString("test3"),
+ 100,
+ 200,
+ 300,
+ TimestampData.fromTimestamp(
+ java.sql.Timestamp.valueOf("2023-01-01
00:00:00.000")),
+
TimestampData.fromTimestamp(java.sql.Timestamp.valueOf("2023-01-01 00:00:00")),
+
LocalZonedTimestampData.fromInstant(Instant.parse("2023-01-01T00:00:00.000Z")),
+
LocalZonedTimestampData.fromInstant(Instant.parse("2023-01-01T00:00:00.000Z")),
+ null
+ };
+ BinaryRecordData recordData =
+ new
BinaryRecordDataGenerator(schema.getColumnDataTypes().toArray(new DataType[0]))
+ .generate(testData);
+
+ Assertions.assertEquals(
+ GenericRowData.ofKind(
+ RowKind.INSERT,
+
org.apache.flink.table.data.binary.BinaryStringData.fromString("pk"),
+ true,
+ new byte[] {1, 2},
+ new byte[] {3, 4},
+ new byte[] {5, 6, 7},
+ (byte) 1,
+ (short) 2,
+ 3,
+ 4L,
+ 5.1f,
+ 6.2,
+ org.apache.flink.table.data.DecimalData.fromBigDecimal(
+ new BigDecimal("7.123"), 6, 3),
+
org.apache.flink.table.data.binary.BinaryStringData.fromString("test1"),
+
org.apache.flink.table.data.binary.BinaryStringData.fromString("test2"),
+
org.apache.flink.table.data.binary.BinaryStringData.fromString("test3"),
+ 100,
+ 200,
+ 300,
+
org.apache.flink.table.data.TimestampData.fromTimestamp(
+ Timestamp.valueOf("2023-01-01 00:00:00.000")),
+
org.apache.flink.table.data.TimestampData.fromTimestamp(
+ Timestamp.valueOf("2023-01-01 00:00:00")),
+ // plus 8 hours.
+ org.apache.flink.table.data.TimestampData.fromInstant(
+ Instant.parse("2023-01-01T08:00:00.000Z")),
+ org.apache.flink.table.data.TimestampData.fromInstant(
+ Instant.parse("2023-01-01T08:00:00.000Z")),
+ null),
+ tableSchemaInfo.getRowDataFromRecordData(recordData));
+ }
+}
diff --git
a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-kafka/src/test/java/org/apache/flink/cdc/connectors/kafka/json/canal/CanalJsonSerializationSchemaTest.java
b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-kafka/src/test/java/org/apache/flink/cdc/connectors/kafka/json/canal/CanalJsonSerializationSchemaTest.java
new file mode 100644
index 000000000..52181fd6b
--- /dev/null
+++
b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-kafka/src/test/java/org/apache/flink/cdc/connectors/kafka/json/canal/CanalJsonSerializationSchemaTest.java
@@ -0,0 +1,133 @@
+/*
+ * 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.kafka.json.canal;
+
+import org.apache.flink.api.common.serialization.SerializationSchema;
+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.DataTypes;
+import org.apache.flink.cdc.common.types.RowType;
+import org.apache.flink.cdc.connectors.kafka.json.ChangeLogJsonFormatFactory;
+import org.apache.flink.cdc.connectors.kafka.json.JsonSerializationType;
+import org.apache.flink.cdc.connectors.kafka.json.MockInitializationContext;
+import org.apache.flink.cdc.runtime.typeutils.BinaryRecordDataGenerator;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.util.jackson.JacksonMapperFactory;
+
+import
org.apache.flink.shaded.jackson2.com.fasterxml.jackson.core.JsonGenerator;
+import
org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.JsonNode;
+import
org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.ObjectMapper;
+
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+/** Tests for {@link CanalJsonSerializationSchema}. */
+public class CanalJsonSerializationSchemaTest {
+
+ public static final TableId TABLE_1 =
+ TableId.tableId("default_namespace", "default_schema", "table1");
+
+ @Test
+ public void testSerialize() throws Exception {
+ ObjectMapper mapper =
+ JacksonMapperFactory.createObjectMapper()
+
.configure(JsonGenerator.Feature.WRITE_BIGDECIMAL_AS_PLAIN, false);
+ SerializationSchema<Event> serializationSchema =
+ ChangeLogJsonFormatFactory.createSerializationSchema(
+ new Configuration(), JsonSerializationType.CANAL_JSON);
+ serializationSchema.open(new MockInitializationContext());
+
+ // create table
+ Schema schema =
+ Schema.newBuilder()
+ .physicalColumn("col1", DataTypes.STRING())
+ .physicalColumn("col2", DataTypes.STRING())
+ .primaryKey("col1")
+ .build();
+ CreateTableEvent createTableEvent = new CreateTableEvent(TABLE_1,
schema);
+ Assertions.assertNull(serializationSchema.serialize(createTableEvent));
+
+ // insert
+ 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("1")
+ }));
+ JsonNode expected =
+ mapper.readTree(
+
"{\"old\":null,\"data\":[{\"col1\":\"1\",\"col2\":\"1\"}],\"type\":\"INSERT\"}");
+ JsonNode actual =
mapper.readTree(serializationSchema.serialize(insertEvent1));
+ Assertions.assertEquals(expected, actual);
+
+ DataChangeEvent insertEvent2 =
+ DataChangeEvent.insertEvent(
+ TABLE_1,
+ generator.generate(
+ new Object[] {
+ BinaryStringData.fromString("2"),
+ BinaryStringData.fromString("2")
+ }));
+ expected =
+ mapper.readTree(
+
"{\"old\":null,\"data\":[{\"col1\":\"2\",\"col2\":\"2\"}],\"type\":\"INSERT\"}");
+ actual = mapper.readTree(serializationSchema.serialize(insertEvent2));
+ Assertions.assertEquals(expected, actual);
+
+ DataChangeEvent deleteEvent =
+ DataChangeEvent.deleteEvent(
+ TABLE_1,
+ generator.generate(
+ new Object[] {
+ BinaryStringData.fromString("2"),
+ BinaryStringData.fromString("2")
+ }));
+ expected =
+ mapper.readTree(
+
"{\"old\":[{\"col1\":\"2\",\"col2\":\"2\"}],\"data\":null,\"type\":\"DELETE\"}");
+ actual = mapper.readTree(serializationSchema.serialize(deleteEvent));
+ Assertions.assertEquals(expected, actual);
+
+ DataChangeEvent updateEvent =
+ DataChangeEvent.updateEvent(
+ TABLE_1,
+ generator.generate(
+ new Object[] {
+ BinaryStringData.fromString("1"),
+ BinaryStringData.fromString("1")
+ }),
+ generator.generate(
+ new Object[] {
+ BinaryStringData.fromString("1"),
+ BinaryStringData.fromString("x")
+ }));
+ expected =
+ mapper.readTree(
+
"{\"old\":[{\"col1\":\"1\",\"col2\":\"1\"}],\"data\":[{\"col1\":\"1\",\"col2\":\"x\"}],\"type\":\"UPDATE\"}");
+ actual = mapper.readTree(serializationSchema.serialize(updateEvent));
+ Assertions.assertEquals(expected, actual);
+ }
+}
diff --git
a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-kafka/src/test/java/org/apache/flink/cdc/connectors/kafka/json/debezium/DebeziumJsonSerializationSchemaTest.java
b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-kafka/src/test/java/org/apache/flink/cdc/connectors/kafka/json/debezium/DebeziumJsonSerializationSchemaTest.java
new file mode 100644
index 000000000..7a47bf8bd
--- /dev/null
+++
b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-kafka/src/test/java/org/apache/flink/cdc/connectors/kafka/json/debezium/DebeziumJsonSerializationSchemaTest.java
@@ -0,0 +1,128 @@
+/*
+ * 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.kafka.json.debezium;
+
+import org.apache.flink.api.common.serialization.SerializationSchema;
+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.DataTypes;
+import org.apache.flink.cdc.common.types.RowType;
+import org.apache.flink.cdc.connectors.kafka.json.ChangeLogJsonFormatFactory;
+import org.apache.flink.cdc.connectors.kafka.json.JsonSerializationType;
+import org.apache.flink.cdc.connectors.kafka.json.MockInitializationContext;
+import org.apache.flink.cdc.runtime.typeutils.BinaryRecordDataGenerator;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.util.jackson.JacksonMapperFactory;
+
+import
org.apache.flink.shaded.jackson2.com.fasterxml.jackson.core.JsonGenerator;
+import
org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.JsonNode;
+import
org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.ObjectMapper;
+
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+/** Tests for {@link DebeziumJsonSerializationSchema}. */
+public class DebeziumJsonSerializationSchemaTest {
+
+ public static final TableId TABLE_1 =
+ TableId.tableId("default_namespace", "default_schema", "table1");
+
+ @Test
+ public void testSerialize() throws Exception {
+ ObjectMapper mapper =
+ JacksonMapperFactory.createObjectMapper()
+
.configure(JsonGenerator.Feature.WRITE_BIGDECIMAL_AS_PLAIN, false);
+ SerializationSchema<Event> serializationSchema =
+ ChangeLogJsonFormatFactory.createSerializationSchema(
+ new Configuration(),
JsonSerializationType.DEBEZIUM_JSON);
+ serializationSchema.open(new MockInitializationContext());
+ // create table
+ Schema schema =
+ Schema.newBuilder()
+ .physicalColumn("col1", DataTypes.STRING())
+ .physicalColumn("col2", DataTypes.STRING())
+ .primaryKey("col1")
+ .build();
+ CreateTableEvent createTableEvent = new CreateTableEvent(TABLE_1,
schema);
+ Assertions.assertNull(serializationSchema.serialize(createTableEvent));
+ BinaryRecordDataGenerator generator =
+ new BinaryRecordDataGenerator(RowType.of(DataTypes.STRING(),
DataTypes.STRING()));
+ // insert
+ DataChangeEvent insertEvent1 =
+ DataChangeEvent.insertEvent(
+ TABLE_1,
+ generator.generate(
+ new Object[] {
+ BinaryStringData.fromString("1"),
+ BinaryStringData.fromString("1")
+ }));
+ JsonNode expected =
+ mapper.readTree(
+
"{\"before\":null,\"after\":{\"col1\":\"1\",\"col2\":\"1\"},\"op\":\"c\"}");
+ JsonNode actual =
mapper.readTree(serializationSchema.serialize(insertEvent1));
+ Assertions.assertEquals(expected, actual);
+ DataChangeEvent insertEvent2 =
+ DataChangeEvent.insertEvent(
+ TABLE_1,
+ generator.generate(
+ new Object[] {
+ BinaryStringData.fromString("2"),
+ BinaryStringData.fromString("2")
+ }));
+ expected =
+ mapper.readTree(
+
"{\"before\":null,\"after\":{\"col1\":\"2\",\"col2\":\"2\"},\"op\":\"c\"}");
+ actual = mapper.readTree(serializationSchema.serialize(insertEvent2));
+ Assertions.assertEquals(expected, actual);
+ DataChangeEvent deleteEvent =
+ DataChangeEvent.deleteEvent(
+ TABLE_1,
+ generator.generate(
+ new Object[] {
+ BinaryStringData.fromString("2"),
+ BinaryStringData.fromString("2")
+ }));
+ expected =
+ mapper.readTree(
+
"{\"before\":{\"col1\":\"2\",\"col2\":\"2\"},\"after\":null,\"op\":\"d\"}");
+ actual = mapper.readTree(serializationSchema.serialize(deleteEvent));
+ Assertions.assertEquals(expected, actual);
+ DataChangeEvent updateEvent =
+ DataChangeEvent.updateEvent(
+ TABLE_1,
+ generator.generate(
+ new Object[] {
+ BinaryStringData.fromString("1"),
+ BinaryStringData.fromString("1")
+ }),
+ generator.generate(
+ new Object[] {
+ BinaryStringData.fromString("1"),
+ BinaryStringData.fromString("x")
+ }));
+ expected =
+ mapper.readTree(
+
"{\"before\":{\"col1\":\"1\",\"col2\":\"1\"},\"after\":{\"col1\":\"1\",\"col2\":\"x\"},\"op\":\"u\"}");
+ actual = mapper.readTree(serializationSchema.serialize(updateEvent));
+ Assertions.assertEquals(expected, actual);
+ }
+}
diff --git
a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-kafka/src/test/java/org/apache/flink/cdc/connectors/kafka/sink/KafkaDataSinkFactoryTest.java
b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-kafka/src/test/java/org/apache/flink/cdc/connectors/kafka/sink/KafkaDataSinkFactoryTest.java
new file mode 100644
index 000000000..1f1957bea
--- /dev/null
+++
b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-kafka/src/test/java/org/apache/flink/cdc/connectors/kafka/sink/KafkaDataSinkFactoryTest.java
@@ -0,0 +1,47 @@
+/*
+ * 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.kafka.sink;
+
+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.shaded.guava31.com.google.common.collect.ImmutableMap;
+
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+/** Tests for {@link KafkaDataSinkFactory}. */
+public class KafkaDataSinkFactoryTest {
+
+ @Test
+ public void testCreateDataSink() {
+ DataSinkFactory sinkFactory =
+ FactoryDiscoveryUtils.getFactoryByIdentifier("kafka",
DataSinkFactory.class);
+ Assertions.assertTrue(sinkFactory instanceof KafkaDataSinkFactory);
+
+ Configuration conf = Configuration.fromMap(ImmutableMap.<String,
String>builder().build());
+ DataSink dataSink =
+ sinkFactory.createDataSink(
+ new FactoryHelper.DefaultContext(
+ conf, conf,
Thread.currentThread().getContextClassLoader()));
+ Assertions.assertTrue(dataSink instanceof KafkaDataSink);
+ }
+}
diff --git
a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-kafka/src/test/java/org/apache/flink/cdc/connectors/kafka/sink/KafkaDataSinkITCase.java
b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-kafka/src/test/java/org/apache/flink/cdc/connectors/kafka/sink/KafkaDataSinkITCase.java
new file mode 100644
index 000000000..2e2656596
--- /dev/null
+++
b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-kafka/src/test/java/org/apache/flink/cdc/connectors/kafka/sink/KafkaDataSinkITCase.java
@@ -0,0 +1,495 @@
+/*
+ * 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.kafka.sink;
+
+import org.apache.flink.api.common.restartstrategy.RestartStrategies;
+import org.apache.flink.cdc.common.configuration.Configuration;
+import org.apache.flink.cdc.common.data.binary.BinaryStringData;
+import org.apache.flink.cdc.common.event.AddColumnEvent;
+import org.apache.flink.cdc.common.event.CreateTableEvent;
+import org.apache.flink.cdc.common.event.DataChangeEvent;
+import org.apache.flink.cdc.common.event.DropColumnEvent;
+import org.apache.flink.cdc.common.event.Event;
+import org.apache.flink.cdc.common.event.RenameColumnEvent;
+import org.apache.flink.cdc.common.event.TableId;
+import org.apache.flink.cdc.common.factories.FactoryHelper;
+import org.apache.flink.cdc.common.schema.Column;
+import org.apache.flink.cdc.common.schema.Schema;
+import org.apache.flink.cdc.common.sink.FlinkSinkProvider;
+import org.apache.flink.cdc.common.types.DataTypes;
+import org.apache.flink.cdc.common.types.RowType;
+import org.apache.flink.cdc.connectors.kafka.json.JsonSerializationType;
+import org.apache.flink.cdc.runtime.typeutils.BinaryRecordDataGenerator;
+import org.apache.flink.cdc.runtime.typeutils.EventTypeInfo;
+import org.apache.flink.streaming.api.datastream.DataStream;
+import org.apache.flink.streaming.api.environment.LocalStreamEnvironment;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.util.TestLogger;
+import org.apache.flink.util.jackson.JacksonMapperFactory;
+
+import
org.apache.flink.shaded.jackson2.com.fasterxml.jackson.core.JsonGenerator;
+import
org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.JsonNode;
+import
org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.ObjectMapper;
+
+import org.apache.kafka.clients.CommonClientConfigs;
+import org.apache.kafka.clients.admin.AdminClient;
+import org.apache.kafka.clients.admin.CreateTopicsResult;
+import org.apache.kafka.clients.admin.DeleteTopicsResult;
+import org.apache.kafka.clients.admin.NewTopic;
+import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.testcontainers.containers.KafkaContainer;
+import org.testcontainers.containers.Network;
+
+import java.io.IOException;
+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 java.util.Properties;
+import java.util.UUID;
+import java.util.concurrent.ExecutionException;
+import java.util.stream.Collectors;
+
+import static org.apache.flink.util.DockerImageVersions.KAFKA;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.fail;
+
+/** Tests for using {@link KafkaDataSink} writing to a Kafka cluster. */
+class KafkaDataSinkITCase extends TestLogger {
+
+ private static final Logger LOG =
LoggerFactory.getLogger(KafkaDataSinkITCase.class);
+ private static final String INTER_CONTAINER_KAFKA_ALIAS = "kafka";
+ private static final Network NETWORK = Network.newNetwork();
+ private static final int ZK_TIMEOUT_MILLIS = 30000;
+ private static final short TOPIC_REPLICATION_FACTOR = 1;
+ private static AdminClient admin;
+
+ private String topic;
+
+ private TableId table1;
+
+ public static final KafkaContainer KAFKA_CONTAINER =
+ KafkaUtil.createKafkaContainer(KAFKA, LOG)
+ .withEmbeddedZookeeper()
+ .withNetwork(NETWORK)
+ .withNetworkAliases(INTER_CONTAINER_KAFKA_ALIAS);
+
+ @BeforeAll
+ public static void setupAdmin() {
+ KAFKA_CONTAINER.start();
+ Map<String, Object> properties = new HashMap<>();
+ properties.put(
+ CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG,
+ KAFKA_CONTAINER.getBootstrapServers());
+ admin = AdminClient.create(properties);
+ }
+
+ @AfterAll
+ public static void teardownAdmin() {
+ admin.close();
+ KAFKA_CONTAINER.stop();
+ }
+
+ @BeforeEach
+ public void setUp() throws ExecutionException, InterruptedException {
+ table1 =
+ TableId.tableId(
+ "default_namespace", "default_schema",
UUID.randomUUID().toString());
+ topic = table1.toString();
+ createTestTopic(topic, 1, TOPIC_REPLICATION_FACTOR);
+ }
+
+ @AfterEach
+ public void tearDown() throws ExecutionException, InterruptedException {
+ deleteTestTopic(topic);
+ }
+
+ private void createTestTopic(String topic, int numPartitions, short
replicationFactor)
+ throws ExecutionException, InterruptedException {
+ final CreateTopicsResult result =
+ admin.createTopics(
+ Collections.singletonList(
+ new NewTopic(topic, numPartitions,
replicationFactor)));
+ result.all().get();
+ }
+
+ private void deleteTestTopic(String topic) throws ExecutionException,
InterruptedException {
+ final DeleteTopicsResult result =
admin.deleteTopics(Collections.singletonList(topic));
+ result.all().get();
+ }
+
+ private List<Event> createSourceEvents() {
+ List<Event> events = new ArrayList<>();
+ // create table
+ Schema schema =
+ Schema.newBuilder()
+ .physicalColumn("col1", DataTypes.STRING())
+ .physicalColumn("col2", DataTypes.STRING())
+ .primaryKey("col1")
+ .build();
+ CreateTableEvent createTableEvent = new CreateTableEvent(table1,
schema);
+ events.add(createTableEvent);
+
+ BinaryRecordDataGenerator generator =
+ new BinaryRecordDataGenerator(RowType.of(DataTypes.STRING(),
DataTypes.STRING()));
+ // insert
+ DataChangeEvent insertEvent1 =
+ DataChangeEvent.insertEvent(
+ table1,
+ generator.generate(
+ new Object[] {
+ BinaryStringData.fromString("1"),
+ BinaryStringData.fromString("1")
+ }));
+ events.add(insertEvent1);
+ DataChangeEvent insertEvent2 =
+ DataChangeEvent.insertEvent(
+ table1,
+ generator.generate(
+ new Object[] {
+ BinaryStringData.fromString("2"),
+ BinaryStringData.fromString("2")
+ }));
+ events.add(insertEvent2);
+ DataChangeEvent insertEvent3 =
+ DataChangeEvent.insertEvent(
+ table1,
+ generator.generate(
+ new Object[] {
+ BinaryStringData.fromString("3"),
+ BinaryStringData.fromString("3")
+ }));
+ events.add(insertEvent3);
+
+ // add column
+ AddColumnEvent.ColumnWithPosition columnWithPosition =
+ new AddColumnEvent.ColumnWithPosition(
+ Column.physicalColumn("col3", DataTypes.STRING()));
+ AddColumnEvent addColumnEvent =
+ new AddColumnEvent(table1,
Collections.singletonList(columnWithPosition));
+ events.add(addColumnEvent);
+
+ // rename column
+ Map<String, String> nameMapping = new HashMap<>();
+ nameMapping.put("col2", "newCol2");
+ nameMapping.put("col3", "newCol3");
+ RenameColumnEvent renameColumnEvent = new RenameColumnEvent(table1,
nameMapping);
+ events.add(renameColumnEvent);
+
+ // drop column
+ DropColumnEvent dropColumnEvent =
+ new DropColumnEvent(table1,
Collections.singletonList("newCol2"));
+ events.add(dropColumnEvent);
+
+ // delete
+ events.add(
+ DataChangeEvent.deleteEvent(
+ table1,
+ generator.generate(
+ new Object[] {
+ BinaryStringData.fromString("1"),
+ BinaryStringData.fromString("1")
+ })));
+
+ // update
+ events.add(
+ DataChangeEvent.updateEvent(
+ table1,
+ generator.generate(
+ new Object[] {
+ BinaryStringData.fromString("2"),
+ BinaryStringData.fromString("")
+ }),
+ generator.generate(
+ new Object[] {
+ BinaryStringData.fromString("2"),
+ BinaryStringData.fromString("x")
+ })));
+ return events;
+ }
+
+ @Test
+ void testDebeziumJsonFormat() throws Exception {
+ final StreamExecutionEnvironment env = new LocalStreamEnvironment();
+ env.enableCheckpointing(1000L);
+ env.setRestartStrategy(RestartStrategies.noRestart());
+ final DataStream<Event> source =
+ env.fromCollection(createSourceEvents(), new EventTypeInfo());
+ Map<String, String> config = new HashMap<>();
+ Properties properties = getKafkaClientConfiguration();
+ properties.forEach(
+ (key, value) ->
+ config.put(
+ KafkaDataSinkOptions.PROPERTIES_PREFIX +
key.toString(),
+ value.toString()));
+ source.sinkTo(
+ ((FlinkSinkProvider)
+ (new KafkaDataSinkFactory()
+ .createDataSink(
+ new
FactoryHelper.DefaultContext(
+
Configuration.fromMap(config),
+
Configuration.fromMap(new HashMap<>()),
+
this.getClass().getClassLoader()))
+ .getEventSinkProvider()))
+ .getSink());
+ env.execute();
+
+ final List<ConsumerRecord<byte[], byte[]>> collectedRecords =
+ drainAllRecordsFromTopic(topic, false);
+ final long recordsCount = 5;
+ assertThat(recordsCount).isEqualTo(collectedRecords.size());
+ ObjectMapper mapper =
+ JacksonMapperFactory.createObjectMapper()
+
.configure(JsonGenerator.Feature.WRITE_BIGDECIMAL_AS_PLAIN, false);
+ List<JsonNode> expected =
+ Arrays.asList(
+ mapper.readTree(
+
"{\"before\":null,\"after\":{\"col1\":\"1\",\"col2\":\"1\"},\"op\":\"c\"}"),
+ mapper.readTree(
+
"{\"before\":null,\"after\":{\"col1\":\"2\",\"col2\":\"2\"},\"op\":\"c\"}"),
+ mapper.readTree(
+
"{\"before\":null,\"after\":{\"col1\":\"3\",\"col2\":\"3\"},\"op\":\"c\"}"),
+ mapper.readTree(
+
"{\"before\":{\"col1\":\"1\",\"newCol3\":\"1\"},\"after\":null,\"op\":\"d\"}"),
+ mapper.readTree(
+
"{\"before\":{\"col1\":\"2\",\"newCol3\":\"\"},\"after\":{\"col1\":\"2\",\"newCol3\":\"x\"},\"op\":\"u\"}"));
+ assertThat(deserializeValues(collectedRecords)).containsAll(expected);
+ checkProducerLeak();
+ }
+
+ @Test
+ void testCanalJsonFormat() throws Exception {
+ final StreamExecutionEnvironment env = new LocalStreamEnvironment();
+ env.enableCheckpointing(1000L);
+ env.setRestartStrategy(RestartStrategies.noRestart());
+ final DataStream<Event> source =
+ env.fromCollection(createSourceEvents(), new EventTypeInfo());
+ Map<String, String> config = new HashMap<>();
+ Properties properties = getKafkaClientConfiguration();
+ properties.forEach(
+ (key, value) ->
+ config.put(
+ KafkaDataSinkOptions.PROPERTIES_PREFIX +
key.toString(),
+ value.toString()));
+ config.put(
+ KafkaDataSinkOptions.VALUE_FORMAT.key(),
+ JsonSerializationType.CANAL_JSON.toString());
+ source.sinkTo(
+ ((FlinkSinkProvider)
+ (new KafkaDataSinkFactory()
+ .createDataSink(
+ new
FactoryHelper.DefaultContext(
+
Configuration.fromMap(config),
+
Configuration.fromMap(new HashMap<>()),
+
this.getClass().getClassLoader()))
+ .getEventSinkProvider()))
+ .getSink());
+ env.execute();
+
+ final List<ConsumerRecord<byte[], byte[]>> collectedRecords =
+ drainAllRecordsFromTopic(topic, false);
+ final long recordsCount = 5;
+ assertThat(recordsCount).isEqualTo(collectedRecords.size());
+ for (ConsumerRecord<byte[], byte[]> consumerRecord : collectedRecords)
{
+ assertThat(
+ consumerRecord
+ .headers()
+ .headers(
+
PipelineKafkaRecordSerializationSchema
+ .TABLE_NAME_HEADER_KEY)
+ .iterator())
+ .isExhausted();
+ }
+ ObjectMapper mapper =
+ JacksonMapperFactory.createObjectMapper()
+
.configure(JsonGenerator.Feature.WRITE_BIGDECIMAL_AS_PLAIN, false);
+ List<JsonNode> expected =
+ Arrays.asList(
+ mapper.readTree(
+
"{\"old\":null,\"data\":[{\"col1\":\"1\",\"col2\":\"1\"}],\"type\":\"INSERT\"}"),
+ mapper.readTree(
+
"{\"old\":null,\"data\":[{\"col1\":\"2\",\"col2\":\"2\"}],\"type\":\"INSERT\"}"),
+ mapper.readTree(
+
"{\"old\":null,\"data\":[{\"col1\":\"3\",\"col2\":\"3\"}],\"type\":\"INSERT\"}"),
+ mapper.readTree(
+
"{\"old\":[{\"col1\":\"1\",\"newCol3\":\"1\"}],\"data\":null,\"type\":\"DELETE\"}"),
+ mapper.readTree(
+
"{\"old\":[{\"col1\":\"2\",\"newCol3\":\"\"}],\"data\":[{\"col1\":\"2\",\"newCol3\":\"x\"}],\"type\":\"UPDATE\"}"));
+ assertThat(deserializeValues(collectedRecords)).containsAll(expected);
+ checkProducerLeak();
+ }
+
+ @Test
+ void testTopicAndHeaderOption() throws Exception {
+ final StreamExecutionEnvironment env = new LocalStreamEnvironment();
+ env.enableCheckpointing(1000L);
+ env.setRestartStrategy(RestartStrategies.noRestart());
+ final DataStream<Event> source =
+ env.fromCollection(createSourceEvents(), new EventTypeInfo());
+ Map<String, String> config = new HashMap<>();
+ config.put(KafkaDataSinkOptions.TOPIC.key(), "test_topic");
+
config.put(KafkaDataSinkOptions.SINK_ADD_TABLEID_TO_HEADER_ENABLED.key(),
"true");
+ Properties properties = getKafkaClientConfiguration();
+ properties.forEach(
+ (key, value) ->
+ config.put(
+ KafkaDataSinkOptions.PROPERTIES_PREFIX +
key.toString(),
+ value.toString()));
+ source.sinkTo(
+ ((FlinkSinkProvider)
+ (new KafkaDataSinkFactory()
+ .createDataSink(
+ new
FactoryHelper.DefaultContext(
+
Configuration.fromMap(config),
+
Configuration.fromMap(new HashMap<>()),
+
this.getClass().getClassLoader()))
+ .getEventSinkProvider()))
+ .getSink());
+ env.execute();
+
+ final List<ConsumerRecord<byte[], byte[]>> collectedRecords =
+ drainAllRecordsFromTopic("test_topic", false);
+ final long recordsCount = 5;
+ assertThat(recordsCount).isEqualTo(collectedRecords.size());
+ for (ConsumerRecord<byte[], byte[]> consumerRecord : collectedRecords)
{
+ assertThat(
+ new String(
+ consumerRecord
+ .headers()
+ .headers(
+
PipelineKafkaRecordSerializationSchema
+
.NAMESPACE_HEADER_KEY)
+ .iterator()
+ .next()
+ .value()))
+ .isEqualTo(table1.getNamespace());
+ assertThat(
+ new String(
+ consumerRecord
+ .headers()
+ .headers(
+
PipelineKafkaRecordSerializationSchema
+
.SCHEMA_NAME_HEADER_KEY)
+ .iterator()
+ .next()
+ .value()))
+ .isEqualTo(table1.getSchemaName());
+ assertThat(
+ new String(
+ consumerRecord
+ .headers()
+ .headers(
+
PipelineKafkaRecordSerializationSchema
+
.TABLE_NAME_HEADER_KEY)
+ .iterator()
+ .next()
+ .value()))
+ .isEqualTo(table1.getTableName());
+ }
+ ObjectMapper mapper =
+ JacksonMapperFactory.createObjectMapper()
+
.configure(JsonGenerator.Feature.WRITE_BIGDECIMAL_AS_PLAIN, false);
+ List<JsonNode> expected =
+ Arrays.asList(
+ mapper.readTree(
+
"{\"before\":null,\"after\":{\"col1\":\"1\",\"col2\":\"1\"},\"op\":\"c\"}"),
+ mapper.readTree(
+
"{\"before\":null,\"after\":{\"col1\":\"2\",\"col2\":\"2\"},\"op\":\"c\"}"),
+ mapper.readTree(
+
"{\"before\":null,\"after\":{\"col1\":\"3\",\"col2\":\"3\"},\"op\":\"c\"}"),
+ mapper.readTree(
+
"{\"before\":{\"col1\":\"1\",\"newCol3\":\"1\"},\"after\":null,\"op\":\"d\"}"),
+ mapper.readTree(
+
"{\"before\":{\"col1\":\"2\",\"newCol3\":\"\"},\"after\":{\"col1\":\"2\",\"newCol3\":\"x\"},\"op\":\"u\"}"));
+ assertThat(deserializeValues(collectedRecords)).containsAll(expected);
+ checkProducerLeak();
+ }
+
+ private List<ConsumerRecord<byte[], byte[]>> drainAllRecordsFromTopic(
+ String topic, boolean committed) {
+ Properties properties = getKafkaClientConfiguration();
+ return KafkaUtil.drainAllRecordsFromTopic(topic, properties,
committed);
+ }
+
+ private void checkProducerLeak() throws InterruptedException {
+ List<Map.Entry<Thread, StackTraceElement[]>> leaks = null;
+ for (int tries = 0; tries < 10; tries++) {
+ leaks =
+ Thread.getAllStackTraces().entrySet().stream()
+ .filter(this::findAliveKafkaThread)
+ .collect(Collectors.toList());
+ if (leaks.isEmpty()) {
+ return;
+ }
+ Thread.sleep(1000);
+ }
+
+ for (Map.Entry<Thread, StackTraceElement[]> leak : leaks) {
+ leak.getKey().stop();
+ }
+ fail(
+ "Detected producer leaks:\n"
+ +
leaks.stream().map(this::format).collect(Collectors.joining("\n\n")));
+ }
+
+ private static List<JsonNode>
deserializeValues(List<ConsumerRecord<byte[], byte[]>> records)
+ throws IOException {
+ ObjectMapper mapper =
+ JacksonMapperFactory.createObjectMapper()
+
.configure(JsonGenerator.Feature.WRITE_BIGDECIMAL_AS_PLAIN, false);
+ List<JsonNode> result = new ArrayList<>();
+ for (ConsumerRecord<byte[], byte[]> record : records) {
+ result.add(mapper.readTree(record.value()));
+ }
+ return result;
+ }
+
+ private static Properties getKafkaClientConfiguration() {
+ final Properties standardProps = new Properties();
+ standardProps.put("bootstrap.servers",
KAFKA_CONTAINER.getBootstrapServers());
+ standardProps.put("group.id", UUID.randomUUID().toString());
+ standardProps.put("enable.auto.commit", false);
+ standardProps.put("auto.offset.reset", "earliest");
+ standardProps.put("max.partition.fetch.bytes", 256);
+ standardProps.put("zookeeper.session.timeout.ms", ZK_TIMEOUT_MILLIS);
+ standardProps.put("zookeeper.connection.timeout.ms",
ZK_TIMEOUT_MILLIS);
+ return standardProps;
+ }
+
+ private String format(Map.Entry<Thread, StackTraceElement[]> leak) {
+ String stackTrace =
+ Arrays.stream(leak.getValue())
+ .map(StackTraceElement::toString)
+ .collect(Collectors.joining("\n"));
+ return leak.getKey().getName() + ":\n" + stackTrace;
+ }
+
+ private boolean findAliveKafkaThread(Map.Entry<Thread,
StackTraceElement[]> threadStackTrace) {
+ return threadStackTrace.getKey().getState() != Thread.State.TERMINATED
+ &&
threadStackTrace.getKey().getName().contains("kafka-producer-network-thread");
+ }
+}
diff --git
a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-kafka/src/test/java/org/apache/flink/cdc/connectors/kafka/sink/KafkaUtil.java
b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-kafka/src/test/java/org/apache/flink/cdc/connectors/kafka/sink/KafkaUtil.java
new file mode 100644
index 000000000..d1f36f657
--- /dev/null
+++
b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-kafka/src/test/java/org/apache/flink/cdc/connectors/kafka/sink/KafkaUtil.java
@@ -0,0 +1,188 @@
+/*
+ * 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.kafka.sink;
+
+import org.apache.flink.util.StringUtils;
+
+import org.apache.kafka.clients.consumer.ConsumerConfig;
+import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.apache.kafka.clients.consumer.ConsumerRecords;
+import org.apache.kafka.clients.consumer.KafkaConsumer;
+import org.apache.kafka.common.KafkaException;
+import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.common.serialization.ByteArrayDeserializer;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.testcontainers.containers.KafkaContainer;
+import org.testcontainers.containers.output.Slf4jLogConsumer;
+import org.testcontainers.utility.DockerImageName;
+
+import java.time.Duration;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.Properties;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+/** Collection of methods to interact with a Kafka cluster. */
+public class KafkaUtil {
+
+ private static final Logger LOG = LoggerFactory.getLogger(KafkaUtil.class);
+ private static final Duration CONSUMER_POLL_DURATION =
Duration.ofSeconds(1);
+
+ private KafkaUtil() {}
+
+ /**
+ * This method helps to set commonly used Kafka configurations and aligns
the internal Kafka log
+ * levels with the ones used by the capturing logger.
+ *
+ * @param dockerImageVersion describing the Kafka image
+ * @param logger to derive the log level from
+ * @return configured Kafka container
+ */
+ public static KafkaContainer createKafkaContainer(String
dockerImageVersion, Logger logger) {
+ return createKafkaContainer(dockerImageVersion, logger, null);
+ }
+
+ /**
+ * This method helps to set commonly used Kafka configurations and aligns
the internal Kafka log
+ * levels with the ones used by the capturing logger, and set the prefix
of logger.
+ */
+ public static KafkaContainer createKafkaContainer(
+ String dockerImageVersion, Logger logger, String loggerPrefix) {
+ String logLevel;
+ if (logger.isTraceEnabled()) {
+ logLevel = "TRACE";
+ } else if (logger.isDebugEnabled()) {
+ logLevel = "DEBUG";
+ } else if (logger.isInfoEnabled()) {
+ logLevel = "INFO";
+ } else if (logger.isWarnEnabled()) {
+ logLevel = "WARN";
+ } else if (logger.isErrorEnabled()) {
+ logLevel = "ERROR";
+ } else {
+ logLevel = "OFF";
+ }
+
+ Slf4jLogConsumer logConsumer = new Slf4jLogConsumer(logger);
+ if (!StringUtils.isNullOrWhitespaceOnly(loggerPrefix)) {
+ logConsumer.withPrefix(loggerPrefix);
+ }
+ return new KafkaContainer(DockerImageName.parse(dockerImageVersion))
+ .withEnv("KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR", "1")
+ .withEnv("KAFKA_TRANSACTION_STATE_LOG_MIN_ISR", "1")
+ .withEnv("KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR", "1")
+ .withEnv("KAFKA_CONFLUENT_SUPPORT_METRICS_ENABLE", "false")
+ .withEnv("KAFKA_LOG4J_ROOT_LOGLEVEL", logLevel)
+ .withEnv("KAFKA_LOG4J_LOGGERS", "state.change.logger=" +
logLevel)
+ .withEnv("KAFKA_CONFLUENT_SUPPORT_METRICS_ENABLE", "false")
+ .withEnv(
+ "KAFKA_TRANSACTION_MAX_TIMEOUT_MS",
+ String.valueOf(Duration.ofHours(2).toMillis()))
+ .withEnv("KAFKA_LOG4J_TOOLS_ROOT_LOGLEVEL", logLevel)
+ .withLogConsumer(logConsumer);
+ }
+
+ /**
+ * Drain all records available from the given topic from the beginning
until the current highest
+ * offset.
+ *
+ * <p>This method will fetch the latest offsets for the partitions once
and only return records
+ * until that point.
+ *
+ * @param topic to fetch from
+ * @param properties used to configure the created {@link KafkaConsumer}
+ * @param committed determines the mode {@link
ConsumerConfig#ISOLATION_LEVEL_CONFIG} with which
+ * the consumer reads the records.
+ * @return all {@link ConsumerRecord} in the topic
+ * @throws KafkaException
+ */
+ public static List<ConsumerRecord<byte[], byte[]>>
drainAllRecordsFromTopic(
+ String topic, Properties properties, boolean committed) throws
KafkaException {
+ final Properties consumerConfig = new Properties();
+ consumerConfig.putAll(properties);
+ consumerConfig.put(
+ ConsumerConfig.ISOLATION_LEVEL_CONFIG,
+ committed ? "read_committed" : "read_uncommitted");
+ return drainAllRecordsFromTopic(topic, consumerConfig);
+ }
+
+ /**
+ * Drain all records available from the given topic from the beginning
until the current highest
+ * offset.
+ *
+ * <p>This method will fetch the latest offsets for the partitions once
and only return records
+ * until that point.
+ *
+ * @param topic to fetch from
+ * @param properties used to configure the created {@link KafkaConsumer}
+ * @return all {@link ConsumerRecord} in the topic
+ * @throws KafkaException
+ */
+ public static List<ConsumerRecord<byte[], byte[]>>
drainAllRecordsFromTopic(
+ String topic, Properties properties) throws KafkaException {
+ final Properties consumerConfig = new Properties();
+ consumerConfig.putAll(properties);
+ consumerConfig.put("key.deserializer",
ByteArrayDeserializer.class.getName());
+ consumerConfig.put("value.deserializer",
ByteArrayDeserializer.class.getName());
+ try (KafkaConsumer<byte[], byte[]> consumer = new
KafkaConsumer<>(consumerConfig)) {
+ Set<TopicPartition> topicPartitions = getAllPartitions(consumer,
topic);
+ Map<TopicPartition, Long> endOffsets =
consumer.endOffsets(topicPartitions);
+ consumer.assign(topicPartitions);
+ consumer.seekToBeginning(topicPartitions);
+
+ final List<ConsumerRecord<byte[], byte[]>> consumerRecords = new
ArrayList<>();
+ while (!topicPartitions.isEmpty()) {
+ ConsumerRecords<byte[], byte[]> records =
consumer.poll(CONSUMER_POLL_DURATION);
+ LOG.debug("Fetched {} records from topic {}.",
records.count(), topic);
+
+ // Remove partitions from polling which have reached its end.
+ final List<TopicPartition> finishedPartitions = new
ArrayList<>();
+ for (final TopicPartition topicPartition : topicPartitions) {
+ final long position = consumer.position(topicPartition);
+ final long endOffset = endOffsets.get(topicPartition);
+ LOG.debug(
+ "Endoffset {} and current position {} for
partition {}",
+ endOffset,
+ position,
+ topicPartition.partition());
+ if (endOffset - position > 0) {
+ continue;
+ }
+ finishedPartitions.add(topicPartition);
+ }
+ if (topicPartitions.removeAll(finishedPartitions)) {
+ consumer.assign(topicPartitions);
+ }
+ for (ConsumerRecord<byte[], byte[]> r : records) {
+ consumerRecords.add(r);
+ }
+ }
+ return consumerRecords;
+ }
+ }
+
+ private static Set<TopicPartition> getAllPartitions(
+ KafkaConsumer<byte[], byte[]> consumer, String topic) {
+ return consumer.partitionsFor(topic).stream()
+ .map(info -> new TopicPartition(info.topic(),
info.partition()))
+ .collect(Collectors.toSet());
+ }
+}
diff --git
a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-kafka/src/test/resources/log4j2-test.properties
b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-kafka/src/test/resources/log4j2-test.properties
new file mode 100644
index 000000000..a9d045e0e
--- /dev/null
+++
b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-kafka/src/test/resources/log4j2-test.properties
@@ -0,0 +1,26 @@
+################################################################################
+# Copyright 2023 Ververica Inc.
+#
+# Licensed 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 OFF to not flood build logs
+# set manually to INFO for debugging purposes
+rootLogger.level=INFO
+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 eab79f486..c5265af9f 100644
--- a/flink-cdc-connect/flink-cdc-pipeline-connectors/pom.xml
+++ b/flink-cdc-connect/flink-cdc-pipeline-connectors/pom.xml
@@ -32,6 +32,7 @@ limitations under the License.
<module>flink-cdc-pipeline-connector-mysql</module>
<module>flink-cdc-pipeline-connector-doris</module>
<module>flink-cdc-pipeline-connector-starrocks</module>
+ <module>flink-cdc-pipeline-connector-kafka</module>
</modules>
<dependencies>
