This is an automated email from the ASF dual-hosted git repository.
davidzollo pushed a commit to branch dev
in repository https://gitbox.apache.org/repos/asf/seatunnel.git
The following commit(s) were added to refs/heads/dev by this push:
new 358ce6172c [Docs] Add verified MySQL CDC to Kafka recipe and E2E
coverage (#11466)
358ce6172c is described below
commit 358ce6172c5efd8da41438a8a0236bfb00d6ecc5
Author: Daniel <[email protected]>
AuthorDate: Sat Jul 18 10:07:04 2026 +0800
[Docs] Add verified MySQL CDC to Kafka recipe and E2E coverage (#11466)
Co-authored-by: DanielLeens <[email protected]>
---
.../locally/quick-start-seatunnel-engine.md | 2 +-
.../getting-started/locally/run-your-first-job.md | 4 +-
.../getting-started/recipes/mysql-cdc-to-kafka.md | 152 ++++++++
docs/en/getting-started/recipes/overview.md | 3 +
docs/sidebars.js | 1 +
.../locally/quick-start-seatunnel-engine.md | 2 +-
.../getting-started/locally/run-your-first-job.md | 4 +-
.../getting-started/recipes/mysql-cdc-to-kafka.md | 146 ++++++++
docs/zh/getting-started/recipes/overview.md | 3 +
.../e2e/connector/kafka/KafkaRecipeIT.java | 415 +++++++++++++++++++++
.../src/test/resources/ddl/recipe_shop_orders.sql | 31 ++
.../kafka/mysqlcdc_to_kafka_with_transforms.conf | 63 ++++
12 files changed, 822 insertions(+), 4 deletions(-)
diff --git a/docs/en/getting-started/locally/quick-start-seatunnel-engine.md
b/docs/en/getting-started/locally/quick-start-seatunnel-engine.md
index 6c4eff10e9..8e3bc369d5 100644
--- a/docs/en/getting-started/locally/quick-start-seatunnel-engine.md
+++ b/docs/en/getting-started/locally/quick-start-seatunnel-engine.md
@@ -245,7 +245,7 @@ Recommendation:
- Start with [Getting Started Overview](../overview.md) if you want a guided
reading path across deployment, quick start, and configuration.
- Use [Job Configuration Guide](../job-configuration-guide.md) when you are
ready to replace the sample source and sink with real connectors.
-- If you want runnable source-to-sink examples next, continue with [MySQL CDC
to Doris](../recipes/mysql-cdc-to-doris.md), [JDBC to
S3](../recipes/jdbc-to-s3.md), [Kafka to
Iceberg](../recipes/kafka-to-iceberg.md), [Http to
JDBC](../recipes/http-to-jdbc.md), [File to
StarRocks](../recipes/file-to-starrocks.md), or [Multi-table
CDC](../recipes/multi-table-cdc.md).
+- If you want a newly verified source-to-sink walkthrough next, continue with
[MySQL CDC to Kafka](../recipes/mysql-cdc-to-kafka.md). For other pipeline
shapes, see [MySQL CDC to Doris](../recipes/mysql-cdc-to-doris.md), [JDBC to
S3](../recipes/jdbc-to-s3.md), [Kafka to
Iceberg](../recipes/kafka-to-iceberg.md), [Http to
JDBC](../recipes/http-to-jdbc.md), [File to
StarRocks](../recipes/file-to-starrocks.md), or [Multi-table
CDC](../recipes/multi-table-cdc.md).
- Start writing your own config file, choose the
[connector](../../connectors/source) you want to use, and configure the
parameters according to the connector documentation.
- If you want to deploy a multi-node SeaTunnel Engine cluster, continue with
[SeaTunnel Engine(Zeta) Deployment](../../engines/zeta/deployment.md).
- See [SeaTunnel Engine(Zeta)](../../engines/zeta/about.md) if you want to
learn more about SeaTunnel Engine.
diff --git a/docs/en/getting-started/locally/run-your-first-job.md
b/docs/en/getting-started/locally/run-your-first-job.md
index 9bde8145bf..c0d42defbf 100644
--- a/docs/en/getting-started/locally/run-your-first-job.md
+++ b/docs/en/getting-started/locally/run-your-first-job.md
@@ -90,8 +90,10 @@ If this works, your basic local path is healthy and you can
move on to real pipe
## Next step
- For the full local walkthrough, continue with [Quick Start With SeaTunnel
Engine](quick-start-seatunnel-engine.md).
-- For runnable source-to-sink examples, start with these recipes:
+- For a newly verified source-to-sink walkthrough, start with [MySQL CDC to
Kafka](../recipes/mysql-cdc-to-kafka.md).
+- For other pipeline shapes, continue with:
- [MySQL CDC to Doris](../recipes/mysql-cdc-to-doris.md)
+ - [MySQL CDC to Kafka](../recipes/mysql-cdc-to-kafka.md)
- [JDBC to S3](../recipes/jdbc-to-s3.md)
- [Kafka to Iceberg](../recipes/kafka-to-iceberg.md)
- [Http to JDBC](../recipes/http-to-jdbc.md)
diff --git a/docs/en/getting-started/recipes/mysql-cdc-to-kafka.md
b/docs/en/getting-started/recipes/mysql-cdc-to-kafka.md
new file mode 100644
index 0000000000..235869f839
--- /dev/null
+++ b/docs/en/getting-started/recipes/mysql-cdc-to-kafka.md
@@ -0,0 +1,152 @@
+---
+sidebar_position: 2
+title: MySQL CDC to Kafka with metadata headers and field shaping
+---
+
+# MySQL CDC to Kafka with metadata headers and field shaping
+
+This page keeps the scope intentionally narrow: it documents the exact MySQL
CDC to Kafka job that was validated end to end in Docker on July 16, 2026.
+
+The verified pipeline is:
+
+- `MySQL-CDC` reads the snapshot and later binlog changes from `shop.orders`
+- `Metadata` exposes database, table, and row kind as normal fields
+- `Sql` renames business fields and fills custom fields
+- `Kafka` writes JSON payloads while moving selected metadata into Kafka
headers
+
+## Validation environment prerequisites
+
+The validated Docker run had these prerequisites in place before the job
started:
+
+- MySQL ran with the GTID/binlog settings from `docker/server-gtids/my.cnf`.
+- The MySQL JDBC driver JAR was present under the `MySQL-CDC` plugin `lib`
directory.
+- A CDC user named `st_user_source` existed with `SELECT`, `RELOAD`, `SHOW
DATABASES`, `REPLICATION SLAVE`, `REPLICATION CLIENT`, and `LOCK TABLES`
privileges.
+- Kafka topic `recipe_mysql_orders` was created before the SeaTunnel job
started.
+
+## Verified source data
+
+The E2E test created this MySQL table and loaded two initial rows:
+
+```sql
+CREATE TABLE orders (
+ id BIGINT NOT NULL PRIMARY KEY,
+ order_no VARCHAR(64) NOT NULL,
+ user_id BIGINT NOT NULL,
+ status INT NOT NULL,
+ amount DECIMAL(10, 2) NOT NULL
+);
+
+INSERT INTO orders (id, order_no, user_id, status, amount) VALUES
+ (1001, 'ORD-1001', 501, 0, 19.99),
+ (1002, 'ORD-1002', 502, 1, 29.99);
+```
+
+After the job started, the test executed these incremental changes:
+
+```sql
+UPDATE shop.orders SET status = 2, amount = 39.99 WHERE id = 1001;
+
+INSERT INTO shop.orders (id, order_no, user_id, status, amount) VALUES
+ (1003, 'ORD-1003', 503, 0, 59.99);
+```
+
+## Exact job config validated in Docker
+
+The following config is the exact job config that was validated in the Docker
test environment. The hostnames are Docker network aliases from that
environment.
+
+```hocon
+env {
+ parallelism = 1
+ job.mode = "STREAMING"
+}
+
+source {
+ MySQL-CDC {
+ plugin_output = "mysql_orders_raw"
+ url = "jdbc:mysql://mysql_cdc_e2e:3306/shop"
+ username = "st_user_source"
+ password = "mysqlpw"
+ server-id = 5601-5604
+ table-names = ["shop.orders"]
+ startup.mode = "initial"
+ schema-changes.enabled = false
+ }
+}
+
+transform {
+ Metadata {
+ plugin_input = "mysql_orders_raw"
+ plugin_output = "mysql_orders_with_meta"
+ metadata_fields {
+ Database = source_database
+ Table = source_table
+ RowKind = change_type
+ }
+ }
+
+ Sql {
+ plugin_input = "mysql_orders_with_meta"
+ plugin_output = "kafka_orders"
+ query = "select id as order_id, order_no, user_id, amount, case when
status = 0 then 'CREATED' when status = 1 then 'PAID' when status = 2 then
'SHIPPED' else 'OTHER' end as status_name, source_database, source_table,
change_type, CONCAT(source_database, '.', source_table) as source_name,
'mysql_cdc' as sync_source from dual where id is not null"
+ }
+}
+
+sink {
+ Kafka {
+ plugin_input = "kafka_orders"
+ bootstrap.servers = "kafkaCluster:9092"
+ topic = "recipe_mysql_orders"
+ format = json
+ partition_key_fields = ["order_id"]
+ kafka_headers_fields = ["source_database", "source_table", "change_type"]
+ }
+}
+```
+
+## What the validation proved
+
+The Docker E2E test asserted all of the following:
+
+- The snapshot phase produced Kafka records for the initial MySQL rows.
+- Record `1001` was written with payload fields `order_id`, `status_name`,
`source_name`, and `sync_source`.
+- On snapshot record `1001`, Kafka headers contained `source_database=shop`,
`source_table=orders`, and `change_type=+I`.
+- On the same snapshot record, those metadata fields were removed from the
JSON payload after `kafka_headers_fields` was applied.
+- After `UPDATE shop.orders SET status = 2 ... WHERE id = 1001`, the latest
record for `1001` carried `change_type=+U` and `status_name=SHIPPED`.
+- After inserting `1003`, the latest record for `1003` carried
`change_type=+I` and `status_name=CREATED`.
+
+## Why each step exists in this recipe
+
+### 1. `Metadata` exposes CDC fields that the sink can reuse
+
+This validated job only exposed the fields that were later consumed by the
Kafka sink or SQL transform:
+
+- `source_database`
+- `source_table`
+- `change_type`
+
+### 2. `Sql` keeps business shaping in one place
+
+In this validated run, the SQL produced these observed results:
+
+- row `1001` was written with `status_name=CREATED` during the snapshot phase
+- row `1002` was written with `status_name=PAID` during the snapshot phase
+- row `1001` later changed to `status_name=SHIPPED` after the update
+- row `1003` was written with `status_name=CREATED` after the insert
+- the asserted records also included `sync_source`, and snapshot row `1001`
included `source_name`
+
+### 3. `kafka_headers_fields` changes both headers and payload
+
+This behavior was asserted on the snapshot record for row `1001`. After these
three fields were moved into Kafka headers:
+
+- `source_database`
+- `source_table`
+- `change_type`
+
+they were no longer present in that JSON payload.
+
+## Related docs
+
+- [`MySQL-CDC` source connector](../../connectors/source/MySQL-CDC.md)
+- [`Kafka` sink connector](../../connectors/sink/Kafka.md)
+- [`Metadata` transform](../../transforms/metadata.md)
+- [`Sql` transform](../../transforms/sql.md)
diff --git a/docs/en/getting-started/recipes/overview.md
b/docs/en/getting-started/recipes/overview.md
index 9bab5d5a73..053d24c262 100644
--- a/docs/en/getting-started/recipes/overview.md
+++ b/docs/en/getting-started/recipes/overview.md
@@ -6,11 +6,14 @@ slug: /getting-started/recipes
These recipes are best read after your first local job succeeds. Instead of
reading every example in order, start with the pipeline shape that is closest
to your real source and sink.
+The MySQL CDC to Kafka recipe in this section was cross-checked against a
Docker E2E run on July 16, 2026. Its config, observed results, and
prerequisites were all aligned with that verified path before this page was
updated.
+
## Choose A Recipe By Pipeline Goal
| Goal | Start here |
| --- | --- |
| CDC from MySQL into an analytics database | [MySQL CDC to
Doris](./mysql-cdc-to-doris.md) |
+| CDC from MySQL into Kafka with metadata headers | [MySQL CDC to
Kafka](./mysql-cdc-to-kafka.md) |
| JDBC extraction into object storage | [JDBC to S3](./jdbc-to-s3.md) |
| Streaming from Kafka into a table format | [Kafka to
Iceberg](./kafka-to-iceberg.md) |
| HTTP ingestion into a relational target | [HTTP to JDBC](./http-to-jdbc.md) |
diff --git a/docs/sidebars.js b/docs/sidebars.js
index f6eb419f93..95765c4daa 100644
--- a/docs/sidebars.js
+++ b/docs/sidebars.js
@@ -161,6 +161,7 @@ const sidebars = {
},
"items": [
"getting-started/recipes/mysql-cdc-to-doris",
+ "getting-started/recipes/mysql-cdc-to-kafka",
"getting-started/recipes/jdbc-to-s3",
"getting-started/recipes/kafka-to-iceberg",
"getting-started/recipes/http-to-jdbc",
diff --git a/docs/zh/getting-started/locally/quick-start-seatunnel-engine.md
b/docs/zh/getting-started/locally/quick-start-seatunnel-engine.md
index 1e2ca3cb8d..9fc7cf38f2 100644
--- a/docs/zh/getting-started/locally/quick-start-seatunnel-engine.md
+++ b/docs/zh/getting-started/locally/quick-start-seatunnel-engine.md
@@ -243,7 +243,7 @@ Total Failed Count : 0
- 如果你想先建立整体路径感，可以返回阅读[快速入门总览](../overview.md)。
- 当你准备把示例 Source 和 Sink
替换成真实连接器时，建议继续阅读[作业配置指南](../job-configuration-guide.md)。
-- 如果你下一步就想看可直接照着改的源端到目标端示例，可以继续看 [MySQL CDC 到
Doris](../recipes/mysql-cdc-to-doris.md)、[JDBC 到
S3](../recipes/jdbc-to-s3.md)、[Kafka 到
Iceberg](../recipes/kafka-to-iceberg.md)、[Http 到
JDBC](../recipes/http-to-jdbc.md)、[File 到
StarRocks](../recipes/file-to-starrocks.md) 和 [多表
CDC](../recipes/multi-table-cdc.md)。
+- 如果你下一步想看一条刚完成端到端核对的新教程，可以先继续看 [MySQL CDC 到
Kafka](../recipes/mysql-cdc-to-kafka.md)。如果你想按别的链路形态继续读，再看 [MySQL CDC 到
Doris](../recipes/mysql-cdc-to-doris.md)、[JDBC 到
S3](../recipes/jdbc-to-s3.md)、[Kafka 到
Iceberg](../recipes/kafka-to-iceberg.md)、[Http 到
JDBC](../recipes/http-to-jdbc.md)、[File 到
StarRocks](../recipes/file-to-starrocks.md) 和 [多表
CDC](../recipes/multi-table-cdc.md)。
- 开始编写您自己的配置文件，选择您想要使用的[连接器](../../connectors/source)，并根据连接器的文档配置参数。
- 如果您要部署多节点 SeaTunnel Engine 集群，请继续阅读[SeaTunnel Engine(Zeta)
安装部署](../../engines/zeta/deployment.md)。
- 如果您想进一步了解 SeaTunnel Engine，请参阅[SeaTunnel引擎](../../engines/zeta/about.md)。
diff --git a/docs/zh/getting-started/locally/run-your-first-job.md
b/docs/zh/getting-started/locally/run-your-first-job.md
index aaadfadf00..8a1b50cd8f 100644
--- a/docs/zh/getting-started/locally/run-your-first-job.md
+++ b/docs/zh/getting-started/locally/run-your-first-job.md
@@ -90,8 +90,10 @@ cd "apache-seatunnel-${version}"
## 下一步
- 如果你想看完整本地链路，请继续看 [SeaTunnel 引擎快速开始](quick-start-seatunnel-engine.md)。
-- 如果你想直接看真实源端到目标端示例，请从下面这些教程开始：
+- 如果你想直接看一条刚完成端到端核对的新教程，请先看 [MySQL CDC 到
Kafka](../recipes/mysql-cdc-to-kafka.md)。
+- 如果你想按别的链路形态继续阅读，再看下面这些示例：
- [MySQL CDC 到 Doris](../recipes/mysql-cdc-to-doris.md)
+ - [MySQL CDC 到 Kafka](../recipes/mysql-cdc-to-kafka.md)
- [JDBC 到 S3](../recipes/jdbc-to-s3.md)
- [Kafka 到 Iceberg](../recipes/kafka-to-iceberg.md)
- [Http 到 JDBC](../recipes/http-to-jdbc.md)
diff --git a/docs/zh/getting-started/recipes/mysql-cdc-to-kafka.md
b/docs/zh/getting-started/recipes/mysql-cdc-to-kafka.md
new file mode 100644
index 0000000000..86d0ed84b5
--- /dev/null
+++ b/docs/zh/getting-started/recipes/mysql-cdc-to-kafka.md
@@ -0,0 +1,146 @@
+---
+sidebar_position: 2
+title: MySQL CDC 到 Kafka：消息头元数据与字段整形
+---
+
+# MySQL CDC 到 Kafka：消息头元数据与字段整形
+
+这篇文档只讲一条已经在 Docker 里完成端到端验证的链路。验证时间是 2026 年 7 月 16
日，下面保留的配置、输入数据和结果，都是这次实测里真正跑过并断言过的内容。
+
+这条已验证链路是：
+
+- `MySQL-CDC` 读取 `shop.orders` 的快照和后续 binlog
+- `Metadata` 暴露库名、表名和变更类型
+- `Sql` 重命名字段并补充业务字段
+- `Kafka` 输出 JSON payload，同时把部分元数据写入消息头
+
+## 这次验证环境具备的前置条件
+
+这条 Docker 端到端验证在任务启动前，已经具备了下面这些条件：
+
+- MySQL 使用了 `docker/server-gtids/my.cnf` 里的 GTID / binlog 配置。
+- `MySQL-CDC` 插件的 `lib` 目录里已经放入了 MySQL JDBC driver JAR。
+- 预先创建了名为 `st_user_source` 的 CDC 用户，并授予了 `SELECT`、`RELOAD`、`SHOW
DATABASES`、`REPLICATION SLAVE`、`REPLICATION CLIENT`、`LOCK TABLES` 权限。
+- SeaTunnel 任务启动前，Kafka topic `recipe_mysql_orders` 已经提前创建好。
+
+## 已验证的源表数据
+
+E2E 测试先创建了下面这张表，并插入两条初始数据：
+
+```sql
+CREATE TABLE orders (
+ id BIGINT NOT NULL PRIMARY KEY,
+ order_no VARCHAR(64) NOT NULL,
+ user_id BIGINT NOT NULL,
+ status INT NOT NULL,
+ amount DECIMAL(10, 2) NOT NULL
+);
+
+INSERT INTO orders (id, order_no, user_id, status, amount) VALUES
+ (1001, 'ORD-1001', 501, 0, 19.99),
+ (1002, 'ORD-1002', 502, 1, 29.99);
+```
+
+任务启动后，测试又执行了这两条增量变更：
+
+```sql
+UPDATE shop.orders SET status = 2, amount = 39.99 WHERE id = 1001;
+
+INSERT INTO shop.orders (id, order_no, user_id, status, amount) VALUES
+ (1003, 'ORD-1003', 503, 0, 59.99);
+```
+
+## Docker 实测通过的完整配置
+
+下面这份配置就是 Docker 测试环境里真实跑通过的那份作业配置。里面的主机名是当时测试网络里的服务别名。
+
+```hocon
+env {
+ parallelism = 1
+ job.mode = "STREAMING"
+}
+
+source {
+ MySQL-CDC {
+ plugin_output = "mysql_orders_raw"
+ url = "jdbc:mysql://mysql_cdc_e2e:3306/shop"
+ username = "st_user_source"
+ password = "mysqlpw"
+ server-id = 5601-5604
+ table-names = ["shop.orders"]
+ startup.mode = "initial"
+ schema-changes.enabled = false
+ }
+}
+
+transform {
+ Metadata {
+ plugin_input = "mysql_orders_raw"
+ plugin_output = "mysql_orders_with_meta"
+ metadata_fields {
+ Database = source_database
+ Table = source_table
+ RowKind = change_type
+ }
+ }
+
+ Sql {
+ plugin_input = "mysql_orders_with_meta"
+ plugin_output = "kafka_orders"
+ query = "select id as order_id, order_no, user_id, amount, case when
status = 0 then 'CREATED' when status = 1 then 'PAID' when status = 2 then
'SHIPPED' else 'OTHER' end as status_name, source_database, source_table,
change_type, CONCAT(source_database, '.', source_table) as source_name,
'mysql_cdc' as sync_source from dual where id is not null"
+ }
+}
+
+sink {
+ Kafka {
+ plugin_input = "kafka_orders"
+ bootstrap.servers = "kafkaCluster:9092"
+ topic = "recipe_mysql_orders"
+ format = json
+ partition_key_fields = ["order_id"]
+ kafka_headers_fields = ["source_database", "source_table", "change_type"]
+ }
+}
+```
+
+## 这次端到端验证实际证明了什么
+
+Docker E2E 测试对下面这些结果做了断言：
+
+- 快照阶段成功把初始 MySQL 数据写进了 Kafka。
+- 订单 `1001` 的 payload 中包含 `order_id`、`status_name`、`source_name`、`sync_source`。
+- 在快照阶段 `1001` 这条记录上，Kafka 消息头里包含
`source_database=shop`、`source_table=orders`、`change_type=+I`。
+- 在同一条快照记录上，配置了 `kafka_headers_fields` 之后，这三个字段不会再留在 JSON payload 里。
+- 执行 `UPDATE shop.orders SET status = 2 ... WHERE id = 1001` 后，`1001` 最新一条消息的
`change_type=+U`，`status_name=SHIPPED`。
+- 插入 `1003` 后，`1003` 最新一条消息的 `change_type=+I`，`status_name=CREATED`。
+
+## 为什么这条链路这样写
+
+### 1. `Metadata` 只暴露了后续真的会用到的 CDC 字段
+
+这份已验证配置里，只补了三个字段：
+
+- `source_database`
+- `source_table`
+- `change_type`
+
+### 2. `Sql` 负责统一做业务整形
+
+这次已验证链路里，SQL 实际产出的结果是：
+
+- 快照阶段，`1001` 被写成 `status_name=CREATED`
+- 快照阶段，`1002` 被写成 `status_name=PAID`
+- 更新之后，`1001` 变成了 `status_name=SHIPPED`
+- 插入之后，`1003` 被写成 `status_name=CREATED`
+- 这次被断言的记录里还包含了 `sync_source`，其中快照阶段的 `1001` 还包含 `source_name`
+
+### 3. `kafka_headers_fields` 会同时影响 header 和 payload
+
+这个行为是在快照阶段 `1001` 那条记录上明确断言过的：`source_database`、`source_table`、`change_type`
被写入 Kafka header 后，就不会继续保留在那条 JSON payload 里。
+
+## 相关文档
+
+- [`MySQL-CDC` source 连接器](../../connectors/source/MySQL-CDC.md)
+- [`Kafka` sink 连接器](../../connectors/sink/Kafka.md)
+- [`Metadata` transform](../../transforms/metadata.md)
+- [`Sql` transform](../../transforms/sql.md)
diff --git a/docs/zh/getting-started/recipes/overview.md
b/docs/zh/getting-started/recipes/overview.md
index a0b1d383f1..991fd9de94 100644
--- a/docs/zh/getting-started/recipes/overview.md
+++ b/docs/zh/getting-started/recipes/overview.md
@@ -6,11 +6,14 @@ slug: /getting-started/recipes
这些示例更适合在你已经跑通第一个本地任务之后再阅读。不要按顺序把所有示例都看一遍，而是优先找到最接近你真实 source 和 sink 组合的那条链路。
+本节里的 “MySQL CDC 到 Kafka” 这篇教程，在 2026 年 7 月 16 日对照过一条真实的 Docker E2E
链路。页面里的配置、观测结果和前置条件，都是先和那次验证结果核对一致后才更新的。
+
## 按业务目标选择示例
| 目标 | 推荐入口 |
| --- | --- |
| 从 MySQL CDC 同步到分析型数据库 | [MySQL CDC 到 Doris](./mysql-cdc-to-doris.md) |
+| 从 MySQL CDC 同步到 Kafka，并附带元数据消息头 | [MySQL CDC 到
Kafka](./mysql-cdc-to-kafka.md) |
| 把 JDBC 数据抽取到对象存储 | [JDBC 到 S3](./jdbc-to-s3.md) |
| 从 Kafka 流式写入表格式存储 | [Kafka 到 Iceberg](./kafka-to-iceberg.md) |
| 把 HTTP 数据写入关系型数据库 | [HTTP 到 JDBC](./http-to-jdbc.md) |
diff --git
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-kafka-e2e/src/test/java/org/apache/seatunnel/e2e/connector/kafka/KafkaRecipeIT.java
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-kafka-e2e/src/test/java/org/apache/seatunnel/e2e/connector/kafka/KafkaRecipeIT.java
new file mode 100644
index 0000000000..ea64eeb406
--- /dev/null
+++
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-kafka-e2e/src/test/java/org/apache/seatunnel/e2e/connector/kafka/KafkaRecipeIT.java
@@ -0,0 +1,415 @@
+/*
+ * 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.seatunnel.e2e.connector.kafka;
+
+import org.apache.seatunnel.shade.com.fasterxml.jackson.databind.ObjectMapper;
+import
org.apache.seatunnel.shade.com.fasterxml.jackson.databind.node.ObjectNode;
+
+import
org.apache.seatunnel.connectors.seatunnel.cdc.mysql.testutils.MySqlContainer;
+import
org.apache.seatunnel.connectors.seatunnel.cdc.mysql.testutils.MySqlVersion;
+import
org.apache.seatunnel.connectors.seatunnel.cdc.mysql.testutils.UniqueDatabase;
+import org.apache.seatunnel.e2e.common.TestResource;
+import org.apache.seatunnel.e2e.common.TestSuiteBase;
+import org.apache.seatunnel.e2e.common.container.ContainerExtendedFactory;
+import org.apache.seatunnel.e2e.common.container.EngineType;
+import org.apache.seatunnel.e2e.common.container.TestContainer;
+import org.apache.seatunnel.e2e.common.junit.DisabledOnContainer;
+import org.apache.seatunnel.e2e.common.junit.TestContainerExtension;
+import org.apache.seatunnel.e2e.common.util.JobIdGenerator;
+
+import org.apache.kafka.clients.admin.AdminClient;
+import org.apache.kafka.clients.admin.AdminClientConfig;
+import org.apache.kafka.clients.admin.NewTopic;
+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.clients.consumer.OffsetResetStrategy;
+import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.common.header.Header;
+import org.apache.kafka.common.header.Headers;
+import org.apache.kafka.common.serialization.StringDeserializer;
+
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.TestTemplate;
+import org.testcontainers.containers.Container;
+import org.testcontainers.containers.KafkaContainer;
+import org.testcontainers.containers.output.Slf4jLogConsumer;
+import org.testcontainers.lifecycle.Startables;
+import org.testcontainers.shaded.org.awaitility.Awaitility;
+import org.testcontainers.utility.DockerImageName;
+import org.testcontainers.utility.DockerLoggerFactory;
+
+import lombok.extern.slf4j.Slf4j;
+
+import java.io.IOException;
+import java.nio.charset.StandardCharsets;
+import java.sql.Connection;
+import java.sql.DriverManager;
+import java.sql.SQLException;
+import java.sql.Statement;
+import java.time.Duration;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Properties;
+import java.util.UUID;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicReference;
+import java.util.stream.Stream;
+
+/**
+ * Validates the documented MySQL CDC to Kafka recipe with metadata enrichment
and SQL field
+ * shaping.
+ */
+@Slf4j
+@DisabledOnContainer(
+ value = {},
+ type = {EngineType.SPARK, EngineType.FLINK},
+ disabledReason =
+ "This recipe validates a MySQL CDC streaming pipeline and is
kept on the zeta engine to match the getting-started documentation path.")
+public class KafkaRecipeIT extends TestSuiteBase implements TestResource {
+
+ /** Kafka image used by the documented recipe scenario. */
+ private static final String KAFKA_IMAGE_NAME =
"confluentinc/cp-kafka:7.0.9";
+ /** Network alias referenced by the recipe config. */
+ private static final String KAFKA_HOST = "kafkaCluster";
+ /** MySQL hostname referenced by the recipe config. */
+ private static final String MYSQL_HOST = "mysql_cdc_e2e";
+ /** Source database consumed by the CDC job. */
+ private static final String MYSQL_DATABASE = "shop";
+ /** Application user used for DML verification. */
+ private static final String MYSQL_USER_NAME = "mysqluser";
+ /** Shared password for the MySQL test users. */
+ private static final String MYSQL_USER_PASSWORD = "mysqlpw";
+ /** Administrative MySQL user used to create the CDC account. */
+ private static final String MYSQL_ROOT_USER_NAME = "root";
+ /** CDC reader account configured inside the documented job. */
+ private static final String MYSQL_CDC_USER_NAME = "st_user_source";
+ /** Kafka topic populated by the documented sink config. */
+ private static final String TOPIC_NAME = "recipe_mysql_orders";
+ /** Driver JAR injected into the connector plugin before the job starts. */
+ private static final String DRIVER_JAR =
+
"https://repo1.maven.org/maven2/com/mysql/mysql-connector-j/8.0.32/mysql-connector-j-8.0.32.jar";;
+
+ /** Kafka service used by the recipe job. */
+ private KafkaContainer kafkaContainer;
+ /** MySQL service used by the recipe job. */
+ private MySqlContainer mysqlContainer;
+ /** Helper that materializes the documented source table and seed rows. */
+ private UniqueDatabase shopDatabase;
+
+ /**
+ * Injects the MySQL JDBC driver into the MySQL-CDC plugin directory so
the recipe can run in
+ * the same way as the documented Docker validation.
+ */
+ @TestContainerExtension
+ private final ContainerExtendedFactory extendedFactory =
+ container -> {
+ Container.ExecResult extraCommands =
+ container.execInContainer(
+ "bash",
+ "-c",
+ "mkdir -p /tmp/seatunnel/plugins/MySQL-CDC/lib
&& cd /tmp/seatunnel/plugins/MySQL-CDC/lib && wget "
+ + DRIVER_JAR);
+ Assertions.assertEquals(0, extraCommands.getExitCode(),
extraCommands.getStderr());
+ };
+
+ /** Starts the external systems and prepares the documented source and
sink prerequisites. */
+ @BeforeEach
+ @Override
+ public void startUp() throws Exception {
+ kafkaContainer =
+ new KafkaContainer(DockerImageName.parse(KAFKA_IMAGE_NAME))
+ .withNetwork(NETWORK)
+ .withNetworkAliases(KAFKA_HOST)
+ .withLogConsumer(
+ new Slf4jLogConsumer(
+
DockerLoggerFactory.getLogger(KAFKA_IMAGE_NAME)));
+ mysqlContainer =
+ new MySqlContainer(MySqlVersion.V8_0)
+
.withConfigurationOverride("docker/server-gtids/my.cnf")
+ .withNetwork(NETWORK)
+ .withNetworkAliases(MYSQL_HOST)
+ .withDatabaseName(MYSQL_DATABASE)
+ .withUsername(MYSQL_USER_NAME)
+ .withPassword(MYSQL_USER_PASSWORD)
+ .withLogConsumer(
+ new Slf4jLogConsumer(
+
DockerLoggerFactory.getLogger("mysql-docker-image")));
+ Startables.deepStart(Stream.of(kafkaContainer, mysqlContainer)).join();
+ createCdcUser();
+ shopDatabase = new UniqueDatabase(mysqlContainer, MYSQL_DATABASE);
+
shopDatabase.setTemplateName("recipe_shop_orders").createAndInitialize();
+ createKafkaTopic(TOPIC_NAME);
+ }
+
+ /** Verifies snapshot rows, updates, inserts, payload shaping, and Kafka
headers. */
+ @TestTemplate
+ public void testMysqlCdcToKafkaWithTransforms(TestContainer container)
+ throws InterruptedException {
+ String jobId = String.valueOf(JobIdGenerator.newJobId());
+ AtomicReference<Throwable> jobFailure = new AtomicReference<>();
+ CompletableFuture.runAsync(
+ () -> {
+ try {
+ container.executeJob(
+
"/kafka/mysqlcdc_to_kafka_with_transforms.conf", jobId);
+ } catch (Exception e) {
+ jobFailure.set(e);
+ throw new RuntimeException(e);
+ }
+ });
+
+ Awaitility.await()
+ .atMost(120, TimeUnit.SECONDS)
+ .pollInterval(Duration.ofSeconds(3))
+ .untilAsserted(
+ () -> {
+ assertNoJobFailure(jobFailure);
+ List<ConsumerRecord<String, String>> records =
+ getKafkaRecordData(TOPIC_NAME);
+ Assertions.assertTrue(records.size() >= 2,
"snapshot records missing");
+ ConsumerRecord<String, String> snapshot1001 =
+ findLatestRecordByOrderId(records, 1001L);
+ ConsumerRecord<String, String> snapshot1002 =
+ findLatestRecordByOrderId(records, 1002L);
+ Assertions.assertNotNull(snapshot1001);
+ Assertions.assertNotNull(snapshot1002);
+ Map<String, String> headers =
+
convertHeadersToMap(snapshot1001.headers());
+ Assertions.assertEquals("shop",
headers.get("source_database"));
+ Assertions.assertEquals("orders",
headers.get("source_table"));
+ Assertions.assertEquals("+I",
headers.get("change_type"));
+
+ ObjectNode payload =
parsePayload(snapshot1001.value());
+ Assertions.assertEquals(1001L,
payload.get("order_id").asLong());
+ Assertions.assertEquals("CREATED",
payload.get("status_name").asText());
+ Assertions.assertEquals(
+ "shop.orders",
payload.get("source_name").asText());
+ Assertions.assertEquals(
+ "mysql_cdc",
payload.get("sync_source").asText());
+
Assertions.assertFalse(payload.has("source_database"));
+
Assertions.assertFalse(payload.has("source_table"));
+ Assertions.assertFalse(payload.has("change_type"));
+
+ ObjectNode payload1002 =
parsePayload(snapshot1002.value());
+ Assertions.assertEquals(1002L,
payload1002.get("order_id").asLong());
+ Assertions.assertEquals(
+ "PAID",
payload1002.get("status_name").asText());
+ });
+
+ executeSql(
+ "UPDATE shop.orders SET status = 2, amount = 39.99 WHERE id =
1001",
+ "INSERT INTO shop.orders (id, order_no, user_id, status,
amount) VALUES "
+ + "(1003, 'ORD-1003', 503, 0, 59.99)");
+
+ Awaitility.await()
+ .atMost(120, TimeUnit.SECONDS)
+ .pollInterval(Duration.ofSeconds(3))
+ .untilAsserted(
+ () -> {
+ assertNoJobFailure(jobFailure);
+ List<ConsumerRecord<String, String>> records =
+ getKafkaRecordData(TOPIC_NAME);
+ Assertions.assertTrue(records.size() >= 4, "cdc
records missing");
+
+ ConsumerRecord<String, String> latest1001 =
+ findLatestRecordByOrderId(records, 1001L);
+ ConsumerRecord<String, String> latest1003 =
+ findLatestRecordByOrderId(records, 1003L);
+ Assertions.assertNotNull(latest1001);
+ Assertions.assertNotNull(latest1003);
+
+ Map<String, String> headers1001 =
+ convertHeadersToMap(latest1001.headers());
+ Map<String, String> headers1003 =
+ convertHeadersToMap(latest1003.headers());
+ Assertions.assertEquals("+U",
headers1001.get("change_type"));
+ Assertions.assertEquals("+I",
headers1003.get("change_type"));
+
+ ObjectNode payload1001 =
parsePayload(latest1001.value());
+ ObjectNode payload1003 =
parsePayload(latest1003.value());
+ Assertions.assertEquals(
+ "SHIPPED",
payload1001.get("status_name").asText());
+ Assertions.assertEquals(
+ "CREATED",
payload1003.get("status_name").asText());
+ Assertions.assertEquals(
+ "mysql_cdc",
payload1001.get("sync_source").asText());
+ Assertions.assertEquals(
+ "mysql_cdc",
payload1003.get("sync_source").asText());
+ });
+ }
+
+ /** Releases the Kafka and MySQL containers after each validation run. */
+ @AfterEach
+ @Override
+ public void tearDown() {
+ if (kafkaContainer != null) {
+ kafkaContainer.close();
+ }
+ if (mysqlContainer != null) {
+ mysqlContainer.close();
+ }
+ }
+
+ /** Fails the awaiting assertion as soon as the async job thread surfaces
an exception. */
+ private static void assertNoJobFailure(AtomicReference<Throwable>
jobFailure) {
+ if (jobFailure.get() != null) {
+ Assertions.fail(jobFailure.get());
+ }
+ }
+
+ /** Creates the Kafka topic expected by the documented sink config. */
+ private void createKafkaTopic(String topicName) {
+ Properties adminProps = new Properties();
+ adminProps.put(
+ AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG,
kafkaContainer.getBootstrapServers());
+ try (AdminClient adminClient = AdminClient.create(adminProps)) {
+ NewTopic topic = new NewTopic(topicName, 1, (short) 1);
+ adminClient.createTopics(Arrays.asList(topic)).all().get(60,
TimeUnit.SECONDS);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw new IllegalStateException(
+ "Interrupted while creating Kafka topic " + topicName, e);
+ } catch (ExecutionException | java.util.concurrent.TimeoutException e)
{
+ throw new IllegalStateException("Failed to create Kafka topic " +
topicName, e);
+ }
+ }
+
+ /** Applies source-table updates that should be captured by the CDC job. */
+ private void executeSql(String... statements) {
+ try (Connection connection =
+ DriverManager.getConnection(
+ mysqlContainer.getJdbcUrl(MYSQL_DATABASE),
+ MYSQL_USER_NAME,
+ MYSQL_USER_PASSWORD);
+ Statement statement = connection.createStatement()) {
+ for (String sql : statements) {
+ statement.execute(sql);
+ }
+ } catch (SQLException e) {
+ throw new IllegalStateException("Failed to execute MySQL SQL", e);
+ }
+ }
+
+ /** Creates the CDC account with the privileges required by the recipe. */
+ private void createCdcUser() {
+ executeAdminSql(
+ "CREATE USER IF NOT EXISTS '"
+ + MYSQL_CDC_USER_NAME
+ + "'@'%' IDENTIFIED BY '"
+ + MYSQL_USER_PASSWORD
+ + "'",
+ "GRANT SELECT, RELOAD, SHOW DATABASES, REPLICATION SLAVE,
REPLICATION CLIENT, LOCK TABLES ON *.* TO '"
+ + MYSQL_CDC_USER_NAME
+ + "'@'%'");
+ }
+
+ /** Executes administrative SQL statements as MySQL root. */
+ private void executeAdminSql(String... statements) {
+ try (Connection connection =
+ DriverManager.getConnection(
+ mysqlContainer.getJdbcUrl(MYSQL_DATABASE),
+ MYSQL_ROOT_USER_NAME,
+ MYSQL_USER_PASSWORD);
+ Statement statement = connection.createStatement()) {
+ for (String sql : statements) {
+ statement.execute(sql);
+ }
+ } catch (SQLException e) {
+ throw new IllegalStateException("Failed to execute MySQL admin
SQL", e);
+ }
+ }
+
+ /** Builds a fresh consumer configuration so every assertion reads from
the earliest offset. */
+ private Properties kafkaConsumerConfig() {
+ Properties props = new Properties();
+ props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,
kafkaContainer.getBootstrapServers());
+ props.put(ConsumerConfig.GROUP_ID_CONFIG, "recipe-kafka-" +
UUID.randomUUID());
+ props.put(
+ ConsumerConfig.AUTO_OFFSET_RESET_CONFIG,
+ OffsetResetStrategy.EARLIEST.toString().toLowerCase());
+ props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,
StringDeserializer.class);
+ props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,
StringDeserializer.class);
+ return props;
+ }
+
+ /** Reads the current records from the recipe topic in offset order. */
+ private List<ConsumerRecord<String, String>> getKafkaRecordData(String
topicName) {
+ KafkaConsumer<String, String> consumer = null;
+ try {
+ List<ConsumerRecord<String, String>> data = new ArrayList<>();
+ consumer = new KafkaConsumer<>(kafkaConsumerConfig());
+ consumer.subscribe(Arrays.asList(topicName));
+ Map<TopicPartition, Long> offsets =
+ consumer.endOffsets(Arrays.asList(new
TopicPartition(topicName, 0)));
+ long endOffset = offsets.entrySet().iterator().next().getValue();
+ long lastProcessedOffset = -1L;
+
+ do {
+ ConsumerRecords<String, String> records =
consumer.poll(Duration.ofMillis(100));
+ for (ConsumerRecord<String, String> record : records) {
+ if (lastProcessedOffset < record.offset()) {
+ data.add(record);
+ }
+ lastProcessedOffset = record.offset();
+ }
+ } while (lastProcessedOffset < endOffset - 1);
+ return data;
+ } finally {
+ if (consumer != null) {
+ consumer.close();
+ }
+ }
+ }
+
+ /** Returns the latest Kafka record for one business key so CDC updates
can be asserted. */
+ private static ConsumerRecord<String, String> findLatestRecordByOrderId(
+ List<ConsumerRecord<String, String>> records, long orderId) throws
IOException {
+ ConsumerRecord<String, String> latest = null;
+ ObjectMapper objectMapper = new ObjectMapper();
+ for (ConsumerRecord<String, String> record : records) {
+ ObjectNode payload = objectMapper.readValue(record.value(),
ObjectNode.class);
+ if (payload.has("order_id") && payload.get("order_id").asLong() ==
orderId) {
+ latest = record;
+ }
+ }
+ return latest;
+ }
+
+ /** Converts Kafka headers into a string map for stable assertions. */
+ private static Map<String, String> convertHeadersToMap(Headers headers) {
+ Map<String, String> map = new HashMap<>();
+ for (Header header : headers) {
+ map.put(header.key(), new String(header.value(),
StandardCharsets.UTF_8));
+ }
+ return map;
+ }
+
+ /** Parses a JSON Kafka payload into a mutable object tree for field-level
checks. */
+ private static ObjectNode parsePayload(String payload) throws IOException {
+ return new ObjectMapper().readValue(payload, ObjectNode.class);
+ }
+}
diff --git
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-kafka-e2e/src/test/resources/ddl/recipe_shop_orders.sql
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-kafka-e2e/src/test/resources/ddl/recipe_shop_orders.sql
new file mode 100644
index 0000000000..78cc4b0581
--- /dev/null
+++
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-kafka-e2e/src/test/resources/ddl/recipe_shop_orders.sql
@@ -0,0 +1,31 @@
+--
+-- 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.
+--
+
+USE $DBNAME$;
+
+DROP TABLE IF EXISTS orders;
+CREATE TABLE orders (
+ id BIGINT NOT NULL PRIMARY KEY,
+ order_no VARCHAR(64) NOT NULL,
+ user_id BIGINT NOT NULL,
+ status INT NOT NULL,
+ amount DECIMAL(10, 2) NOT NULL
+);
+
+INSERT INTO orders (id, order_no, user_id, status, amount) VALUES
+ (1001, 'ORD-1001', 501, 0, 19.99),
+ (1002, 'ORD-1002', 502, 1, 29.99);
diff --git
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-kafka-e2e/src/test/resources/kafka/mysqlcdc_to_kafka_with_transforms.conf
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-kafka-e2e/src/test/resources/kafka/mysqlcdc_to_kafka_with_transforms.conf
new file mode 100644
index 0000000000..6d652dfcb1
--- /dev/null
+++
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-kafka-e2e/src/test/resources/kafka/mysqlcdc_to_kafka_with_transforms.conf
@@ -0,0 +1,63 @@
+#
+# 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.
+#
+
+env {
+ parallelism = 1
+ job.mode = "STREAMING"
+}
+
+source {
+ MySQL-CDC {
+ plugin_output = "mysql_orders_raw"
+ url = "jdbc:mysql://mysql_cdc_e2e:3306/shop"
+ username = "st_user_source"
+ password = "mysqlpw"
+ server-id = 5601-5604
+ table-names = ["shop.orders"]
+ startup.mode = "initial"
+ schema-changes.enabled = false
+ }
+}
+
+transform {
+ Metadata {
+ plugin_input = "mysql_orders_raw"
+ plugin_output = "mysql_orders_with_meta"
+ metadata_fields {
+ Database = source_database
+ Table = source_table
+ RowKind = change_type
+ }
+ }
+
+ Sql {
+ plugin_input = "mysql_orders_with_meta"
+ plugin_output = "kafka_orders"
+ query = "select id as order_id, order_no, user_id, amount, case when
status = 0 then 'CREATED' when status = 1 then 'PAID' when status = 2 then
'SHIPPED' else 'OTHER' end as status_name, source_database, source_table,
change_type, CONCAT(source_database, '.', source_table) as source_name,
'mysql_cdc' as sync_source from dual where id is not null"
+ }
+}
+
+sink {
+ Kafka {
+ plugin_input = "kafka_orders"
+ bootstrap.servers = "kafkaCluster:9092"
+ topic = "recipe_mysql_orders"
+ format = json
+ partition_key_fields = ["order_id"]
+ kafka_headers_fields = ["source_database", "source_table", "change_type"]
+ }
+}
