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 d85a7f382f [Docs] Add JDBC-to-JDBC migration recipe (#11490)
d85a7f382f is described below
commit d85a7f382f4eb0f9333443acdd2957da3eca1201
Author: Daniel <[email protected]>
AuthorDate: Sun Jul 19 12:40:04 2026 +0800
[Docs] Add JDBC-to-JDBC migration recipe (#11490)
Co-authored-by: DanielLeens <[email protected]>
---
docs/en/getting-started/recipes/jdbc-to-jdbc.md | 172 +++++++++++++
docs/en/getting-started/recipes/overview.md | 3 +-
docs/zh/getting-started/recipes/jdbc-to-jdbc.md | 172 +++++++++++++
docs/zh/getting-started/recipes/overview.md | 3 +-
.../seatunnel/jdbc/JdbcToJdbcRecipeIT.java | 282 +++++++++++++++++++++
.../resources/jdbc_to_jdbc_with_transform.conf | 59 +++++
6 files changed, 689 insertions(+), 2 deletions(-)
diff --git a/docs/en/getting-started/recipes/jdbc-to-jdbc.md
b/docs/en/getting-started/recipes/jdbc-to-jdbc.md
new file mode 100644
index 0000000000..d62984c8a0
--- /dev/null
+++ b/docs/en/getting-started/recipes/jdbc-to-jdbc.md
@@ -0,0 +1,172 @@
+---
+sidebar_position: 3
+title: JDBC to JDBC
+---
+
+# JDBC to JDBC
+
+Use this recipe when you need a batch migration between relational databases
and want to filter or reshape rows before writing them. The example reads
orders from MySQL, keeps only paid orders, normalizes customer names, fixes the
amount scale, fills a source-system field, and writes the result to PostgreSQL.
+
+## Prerequisites
+
+1. Finish [Run your first job](../locally/run-your-first-job.md).
+
+2. Install the JDBC connector. Follow [Deployment > Download The Connector
Plugins](../locally/deployment.md#download-the-connector-plugins), then keep
the plugin below in `config/plugin_config`:
+
+```plugin_config
+--seatunnel-connectors--
+connector-jdbc
+--end--
+```
+
+```bash
+cd "${SEATUNNEL_HOME}"
+sh bin/install-plugin.sh
+ls connectors | rg 'connector-jdbc'
+```
+
+The `Sql` transform is included in the SeaTunnel distribution and does not
require another connector entry.
+
+3. Put both database drivers into `${SEATUNNEL_HOME}/lib`, then confirm that
SeaTunnel can see them:
+
+```bash
+ls "${SEATUNNEL_HOME}/lib" | rg 'mysql-connector|postgresql'
+```
+
+4. Create the MySQL source table and insert three deterministic rows:
+
+```sql
+CREATE DATABASE IF NOT EXISTS source_db;
+
+CREATE TABLE IF NOT EXISTS source_db.orders (
+ id BIGINT PRIMARY KEY,
+ customer_name VARCHAR(100) NOT NULL,
+ amount DECIMAL(16, 4) NOT NULL,
+ status VARCHAR(20) NOT NULL
+);
+
+TRUNCATE TABLE source_db.orders;
+
+INSERT INTO source_db.orders (id, customer_name, amount, status) VALUES
+ (1001, 'alice chen', 120.5000, 'PAID'),
+ (1002, 'bob li', 80.0000, 'CREATED'),
+ (1003, 'carol wu', 42.0000, 'PAID');
+```
+
+5. Create the PostgreSQL target database and table. The sink user needs
`INSERT` permission on this table.
+
+```sql
+CREATE USER test WITH PASSWORD 'test';
+CREATE DATABASE target_db OWNER test;
+```
+
+Reconnect to `target_db` as `test`, then run:
+
+```sql
+CREATE TABLE IF NOT EXISTS public.paid_orders (
+ id BIGINT PRIMARY KEY,
+ customer_name VARCHAR(100) NOT NULL,
+ amount DECIMAL(12, 2) NOT NULL,
+ source_system VARCHAR(20) NOT NULL
+);
+
+TRUNCATE TABLE public.paid_orders;
+```
+
+If the user or database already exists, reuse it instead of running the
corresponding `CREATE` statement.
+
+## Complete configuration
+
+The `plugin_output` and `plugin_input` values connect the three stages. The
transformed column order also matches the four placeholders in the sink
`query`; keep those two orders aligned when adapting the example.
+
+```hocon
+env {
+ parallelism = 1
+ job.mode = "BATCH"
+}
+
+source {
+ Jdbc {
+ plugin_output = "mysql_orders"
+ driver = "com.mysql.cj.jdbc.Driver"
+ url =
"jdbc:mysql://mysql.example.com:3306/source_db?useSSL=false&allowPublicKeyRetrieval=true"
+ username = "root"
+ password = "password"
+ query = "SELECT id, customer_name, amount, status FROM orders"
+ }
+}
+
+transform {
+ Sql {
+ plugin_input = "mysql_orders"
+ plugin_output = "paid_orders"
+ query = """
+ SELECT
+ id,
+ UPPER(customer_name) AS customer_name,
+ CAST(amount AS DECIMAL(12, 2)) AS amount,
+ 'MYSQL' AS source_system
+ FROM dual
+ WHERE status = 'PAID'
+ """
+ }
+}
+
+sink {
+ Jdbc {
+ plugin_input = "paid_orders"
+ driver = "org.postgresql.Driver"
+ url = "jdbc:postgresql://postgresql.example.com:5432/target_db"
+ username = "test"
+ password = "test"
+ query = "INSERT INTO public.paid_orders (id, customer_name, amount,
source_system) VALUES (?, ?, ?, ?)"
+ }
+}
+```
+
+`dual` is the virtual input table used by the default SeaTunnel SQL transform
engine. It does not refer to a MySQL or PostgreSQL table.
+
+## Run the job
+
+Save the configuration as `config/jdbc-to-jdbc.conf`, replace the hostnames
and credentials with values for your environment, and run SeaTunnel in local
mode:
+
+```bash
+cd "${SEATUNNEL_HOME}"
+./bin/seatunnel.sh --config ./config/jdbc-to-jdbc.conf -m local
+```
+
+## Verify the result
+
+Query the PostgreSQL target after the job finishes:
+
+```sql
+SELECT id, customer_name, amount, source_system
+FROM public.paid_orders
+ORDER BY id;
+```
+
+Expected result:
+
+| id | customer_name | amount | source_system |
+| --- | --- | ---: | --- |
+| 1001 | ALICE CHEN | 120.50 | MYSQL |
+| 1003 | CAROL WU | 42.00 | MYSQL |
+
+This result verifies each transformation independently: order `1002` was
filtered out because it was not paid, names were uppercased, amounts have scale
2, and the constant `source_system` field was added.
+
+## Common pitfalls
+
+- The MySQL or PostgreSQL driver is missing from `${SEATUNNEL_HOME}/lib`, or
its version is incompatible with the database.
+- An example hostname is copied without being replaced by an address that the
SeaTunnel process can resolve.
+- The sample MySQL URL disables TLS and allows public-key retrieval for a
simple development setup. Configure TLS according to your organization's
security requirements before using the job in production.
+- The selected field order changes, but the columns in the sink `INSERT`
statement do not change with it.
+- The target table is not empty and a second run conflicts with its primary
key. Truncate the table for a repeatable tutorial run, or choose an upsert
strategy for production.
+- A source `DECIMAL` value exceeds the precision or scale declared by
`DECIMAL(12, 2)`. Choose a target type that fits the real data before migrating
it.
+- The source changes while a batch job is running. Use a CDC source instead
when changes must be captured continuously.
+
+## Related docs
+
+- [JDBC source](../../connectors/source/Jdbc.md)
+- [SQL transform](../../transforms/sql.md)
+- [JDBC sink](../../connectors/sink/Jdbc.md)
+- [Multi-Table CDC](./multi-table-cdc.md)
diff --git a/docs/en/getting-started/recipes/overview.md
b/docs/en/getting-started/recipes/overview.md
index 053d24c262..7a16e1763c 100644
--- a/docs/en/getting-started/recipes/overview.md
+++ b/docs/en/getting-started/recipes/overview.md
@@ -6,7 +6,7 @@ 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.
+The recipes in this section include concrete prerequisites, complete
configurations, and expected results so that you can validate each pipeline in
your own environment.
## Choose A Recipe By Pipeline Goal
@@ -14,6 +14,7 @@ The MySQL CDC to Kafka recipe in this section was
cross-checked against a Docker
| --- | --- |
| 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) |
+| Batch migration between relational databases with row transformation | [JDBC
to JDBC](./jdbc-to-jdbc.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/zh/getting-started/recipes/jdbc-to-jdbc.md
b/docs/zh/getting-started/recipes/jdbc-to-jdbc.md
new file mode 100644
index 0000000000..1b9fd11874
--- /dev/null
+++ b/docs/zh/getting-started/recipes/jdbc-to-jdbc.md
@@ -0,0 +1,172 @@
+---
+sidebar_position: 3
+title: JDBC 到 JDBC
+---
+
+# JDBC 到 JDBC
+
+当你需要在关系型数据库之间进行批量迁移，并希望在写入前过滤或调整数据时，可以使用本教程。示例从 MySQL
读取订单，只保留已支付订单，统一客户姓名格式，调整金额精度，填充来源系统字段，最后写入 PostgreSQL。
+
+## 前置条件
+
+1. 先完成[运行第一个本地任务](../locally/run-your-first-job.md)。
+
+2. 安装 JDBC connector。按照[部署 > 下载连接器插件](../locally/deployment.md#下载连接器插件)操作，并在
`config/plugin_config` 中保留以下插件：
+
+```plugin_config
+--seatunnel-connectors--
+connector-jdbc
+--end--
+```
+
+```bash
+cd "${SEATUNNEL_HOME}"
+sh bin/install-plugin.sh
+ls connectors | rg 'connector-jdbc'
+```
+
+`Sql` transform 已包含在 SeaTunnel 发布包中，不需要再添加 connector 配置。
+
+3. 把两个数据库的驱动都放入 `${SEATUNNEL_HOME}/lib`，然后确认 SeaTunnel 可以看到它们：
+
+```bash
+ls "${SEATUNNEL_HOME}/lib" | rg 'mysql-connector|postgresql'
+```
+
+4. 创建 MySQL 源表，并插入三条固定数据：
+
+```sql
+CREATE DATABASE IF NOT EXISTS source_db;
+
+CREATE TABLE IF NOT EXISTS source_db.orders (
+ id BIGINT PRIMARY KEY,
+ customer_name VARCHAR(100) NOT NULL,
+ amount DECIMAL(16, 4) NOT NULL,
+ status VARCHAR(20) NOT NULL
+);
+
+TRUNCATE TABLE source_db.orders;
+
+INSERT INTO source_db.orders (id, customer_name, amount, status) VALUES
+ (1001, 'alice chen', 120.5000, 'PAID'),
+ (1002, 'bob li', 80.0000, 'CREATED'),
+ (1003, 'carol wu', 42.0000, 'PAID');
+```
+
+5. 创建 PostgreSQL 目标库和目标表。Sink 用户需要具有这张表的 `INSERT` 权限。
+
+```sql
+CREATE USER test WITH PASSWORD 'test';
+CREATE DATABASE target_db OWNER test;
+```
+
+使用 `test` 用户重新连接 `target_db`，然后执行：
+
+```sql
+CREATE TABLE IF NOT EXISTS public.paid_orders (
+ id BIGINT PRIMARY KEY,
+ customer_name VARCHAR(100) NOT NULL,
+ amount DECIMAL(12, 2) NOT NULL,
+ source_system VARCHAR(20) NOT NULL
+);
+
+TRUNCATE TABLE public.paid_orders;
+```
+
+如果用户或数据库已经存在，请直接复用，不要重复执行对应的 `CREATE` 语句。
+
+## 完整配置
+
+三个阶段通过 `plugin_output` 和 `plugin_input` 串联。转换后的字段顺序还必须与 sink `query`
中的四个占位符顺序一致；修改示例时需要同步调整这两处顺序。
+
+```hocon
+env {
+ parallelism = 1
+ job.mode = "BATCH"
+}
+
+source {
+ Jdbc {
+ plugin_output = "mysql_orders"
+ driver = "com.mysql.cj.jdbc.Driver"
+ url =
"jdbc:mysql://mysql.example.com:3306/source_db?useSSL=false&allowPublicKeyRetrieval=true"
+ username = "root"
+ password = "password"
+ query = "SELECT id, customer_name, amount, status FROM orders"
+ }
+}
+
+transform {
+ Sql {
+ plugin_input = "mysql_orders"
+ plugin_output = "paid_orders"
+ query = """
+ SELECT
+ id,
+ UPPER(customer_name) AS customer_name,
+ CAST(amount AS DECIMAL(12, 2)) AS amount,
+ 'MYSQL' AS source_system
+ FROM dual
+ WHERE status = 'PAID'
+ """
+ }
+}
+
+sink {
+ Jdbc {
+ plugin_input = "paid_orders"
+ driver = "org.postgresql.Driver"
+ url = "jdbc:postgresql://postgresql.example.com:5432/target_db"
+ username = "test"
+ password = "test"
+ query = "INSERT INTO public.paid_orders (id, customer_name, amount,
source_system) VALUES (?, ?, ?, ?)"
+ }
+}
+```
+
+`dual` 是默认 SeaTunnel SQL transform 引擎使用的虚拟输入表，不是 MySQL 或 PostgreSQL 中的真实表。
+
+## 运行任务
+
+把配置保存为 `config/jdbc-to-jdbc.conf`，按实际环境替换主机名和凭据，然后使用 local 模式运行 SeaTunnel：
+
+```bash
+cd "${SEATUNNEL_HOME}"
+./bin/seatunnel.sh --config ./config/jdbc-to-jdbc.conf -m local
+```
+
+## 验证结果
+
+任务结束后查询 PostgreSQL 目标表：
+
+```sql
+SELECT id, customer_name, amount, source_system
+FROM public.paid_orders
+ORDER BY id;
+```
+
+预期结果：
+
+| id | customer_name | amount | source_system |
+| --- | --- | ---: | --- |
+| 1001 | ALICE CHEN | 120.50 | MYSQL |
+| 1003 | CAROL WU | 42.00 | MYSQL |
+
+这组结果分别验证了每一步转换：订单 `1002` 因未支付而被过滤，姓名已转为大写，金额保留两位小数，并新增了固定值为 `MYSQL` 的
`source_system` 字段。
+
+## 常见问题
+
+- `${SEATUNNEL_HOME}/lib` 中缺少 MySQL 或 PostgreSQL 驱动，或者驱动版本与数据库不兼容。
+- 原样复制示例主机名，却没有替换为 SeaTunnel 进程能够解析的实际地址。
+- 示例 MySQL URL 为简化开发环境而关闭了 TLS，并允许公钥检索。用于生产环境前，应按照所在组织的安全要求配置 TLS。
+- 调整了 `SELECT` 字段顺序，却没有同步调整 sink `INSERT` 语句中的列顺序。
+- 目标表中已有数据，第二次运行时与主键冲突。教程重复执行前可以清空目标表；生产环境应根据业务选择 upsert 策略。
+- 源端 `DECIMAL` 数据超出 `DECIMAL(12, 2)` 的精度或小数位范围。正式迁移前需要按真实数据选择合适的目标类型。
+- 批处理任务运行期间源表仍在变化。如果需要持续捕获变更，应改用 CDC source。
+
+## 相关文档
+
+- [JDBC source](../../connectors/source/Jdbc.md)
+- [SQL transform](../../transforms/sql.md)
+- [JDBC sink](../../connectors/sink/Jdbc.md)
+- [多表 CDC](./multi-table-cdc.md)
diff --git a/docs/zh/getting-started/recipes/overview.md
b/docs/zh/getting-started/recipes/overview.md
index 991fd9de94..5e25b0397d 100644
--- a/docs/zh/getting-started/recipes/overview.md
+++ b/docs/zh/getting-started/recipes/overview.md
@@ -6,7 +6,7 @@ slug: /getting-started/recipes
这些示例更适合在你已经跑通第一个本地任务之后再阅读。不要按顺序把所有示例都看一遍，而是优先找到最接近你真实 source 和 sink 组合的那条链路。
-本节里的 “MySQL CDC 到 Kafka” 这篇教程，在 2026 年 7 月 16 日对照过一条真实的 Docker E2E
链路。页面里的配置、观测结果和前置条件，都是先和那次验证结果核对一致后才更新的。
+本节教程提供了明确的前置条件、完整配置和预期结果，便于你在自己的环境中逐项验证整条链路。
## 按业务目标选择示例
@@ -14,6 +14,7 @@ slug: /getting-started/recipes
| --- | --- |
| 从 MySQL CDC 同步到分析型数据库 | [MySQL CDC 到 Doris](./mysql-cdc-to-doris.md) |
| 从 MySQL CDC 同步到 Kafka，并附带元数据消息头 | [MySQL CDC 到
Kafka](./mysql-cdc-to-kafka.md) |
+| 在关系型数据库之间批量迁移，并进行数据过滤和转换 | [JDBC 到 JDBC](./jdbc-to-jdbc.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-jdbc-e2e/connector-jdbc-e2e-part-1/src/test/java/org/apache/seatunnel/connectors/seatunnel/jdbc/JdbcToJdbcRecipeIT.java
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-jdbc-e2e/connector-jdbc-e2e-part-1/src/test/java/org/apache/seatunnel/connectors/seatunnel/jdbc/JdbcToJdbcRecipeIT.java
new file mode 100644
index 0000000000..ecfc558c58
--- /dev/null
+++
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-jdbc-e2e/connector-jdbc-e2e-part-1/src/test/java/org/apache/seatunnel/connectors/seatunnel/jdbc/JdbcToJdbcRecipeIT.java
@@ -0,0 +1,282 @@
+/*
+ * 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.connectors.seatunnel.jdbc;
+
+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.TestContainer;
+import org.apache.seatunnel.e2e.common.junit.TestContainerExtension;
+
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.TestTemplate;
+import org.testcontainers.containers.Container;
+import org.testcontainers.containers.MySQLContainer;
+import org.testcontainers.containers.PostgreSQLContainer;
+import org.testcontainers.containers.output.Slf4jLogConsumer;
+import org.testcontainers.lifecycle.Startables;
+import org.testcontainers.utility.DockerImageName;
+import org.testcontainers.utility.DockerLoggerFactory;
+
+import lombok.extern.slf4j.Slf4j;
+
+import java.math.BigDecimal;
+import java.sql.Connection;
+import java.sql.DriverManager;
+import java.sql.ResultSet;
+import java.sql.SQLException;
+import java.sql.Statement;
+import java.util.concurrent.TimeUnit;
+import java.util.stream.Stream;
+
+import static org.awaitility.Awaitility.given;
+
+/**
+ * Verifies the JDBC-to-JDBC scenario recipe with real MySQL and PostgreSQL
containers.
+ *
+ * <p>The job filters unpaid orders, uppercases customer names, narrows the
amount scale, adds a
+ * constant source-system field, and writes the transformed rows to PostgreSQL.
+ */
+@Slf4j
+public class JdbcToJdbcRecipeIT extends TestSuiteBase implements TestResource {
+
+ // MySQL image used by the documented source side of the migration.
+ private static final String MYSQL_IMAGE = "mysql:8.0";
+
+ // PostgreSQL image used by the documented target side of the migration.
+ private static final String POSTGRES_IMAGE = "postgres:14-alpine";
+
+ // Stable Docker network alias referenced by the E2E job configuration.
+ private static final String MYSQL_HOST = "mysql-jdbc-recipe";
+
+ // Stable PostgreSQL network alias referenced by the E2E job configuration.
+ private static final String POSTGRES_HOST = "postgresql-jdbc-recipe";
+
+ // MySQL database containing the source orders table.
+ private static final String MYSQL_DATABASE = "source_db";
+
+ // PostgreSQL database containing the migrated paid orders.
+ private static final String POSTGRES_DATABASE = "target_db";
+
+ // Shared test user provisioned independently in both database containers.
+ private static final String DATABASE_USER = "test";
+
+ // Password used only by the isolated database containers in this test.
+ private static final String DATABASE_PASSWORD = "test";
+
+ // MySQL JDBC driver downloaded into each SeaTunnel engine container.
+ private static final String MYSQL_DRIVER_URL =
+
"https://repo1.maven.org/maven2/com/mysql/mysql-connector-j/8.0.32/mysql-connector-j-8.0.32.jar";;
+
+ // PostgreSQL JDBC driver downloaded into each SeaTunnel engine container.
+ private static final String POSTGRES_DRIVER_URL =
+
"https://repo1.maven.org/maven2/org/postgresql/postgresql/42.3.3/postgresql-42.3.3.jar";;
+
+ // MySQL source container shared by all engine invocations in this test
class.
+ private MySQLContainer<?> mysqlContainer;
+
+ // PostgreSQL target container shared by all engine invocations in this
test class.
+ private PostgreSQLContainer<?> postgreSQLContainer;
+
+ // Installs both JDBC drivers in every SeaTunnel engine container before
job submission.
+ @TestContainerExtension
+ private final ContainerExtendedFactory extendedFactory =
+ container -> {
+ Container.ExecResult result =
+ container.execInContainer(
+ "bash",
+ "-c",
+ "mkdir -p /tmp/seatunnel/plugins/Jdbc/lib"
+ + " && cd
/tmp/seatunnel/plugins/Jdbc/lib"
+ + " && curl -fsSLO "
+ + MYSQL_DRIVER_URL
+ + " && curl -fsSLO "
+ + POSTGRES_DRIVER_URL);
+ Assertions.assertEquals(
+ 0,
+ result.getExitCode(),
+ "Failed to download JDBC drivers: " +
result.getStderr());
+ };
+
+ /**
+ * Starts both databases and creates deterministic source and target
tables.
+ *
+ * @throws Exception if a container, JDBC driver, or table initialization
cannot be prepared
+ */
+ @BeforeAll
+ @Override
+ public void startUp() throws Exception {
+ mysqlContainer =
+ new MySQLContainer<>(DockerImageName.parse(MYSQL_IMAGE))
+ .withDatabaseName(MYSQL_DATABASE)
+ .withUsername(DATABASE_USER)
+ .withPassword(DATABASE_PASSWORD)
+ .withNetwork(NETWORK)
+ .withNetworkAliases(MYSQL_HOST)
+ .withLogConsumer(
+ new
Slf4jLogConsumer(DockerLoggerFactory.getLogger(MYSQL_IMAGE)));
+ postgreSQLContainer =
+ new
PostgreSQLContainer<>(DockerImageName.parse(POSTGRES_IMAGE))
+ .withDatabaseName(POSTGRES_DATABASE)
+ .withUsername(DATABASE_USER)
+ .withPassword(DATABASE_PASSWORD)
+ .withNetwork(NETWORK)
+ .withNetworkAliases(POSTGRES_HOST)
+ .withLogConsumer(
+ new Slf4jLogConsumer(
+
DockerLoggerFactory.getLogger(POSTGRES_IMAGE)));
+
+ Startables.deepStart(Stream.of(mysqlContainer,
postgreSQLContainer)).join();
+ Class.forName(mysqlContainer.getDriverClassName());
+ Class.forName(postgreSQLContainer.getDriverClassName());
+
+ given().ignoreExceptions()
+ .await()
+ .atMost(2, TimeUnit.MINUTES)
+ .untilAsserted(
+ () -> {
+ initializeMysqlSource();
+ initializePostgresTarget();
+ });
+ log.info("MySQL source and PostgreSQL target are ready for the recipe
test.");
+ }
+
+ /**
+ * Clears target rows before each engine invocation so stale data cannot
satisfy assertions.
+ *
+ * @throws SQLException if the PostgreSQL target cannot be reset
+ */
+ @BeforeEach
+ public void clearTargetTable() throws SQLException {
+ try (Connection connection = getPostgresConnection();
+ Statement statement = connection.createStatement()) {
+ statement.execute("TRUNCATE TABLE public.paid_orders");
+ }
+ }
+
+ /**
+ * Runs the documented pipeline and checks every transformed PostgreSQL
value.
+ *
+ * @param container SeaTunnel engine container supplied by the E2E
extension
+ * @throws Exception if the job cannot run or the target rows cannot be
queried
+ */
+ @TestTemplate
+ public void testJdbcToJdbcRecipe(TestContainer container) throws Exception
{
+ Container.ExecResult result =
container.executeJob("/jdbc_to_jdbc_with_transform.conf");
+ Assertions.assertEquals(0, result.getExitCode(), result.getStderr());
+
+ try (Connection connection = getPostgresConnection();
+ Statement statement = connection.createStatement();
+ ResultSet rows =
+ statement.executeQuery(
+ "SELECT id, customer_name, amount,
source_system "
+ + "FROM public.paid_orders ORDER BY
id")) {
+ assertRow(rows, 1001L, "ALICE CHEN", new BigDecimal("120.50"));
+ assertRow(rows, 1003L, "CAROL WU", new BigDecimal("42.00"));
+ Assertions.assertFalse(rows.next(), "Only the two PAID orders
should be written");
+ }
+ }
+
+ /**
+ * Creates and seeds the MySQL table used by the documented source query.
+ *
+ * @throws SQLException if the source table cannot be recreated
+ */
+ private void initializeMysqlSource() throws SQLException {
+ try (Connection connection =
+ DriverManager.getConnection(
+ mysqlContainer.getJdbcUrl(), DATABASE_USER,
DATABASE_PASSWORD);
+ Statement statement = connection.createStatement()) {
+ statement.execute("DROP TABLE IF EXISTS orders");
+ statement.execute(
+ "CREATE TABLE orders ("
+ + "id BIGINT PRIMARY KEY, "
+ + "customer_name VARCHAR(100) NOT NULL, "
+ + "amount DECIMAL(16, 4) NOT NULL, "
+ + "status VARCHAR(20) NOT NULL)");
+ statement.execute(
+ "INSERT INTO orders (id, customer_name, amount, status)
VALUES "
+ + "(1001, 'alice chen', 120.5000, 'PAID'), "
+ + "(1002, 'bob li', 80.0000, 'CREATED'), "
+ + "(1003, 'carol wu', 42.0000, 'PAID')");
+ }
+ }
+
+ /**
+ * Creates the PostgreSQL table whose contents are asserted after the job.
+ *
+ * @throws SQLException if the target table cannot be created
+ */
+ private void initializePostgresTarget() throws SQLException {
+ try (Connection connection = getPostgresConnection();
+ Statement statement = connection.createStatement()) {
+ statement.execute(
+ "CREATE TABLE IF NOT EXISTS public.paid_orders ("
+ + "id BIGINT PRIMARY KEY, "
+ + "customer_name VARCHAR(100) NOT NULL, "
+ + "amount DECIMAL(12, 2) NOT NULL, "
+ + "source_system VARCHAR(20) NOT NULL)");
+ }
+ }
+
+ /**
+ * Returns a host-side connection to the PostgreSQL test container.
+ *
+ * @return an open PostgreSQL connection that the caller must close
+ * @throws SQLException if the container database cannot be reached
+ */
+ private Connection getPostgresConnection() throws SQLException {
+ return DriverManager.getConnection(
+ postgreSQLContainer.getJdbcUrl(), DATABASE_USER,
DATABASE_PASSWORD);
+ }
+
+ /**
+ * Asserts one ordered target row, including the constant source-system
field.
+ *
+ * @param rows ordered query result positioned before the expected row
+ * @param id expected order identifier
+ * @param customerName expected normalized customer name
+ * @param amount expected amount with scale two
+ * @throws SQLException if a target value cannot be read
+ */
+ private void assertRow(ResultSet rows, long id, String customerName,
BigDecimal amount)
+ throws SQLException {
+ Assertions.assertTrue(rows.next(), "Expected target row " + id);
+ Assertions.assertEquals(id, rows.getLong("id"));
+ Assertions.assertEquals(customerName, rows.getString("customer_name"));
+ Assertions.assertEquals(amount, rows.getBigDecimal("amount"));
+ Assertions.assertEquals("MYSQL", rows.getString("source_system"));
+ }
+
+ /**
+ * Stops the source and target database containers after all engine
invocations have completed.
+ */
+ @AfterAll
+ @Override
+ public void tearDown() {
+ if (mysqlContainer != null) {
+ mysqlContainer.close();
+ }
+ if (postgreSQLContainer != null) {
+ postgreSQLContainer.close();
+ }
+ }
+}
diff --git
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-jdbc-e2e/connector-jdbc-e2e-part-1/src/test/resources/jdbc_to_jdbc_with_transform.conf
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-jdbc-e2e/connector-jdbc-e2e-part-1/src/test/resources/jdbc_to_jdbc_with_transform.conf
new file mode 100644
index 0000000000..35aa43ab8a
--- /dev/null
+++
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-jdbc-e2e/connector-jdbc-e2e-part-1/src/test/resources/jdbc_to_jdbc_with_transform.conf
@@ -0,0 +1,59 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+env {
+ parallelism = 1
+ job.mode = "BATCH"
+}
+
+source {
+ Jdbc {
+ plugin_output = "mysql_orders"
+ driver = "com.mysql.cj.jdbc.Driver"
+ url =
"jdbc:mysql://mysql-jdbc-recipe:3306/source_db?useSSL=false&allowPublicKeyRetrieval=true"
+ username = "test"
+ password = "test"
+ query = "SELECT id, customer_name, amount, status FROM orders"
+ }
+}
+
+transform {
+ Sql {
+ plugin_input = "mysql_orders"
+ plugin_output = "paid_orders"
+ query = """
+ SELECT
+ id,
+ UPPER(customer_name) AS customer_name,
+ CAST(amount AS DECIMAL(12, 2)) AS amount,
+ 'MYSQL' AS source_system
+ FROM dual
+ WHERE status = 'PAID'
+ """
+ }
+}
+
+sink {
+ Jdbc {
+ plugin_input = "paid_orders"
+ driver = "org.postgresql.Driver"
+ url = "jdbc:postgresql://postgresql-jdbc-recipe:5432/target_db"
+ username = "test"
+ password = "test"
+ query = "INSERT INTO public.paid_orders (id, customer_name, amount,
source_system) VALUES (?, ?, ?, ?)"
+ }
+}
