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 ae49ea3362 [Docs] Add PostgreSQL CDC to Iceberg scenario recipe
(#11482)
ae49ea3362 is described below
commit ae49ea3362f4ed8660bd1dc92d943c6e9cde6e5d
Author: Daniel <[email protected]>
AuthorDate: Sun Jul 19 12:41:40 2026 +0800
[Docs] Add PostgreSQL CDC to Iceberg scenario recipe (#11482)
Co-authored-by: DanielLeens <[email protected]>
---
docs/en/getting-started/recipes/overview.md | 1 +
.../recipes/postgresql-cdc-to-iceberg.md | 228 ++++++++++++++
docs/sidebars.js | 1 +
docs/zh/getting-started/recipes/overview.md | 1 +
.../recipes/postgresql-cdc-to-iceberg.md | 228 ++++++++++++++
.../connector-iceberg-e2e/pom.xml | 7 +
.../iceberg/PostgresCdcToIcebergRecipeIT.java | 332 +++++++++++++++++++++
.../iceberg/postgres_cdc_to_iceberg_recipe.conf | 62 ++++
8 files changed, 860 insertions(+)
diff --git a/docs/en/getting-started/recipes/overview.md
b/docs/en/getting-started/recipes/overview.md
index 7a16e1763c..25dc1282e3 100644
--- a/docs/en/getting-started/recipes/overview.md
+++ b/docs/en/getting-started/recipes/overview.md
@@ -17,6 +17,7 @@ The recipes in this section include concrete prerequisites,
complete configurati
| 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) |
+| CDC from PostgreSQL into a table format | [PostgreSQL CDC to
Iceberg](./postgresql-cdc-to-iceberg.md) |
| HTTP ingestion into a relational target | [HTTP to JDBC](./http-to-jdbc.md) |
| File-based loading into an analytical system | [File to
StarRocks](./file-to-starrocks.md) |
| Multi-table CDC orchestration | [Multi-Table CDC](./multi-table-cdc.md) |
diff --git a/docs/en/getting-started/recipes/postgresql-cdc-to-iceberg.md
b/docs/en/getting-started/recipes/postgresql-cdc-to-iceberg.md
new file mode 100644
index 0000000000..ce13c31f52
--- /dev/null
+++ b/docs/en/getting-started/recipes/postgresql-cdc-to-iceberg.md
@@ -0,0 +1,228 @@
+---
+title: PostgreSQL CDC to Iceberg with field shaping and upserts
+---
+
+# PostgreSQL CDC to Iceberg with field shaping and upserts
+
+This recipe captures an initial PostgreSQL snapshot and subsequent changes,
cleans and enriches each row, and maintains the current state in an Iceberg
table. The example uses PostgreSQL 14 with `pgoutput` and covers snapshot,
update, insert, and delete events.
+
+The pipeline is:
+
+- `Postgres-CDC` reads `sales.inventory.customer_orders` through PostgreSQL
logical replication.
+- `Sql` trims customer names, normalizes status values, and fills
`sync_source`.
+- `Iceberg` uses `id` as its identifier field and applies CDC events in upsert
mode.
+
+## Prerequisites
+
+1. Finish [Run your first job](../locally/run-your-first-job.md) with the Zeta
engine.
+
+2. Install the required connectors:
+
+```plugin_config
+--seatunnel-connectors--
+connector-cdc-postgres
+connector-iceberg
+--end--
+```
+
+```bash
+cd "${SEATUNNEL_HOME}"
+sh bin/install-plugin.sh
+ls connectors | grep -E 'connector-(cdc-postgres|iceberg)'
+```
+
+3. Put a compatible PostgreSQL JDBC driver in `${SEATUNNEL_HOME}/lib` for Zeta:
+
+```bash
+ls "${SEATUNNEL_HOME}/lib" | grep 'postgresql-'
+```
+
+4. Enable logical replication on the PostgreSQL primary and restart PostgreSQL
after changing `wal_level`:
+
+```conf
+wal_level = logical
+max_replication_slots = 10
+max_wal_senders = 10
+```
+
+Confirm the effective settings:
+
+```sql
+SHOW wal_level;
+SHOW max_replication_slots;
+SHOW max_wal_senders;
+```
+
+5. As a PostgreSQL administrator, create the source database and a dedicated
CDC user. Replace the database, user, and password as needed. If either object
already exists, skip its `CREATE` statement and reuse it:
+
+```sql
+CREATE DATABASE sales;
+```
+
+PostgreSQL requires `CREATE DATABASE` to run outside a transaction. After it
completes, run:
+
+```sql
+CREATE ROLE seatunnel_cdc WITH REPLICATION LOGIN PASSWORD 'change_me';
+GRANT CONNECT ON DATABASE sales TO seatunnel_cdc;
+```
+
+Add a matching rule to `pg_hba.conf`. Replace the example CIDR with the
network used by your SeaTunnel nodes:
+
+```conf
+host sales seatunnel_cdc 192.0.2.0/24 scram-sha-256
+```
+
+Logical replication connects to the actual database, so this rule names
`sales` rather than the `replication` keyword used for physical replication.
Reload the PostgreSQL configuration after saving the file:
+
+```sql
+SELECT pg_reload_conf();
+```
+
+6. Choose an empty Iceberg warehouse path writable by every SeaTunnel worker.
A local `file://` warehouse is suitable only when all workers see the same
filesystem. Use HDFS, S3, or another shared catalog storage for a distributed
cluster.
+
+## Prepare the source data
+
+Connect to the `sales` database as an administrator, create the source schema
and table in PostgreSQL 14, then grant the read-only CDC user access:
+
+```sql
+CREATE SCHEMA inventory;
+
+CREATE TABLE inventory.customer_orders (
+ id BIGINT PRIMARY KEY,
+ customer_name VARCHAR(64) NOT NULL,
+ amount NUMERIC(10, 2) NOT NULL,
+ status VARCHAR(16) NOT NULL,
+ updated_at TIMESTAMP NOT NULL
+);
+
+ALTER TABLE inventory.customer_orders REPLICA IDENTITY FULL;
+
+INSERT INTO inventory.customer_orders VALUES
+ (1001, ' Alice Zhang ', 120.50, 'pending', '2026-07-18 09:00:00'),
+ (1002, 'Bob Li', 80.00, 'paid', '2026-07-18 09:05:00');
+
+GRANT USAGE ON SCHEMA inventory TO seatunnel_cdc;
+GRANT SELECT ON TABLE inventory.customer_orders TO seatunnel_cdc;
+
+CREATE PUBLICATION seatunnel_sales_orders_pub
+FOR TABLE inventory.customer_orders;
+```
+
+The administrator creates the publication so that the CDC user can remain
read-only. The job configuration below reuses this publication instead of
trying to create one for every table in the database.
+
+After the initial snapshot reaches Iceberg, run the following changes with the
source application's write account or an administrator account, not the
read-only CDC account:
+
+```sql
+UPDATE inventory.customer_orders
+SET amount = 150.75, status = 'paid', updated_at = '2026-07-18 10:00:00'
+WHERE id = 1001;
+
+INSERT INTO inventory.customer_orders VALUES
+ (1003, ' Carol Wang ', 42.00, 'pending', '2026-07-18 10:05:00');
+
+DELETE FROM inventory.customer_orders WHERE id = 1002;
+```
+
+## Complete job configuration
+
+The following HOCON config implements the complete pipeline. Replace the
example hostname, credentials, slot name, and warehouse path for your
environment. Each concurrently running CDC job on the same PostgreSQL instance
must use a distinct `slot.name`. Keep `publication.name` aligned with the
publication created above.
+
+```hocon
+env {
+ parallelism = 1
+ job.mode = "STREAMING"
+ checkpoint.interval = 3000
+}
+
+source {
+ Postgres-CDC {
+ plugin_output = "postgres_orders_raw"
+ url = "jdbc:postgresql://postgresql.example.com:5432/sales"
+ username = "seatunnel_cdc"
+ password = "change_me"
+ database-names = ["sales"]
+ schema-names = ["inventory"]
+ table-names = ["sales.inventory.customer_orders"]
+ startup.mode = "initial"
+ decoding.plugin.name = "pgoutput"
+ slot.name = ${slot_name}
+ debezium = {
+ "publication.name" = "seatunnel_sales_orders_pub"
+ "publication.autocreate.mode" = "disabled"
+ }
+ }
+}
+
+transform {
+ Sql {
+ plugin_input = "postgres_orders_raw"
+ plugin_output = "iceberg_customer_orders"
+ query = "select id, trim(customer_name) as customer_name, amount,
upper(status) as status_name, updated_at, 'postgresql_cdc' as sync_source from
dual"
+ }
+}
+
+sink {
+ Iceberg {
+ plugin_input = "iceberg_customer_orders"
+ catalog_name = "recipe_catalog"
+ iceberg.catalog.config = {
+ "type" = "hadoop"
+ "warehouse" = ${warehouse}
+ }
+ namespace = "sales_analytics"
+ table = "customer_orders"
+ iceberg.table.primary-keys = "id"
+ iceberg.table.upsert-mode-enabled = true
+ schema_save_mode = "CREATE_SCHEMA_WHEN_NOT_EXIST"
+ data_save_mode = "APPEND_DATA"
+ }
+}
+```
+
+Save the config as `${SEATUNNEL_HOME}/config/postgresql-cdc-to-iceberg.conf`,
replace `${slot_name}` and `${warehouse}` with quoted literal values, and run:
+
+```bash
+cd "${SEATUNNEL_HOME}"
+./bin/seatunnel.sh --config ./config/postgresql-cdc-to-iceberg.conf -m local
+```
+
+For example:
+
+```hocon
+slot.name = "seatunnel_sales_orders"
+"warehouse" = "file:///tmp/seatunnel/iceberg/postgres-cdc-recipe/"
+```
+
+## Verified result
+
+After the incremental SQL is committed and the next Iceberg checkpoint
completes, `sales_analytics.customer_orders` contains exactly these rows:
+
+| id | customer_name | amount | status_name | sync_source |
+| --- | --- | ---: | --- | --- |
+| 1001 | Alice Zhang | 150.75 | PAID | postgresql_cdc |
+| 1003 | Carol Wang | 42.00 | PENDING | postgresql_cdc |
+
+Query the Iceberg table and verify the exact primary-key set and transformed
fields shown above. Row `1002` must no longer exist.
+
+## Operational checks
+
+- Monitor `pg_replication_slots`. A stopped consumer can retain WAL
indefinitely.
+- Drop a slot only after its CDC job is permanently decommissioned; never
share a slot between active jobs.
+- Keep `iceberg.table.primary-keys` aligned with a stable source key. Upsert
mode requires this option explicitly.
+- Ensure checkpoints continue to complete. Iceberg changes become visible
after a successful commit.
+- Do not use a node-local warehouse path in a multi-worker cluster unless the
path is backed by shared storage.
+
+## Troubleshooting
+
+- `wal_level` is not `logical`, or PostgreSQL was not restarted after the
setting changed.
+- The CDC account lacks `REPLICATION`, `CONNECT`, schema `USAGE`, or table
`SELECT` privileges.
+- Another job is already using the configured replication slot.
+- The source table does not use `REPLICA IDENTITY FULL` while the default
safety check is enabled.
+- The Iceberg warehouse is not writable or is not visible from every SeaTunnel
worker.
+- Upsert mode is enabled without an explicit `iceberg.table.primary-keys`
value.
+
+## Related documentation
+
+- [PostgreSQL CDC source](../../connectors/source/PostgreSQL-CDC.md)
+- [Iceberg sink](../../connectors/sink/Iceberg.md)
+- [Sql transform](../../transforms/sql.md)
diff --git a/docs/sidebars.js b/docs/sidebars.js
index 95765c4daa..9ec30ba53a 100644
--- a/docs/sidebars.js
+++ b/docs/sidebars.js
@@ -164,6 +164,7 @@ const sidebars = {
"getting-started/recipes/mysql-cdc-to-kafka",
"getting-started/recipes/jdbc-to-s3",
"getting-started/recipes/kafka-to-iceberg",
+ "getting-started/recipes/postgresql-cdc-to-iceberg",
"getting-started/recipes/http-to-jdbc",
"getting-started/recipes/file-to-starrocks",
"getting-started/recipes/multi-table-cdc"
diff --git a/docs/zh/getting-started/recipes/overview.md
b/docs/zh/getting-started/recipes/overview.md
index 5e25b0397d..ed09d7ce58 100644
--- a/docs/zh/getting-started/recipes/overview.md
+++ b/docs/zh/getting-started/recipes/overview.md
@@ -17,6 +17,7 @@ slug: /getting-started/recipes
| 在关系型数据库之间批量迁移，并进行数据过滤和转换 | [JDBC 到 JDBC](./jdbc-to-jdbc.md) |
| 把 JDBC 数据抽取到对象存储 | [JDBC 到 S3](./jdbc-to-s3.md) |
| 从 Kafka 流式写入表格式存储 | [Kafka 到 Iceberg](./kafka-to-iceberg.md) |
+| 从 PostgreSQL CDC 同步到表格式存储 | [PostgreSQL CDC 到
Iceberg](./postgresql-cdc-to-iceberg.md) |
| 把 HTTP 数据写入关系型数据库 | [HTTP 到 JDBC](./http-to-jdbc.md) |
| 把文件数据加载到分析型系统 | [文件到 StarRocks](./file-to-starrocks.md) |
| 多表 CDC 编排 | [多表 CDC](./multi-table-cdc.md) |
diff --git a/docs/zh/getting-started/recipes/postgresql-cdc-to-iceberg.md
b/docs/zh/getting-started/recipes/postgresql-cdc-to-iceberg.md
new file mode 100644
index 0000000000..538da5afec
--- /dev/null
+++ b/docs/zh/getting-started/recipes/postgresql-cdc-to-iceberg.md
@@ -0,0 +1,228 @@
+---
+title: PostgreSQL CDC 到 Iceberg：字段整形与 Upsert
+---
+
+# PostgreSQL CDC 到 Iceberg：字段整形与 Upsert
+
+这个场景会先读取 PostgreSQL 全量快照，再持续捕获增量变更，对字段进行清洗和补充，最后在 Iceberg 中维护最新状态。示例使用
PostgreSQL 14 和 `pgoutput`，覆盖快照、更新、新增和删除事件。
+
+完整链路是：
+
+- `Postgres-CDC` 通过 PostgreSQL 逻辑复制读取 `sales.inventory.customer_orders`。
+- `Sql` 清理客户名称、统一状态值，并填充 `sync_source`。
+- `Iceberg` 使用 `id` 作为标识字段，以 upsert 模式应用 CDC 事件。
+
+## 前置条件
+
+1. 先使用 Zeta 引擎完成[运行第一个任务](../locally/run-your-first-job.md)。
+
+2. 安装所需 connector：
+
+```plugin_config
+--seatunnel-connectors--
+connector-cdc-postgres
+connector-iceberg
+--end--
+```
+
+```bash
+cd "${SEATUNNEL_HOME}"
+sh bin/install-plugin.sh
+ls connectors | grep -E 'connector-(cdc-postgres|iceberg)'
+```
+
+3. Zeta 引擎需要把兼容版本的 PostgreSQL JDBC 驱动放入 `${SEATUNNEL_HOME}/lib`：
+
+```bash
+ls "${SEATUNNEL_HOME}/lib" | grep 'postgresql-'
+```
+
+4. 在 PostgreSQL 主库启用逻辑复制。修改 `wal_level` 后必须重启 PostgreSQL：
+
+```conf
+wal_level = logical
+max_replication_slots = 10
+max_wal_senders = 10
+```
+
+确认实际生效值：
+
+```sql
+SHOW wal_level;
+SHOW max_replication_slots;
+SHOW max_wal_senders;
+```
+
+5. 使用 PostgreSQL 管理员账号创建源数据库和专用 CDC 用户，并按实际环境替换数据库、用户名和密码。如果数据库或用户已经存在，请跳过对应的
`CREATE` 语句并直接复用：
+
+```sql
+CREATE DATABASE sales;
+```
+
+PostgreSQL 要求在事务外执行 `CREATE DATABASE`。创建完成后再执行：
+
+```sql
+CREATE ROLE seatunnel_cdc WITH REPLICATION LOGIN PASSWORD 'change_me';
+GRANT CONNECT ON DATABASE sales TO seatunnel_cdc;
+```
+
+在 `pg_hba.conf` 中添加匹配规则，并把示例 CIDR 替换为 SeaTunnel 节点所在的实际网段：
+
+```conf
+host sales seatunnel_cdc 192.0.2.0/24 scram-sha-256
+```
+
+逻辑复制连接的是实际数据库，因此这里填写 `sales`，而不是物理复制连接使用的 `replication` 关键字。保存文件后重新加载
PostgreSQL 配置：
+
+```sql
+SELECT pg_reload_conf();
+```
+
+6. 选择所有 SeaTunnel Worker 都能访问且可写的空 Iceberg warehouse。只有所有 Worker
共享同一文件系统时才适合使用本地 `file://` 路径；分布式集群应使用 HDFS、S3 或其他共享 catalog 存储。
+
+## 准备源数据
+
+使用管理员账号连接 `sales` 数据库，在 PostgreSQL 14 中创建源 schema 和表，然后为只读 CDC 用户授权：
+
+```sql
+CREATE SCHEMA inventory;
+
+CREATE TABLE inventory.customer_orders (
+ id BIGINT PRIMARY KEY,
+ customer_name VARCHAR(64) NOT NULL,
+ amount NUMERIC(10, 2) NOT NULL,
+ status VARCHAR(16) NOT NULL,
+ updated_at TIMESTAMP NOT NULL
+);
+
+ALTER TABLE inventory.customer_orders REPLICA IDENTITY FULL;
+
+INSERT INTO inventory.customer_orders VALUES
+ (1001, ' Alice Zhang ', 120.50, 'pending', '2026-07-18 09:00:00'),
+ (1002, 'Bob Li', 80.00, 'paid', '2026-07-18 09:05:00');
+
+GRANT USAGE ON SCHEMA inventory TO seatunnel_cdc;
+GRANT SELECT ON TABLE inventory.customer_orders TO seatunnel_cdc;
+
+CREATE PUBLICATION seatunnel_sales_orders_pub
+FOR TABLE inventory.customer_orders;
+```
+
+publication 由管理员创建，因此 CDC 用户可以保持只读权限。下面的任务配置会复用该 publication，而不是尝试为数据库中的所有表创建
publication。
+
+初始快照进入 Iceberg 后，使用源业务的写入账号或管理员账号执行以下变更，不要使用只读 CDC 账号：
+
+```sql
+UPDATE inventory.customer_orders
+SET amount = 150.75, status = 'paid', updated_at = '2026-07-18 10:00:00'
+WHERE id = 1001;
+
+INSERT INTO inventory.customer_orders VALUES
+ (1003, ' Carol Wang ', 42.00, 'pending', '2026-07-18 10:05:00');
+
+DELETE FROM inventory.customer_orders WHERE id = 1002;
+```
+
+## 完整任务配置
+
+下面的 HOCON 实现了完整链路。请按实际环境替换示例主机名、账号、slot 名称和 warehouse 路径。同一个 PostgreSQL 实例上的并发
CDC 任务必须使用不同的 `slot.name`，并确保 `publication.name` 与上面创建的 publication 一致。
+
+```hocon
+env {
+ parallelism = 1
+ job.mode = "STREAMING"
+ checkpoint.interval = 3000
+}
+
+source {
+ Postgres-CDC {
+ plugin_output = "postgres_orders_raw"
+ url = "jdbc:postgresql://postgresql.example.com:5432/sales"
+ username = "seatunnel_cdc"
+ password = "change_me"
+ database-names = ["sales"]
+ schema-names = ["inventory"]
+ table-names = ["sales.inventory.customer_orders"]
+ startup.mode = "initial"
+ decoding.plugin.name = "pgoutput"
+ slot.name = ${slot_name}
+ debezium = {
+ "publication.name" = "seatunnel_sales_orders_pub"
+ "publication.autocreate.mode" = "disabled"
+ }
+ }
+}
+
+transform {
+ Sql {
+ plugin_input = "postgres_orders_raw"
+ plugin_output = "iceberg_customer_orders"
+ query = "select id, trim(customer_name) as customer_name, amount,
upper(status) as status_name, updated_at, 'postgresql_cdc' as sync_source from
dual"
+ }
+}
+
+sink {
+ Iceberg {
+ plugin_input = "iceberg_customer_orders"
+ catalog_name = "recipe_catalog"
+ iceberg.catalog.config = {
+ "type" = "hadoop"
+ "warehouse" = ${warehouse}
+ }
+ namespace = "sales_analytics"
+ table = "customer_orders"
+ iceberg.table.primary-keys = "id"
+ iceberg.table.upsert-mode-enabled = true
+ schema_save_mode = "CREATE_SCHEMA_WHEN_NOT_EXIST"
+ data_save_mode = "APPEND_DATA"
+ }
+}
+```
+
+把配置保存为 `${SEATUNNEL_HOME}/config/postgresql-cdc-to-iceberg.conf`，将
`${slot_name}` 和 `${warehouse}` 替换为带引号的实际值，然后运行：
+
+```bash
+cd "${SEATUNNEL_HOME}"
+./bin/seatunnel.sh --config ./config/postgresql-cdc-to-iceberg.conf -m local
+```
+
+例如：
+
+```hocon
+slot.name = "seatunnel_sales_orders"
+"warehouse" = "file:///tmp/seatunnel/iceberg/postgres-cdc-recipe/"
+```
+
+## 已验证结果
+
+增量 SQL 提交且下一次 Iceberg checkpoint 成功后，`sales_analytics.customer_orders` 中只包含：
+
+| id | customer_name | amount | status_name | sync_source |
+| --- | --- | ---: | --- | --- |
+| 1001 | Alice Zhang | 150.75 | PAID | postgresql_cdc |
+| 1003 | Carol Wang | 42.00 | PENDING | postgresql_cdc |
+
+查询 Iceberg 表并核对上述主键集合和每个转换字段；被删除的 `1002` 不得继续存在。
+
+## 运行检查
+
+- 持续监控 `pg_replication_slots`，停止消费的 slot 可能无限保留 WAL。
+- 只有对应 CDC 任务永久下线后才能删除 slot；活动任务之间禁止共用 slot。
+- `iceberg.table.primary-keys` 必须对应稳定的源表主键，upsert 模式要求显式配置该参数。
+- 确认 checkpoint 持续成功，Iceberg 变更会在成功提交后可见。
+- 多 Worker 集群不要使用节点本地 warehouse，除非该路径实际由共享存储承载。
+
+## 常见问题
+
+- `wal_level` 不是 `logical`，或者修改配置后没有重启 PostgreSQL。
+- CDC 账号缺少 `REPLICATION`、`CONNECT`、schema `USAGE` 或表 `SELECT` 权限。
+- 配置的 replication slot 已被另一个任务使用。
+- 默认安全检查开启时，源表没有设置 `REPLICA IDENTITY FULL`。
+- Iceberg warehouse 不可写，或并非所有 SeaTunnel Worker 都可见。
+- 开启 upsert 模式，却没有显式设置 `iceberg.table.primary-keys`。
+
+## 相关文档
+
+- [PostgreSQL CDC Source](../../connectors/source/PostgreSQL-CDC.md)
+- [Iceberg Sink](../../connectors/sink/Iceberg.md)
+- [Sql Transform](../../transforms/sql.md)
diff --git
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-iceberg-e2e/pom.xml
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-iceberg-e2e/pom.xml
index ce344d435a..14ee1913d0 100644
--- a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-iceberg-e2e/pom.xml
+++ b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-iceberg-e2e/pom.xml
@@ -85,6 +85,13 @@
<scope>test</scope>
</dependency>
+ <dependency>
+ <groupId>org.apache.seatunnel</groupId>
+ <artifactId>connector-cdc-postgres</artifactId>
+ <version>${project.version}</version>
+ <scope>test</scope>
+ </dependency>
+
<dependency>
<groupId>org.apache.seatunnel</groupId>
<artifactId>connector-cdc-mysql</artifactId>
diff --git
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-iceberg-e2e/src/test/java/org/apache/seatunnel/e2e/connector/iceberg/PostgresCdcToIcebergRecipeIT.java
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-iceberg-e2e/src/test/java/org/apache/seatunnel/e2e/connector/iceberg/PostgresCdcToIcebergRecipeIT.java
new file mode 100644
index 0000000000..d0a05f1962
--- /dev/null
+++
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-iceberg-e2e/src/test/java/org/apache/seatunnel/e2e/connector/iceberg/PostgresCdcToIcebergRecipeIT.java
@@ -0,0 +1,332 @@
+/*
+ * 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.iceberg;
+
+import org.apache.seatunnel.api.configuration.ReadonlyConfig;
+import org.apache.seatunnel.connectors.seatunnel.iceberg.IcebergTableLoader;
+import
org.apache.seatunnel.connectors.seatunnel.iceberg.config.IcebergCommonOptions;
+import
org.apache.seatunnel.connectors.seatunnel.iceberg.config.IcebergSourceConfig;
+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.iceberg.Table;
+import org.apache.iceberg.data.IcebergGenerics;
+import org.apache.iceberg.data.Record;
+import org.apache.iceberg.io.CloseableIterable;
+
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.TestTemplate;
+import org.junit.jupiter.api.condition.DisabledOnOs;
+import org.junit.jupiter.api.condition.OS;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.testcontainers.containers.Container;
+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 java.io.IOException;
+import java.math.BigDecimal;
+import java.sql.Connection;
+import java.sql.DriverManager;
+import java.sql.Statement;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.TimeUnit;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
+
+import static
org.apache.seatunnel.connectors.seatunnel.iceberg.config.IcebergCatalogType.HADOOP;
+import static org.awaitility.Awaitility.await;
+
+/**
+ * Verifies the documented PostgreSQL CDC to Iceberg recipe against snapshot,
update, insert, and
+ * delete events.
+ */
+@DisabledOnContainer(
+ value = {},
+ type = {EngineType.SPARK, EngineType.FLINK},
+ disabledReason = "The recipe uses Zeta job status and cancellation
APIs")
+@DisabledOnOs(OS.WINDOWS)
+public class PostgresCdcToIcebergRecipeIT extends TestSuiteBase implements
TestResource {
+
+ private static final Logger LOG =
LoggerFactory.getLogger(PostgresCdcToIcebergRecipeIT.class);
+
+ private static final String POSTGRES_HOST = "postgres_iceberg_recipe";
+ private static final String POSTGRES_DATABASE = "sales";
+ private static final String POSTGRES_USER = "postgres";
+ private static final String POSTGRES_PASSWORD = "postgres";
+ private static final String CATALOG_ROOT =
"/tmp/seatunnel_mnt/iceberg/postgres-cdc-recipe/";
+ private static final String NAMESPACE = "sales_analytics";
+ private static final String TABLE = "customer_orders";
+ private static final String POSTGRES_DRIVER_URL =
+
"https://repo1.maven.org/maven2/org/postgresql/postgresql/42.5.1/postgresql-42.5.1.jar";;
+ private static final String ZSTD_URL =
+
"https://repo1.maven.org/maven2/com/github/luben/zstd-jni/1.5.5-5/zstd-jni-1.5.5-5.jar";;
+
+ private static final PostgreSQLContainer<?> POSTGRES_CONTAINER =
+ new
PostgreSQLContainer<>(DockerImageName.parse("postgres:14-alpine"))
+ .withNetwork(NETWORK)
+ .withNetworkAliases(POSTGRES_HOST)
+ .withDatabaseName(POSTGRES_DATABASE)
+ .withUsername(POSTGRES_USER)
+ .withPassword(POSTGRES_PASSWORD)
+ .withCommand(
+ "postgres",
+ "-c",
+ "wal_level=logical",
+ "-c",
+ "max_replication_slots=10",
+ "-c",
+ "max_wal_senders=10")
+ .withLogConsumer(
+ new Slf4jLogConsumer(
+
DockerLoggerFactory.getLogger("postgres-iceberg-recipe")));
+
+ /** Prepares connector dependencies and a writable catalog root in the
Zeta container. */
+ @TestContainerExtension
+ protected final ContainerExtendedFactory extendedFactory =
+ container -> {
+ assertCommandSucceeded(
+ container.execInContainer(
+ "sh",
+ "-c",
+ "mkdir -p "
+ + CATALOG_ROOT
+ + "
/tmp/seatunnel/plugins/Postgres-CDC/lib"
+ + " /tmp/seatunnel/plugins/Iceberg/lib"
+ + " && chmod -R 777 "
+ + CATALOG_ROOT));
+ assertCommandSucceeded(
+ container.execInContainer(
+ "sh",
+ "-c",
+ "cd /tmp/seatunnel/plugins/Postgres-CDC/lib &&
wget -q "
+ + POSTGRES_DRIVER_URL));
+ assertCommandSucceeded(
+ container.execInContainer(
+ "sh",
+ "-c",
+ "cd /tmp/seatunnel/plugins/Iceberg/lib && wget
-q " + ZSTD_URL));
+ };
+
+ /** Starts PostgreSQL with logical replication enabled and creates
deterministic source data. */
+ @BeforeAll
+ @Override
+ public void startUp() throws Exception {
+ Startables.deepStart(Stream.of(POSTGRES_CONTAINER)).join();
+ try (Connection connection = getConnection();
+ Statement statement = connection.createStatement()) {
+ statement.execute("CREATE SCHEMA inventory");
+ statement.execute(
+ "CREATE TABLE inventory.customer_orders ("
+ + "id BIGINT PRIMARY KEY, customer_name
VARCHAR(64) NOT NULL, "
+ + "amount NUMERIC(10, 2) NOT NULL, status
VARCHAR(16) NOT NULL, "
+ + "updated_at TIMESTAMP NOT NULL)");
+ statement.execute("ALTER TABLE inventory.customer_orders REPLICA
IDENTITY FULL");
+ statement.execute(
+ "INSERT INTO inventory.customer_orders VALUES "
+ + "(1001, ' Alice Zhang ', 120.50, 'pending',
'2026-07-18 09:00:00'),"
+ + "(1002, 'Bob Li', 80.00, 'paid', '2026-07-18
09:05:00')");
+ }
+ }
+
+ /** Stops the PostgreSQL container created by this test suite. */
+ @AfterAll
+ @Override
+ public void tearDown() {
+ POSTGRES_CONTAINER.close();
+ }
+
+ /**
+ * Runs the exact recipe config and checks both the initial snapshot and
the final CDC state in
+ * Iceberg.
+ */
+ @TestTemplate
+ public void testPostgresCdcToIcebergRecipe(TestContainer container) throws
Exception {
+ String jobId = String.valueOf(System.nanoTime());
+ String slotName = "seatunnel_iceberg_" + jobId;
+ String warehouse = CATALOG_ROOT;
+
+ CompletableFuture<Container.ExecResult> jobFuture =
+ CompletableFuture.supplyAsync(
+ () -> {
+ try {
+ return container.executeJob(
+
"/iceberg/postgres_cdc_to_iceberg_recipe.conf",
+ jobId,
+ "slot_name=" + slotName,
+ "warehouse=file://" + warehouse);
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+ });
+
+ try {
+ await().atMost(2, TimeUnit.MINUTES)
+ .untilAsserted(
+ () -> {
+
Assertions.assertFalse(jobFuture.isCompletedExceptionally());
+ Assertions.assertEquals("RUNNING",
container.getJobStatus(jobId));
+ assertRows(
+ warehouse,
+ expectedRows(
+ 1001L,
+ "Alice Zhang",
+ "120.50",
+ "PENDING",
+ 1002L,
+ "Bob Li",
+ "80.00",
+ "PAID"));
+ });
+
+ applyIncrementalChanges();
+
+ await().atMost(2, TimeUnit.MINUTES)
+ .untilAsserted(
+ () ->
+ assertRows(
+ warehouse,
+ expectedRows(
+ 1001L,
+ "Alice Zhang",
+ "150.75",
+ "PAID",
+ 1003L,
+ "Carol Wang",
+ "42.00",
+ "PENDING")));
+ } finally {
+ Container.ExecResult cancelResult = container.cancelJob(jobId);
+ Assertions.assertEquals(0, cancelResult.getExitCode(),
cancelResult.getStderr());
+ }
+ }
+
+ /** Applies one update, one insert, and one delete after the initial
snapshot is committed. */
+ private void applyIncrementalChanges() throws Exception {
+ try (Connection connection = getConnection();
+ Statement statement = connection.createStatement()) {
+ statement.execute(
+ "UPDATE inventory.customer_orders SET amount = 150.75,
status = 'paid', "
+ + "updated_at = '2026-07-18 10:00:00' WHERE id =
1001");
+ statement.execute(
+ "INSERT INTO inventory.customer_orders VALUES "
+ + "(1003, ' Carol Wang ', 42.00, 'pending',
'2026-07-18 10:05:00')");
+ statement.execute("DELETE FROM inventory.customer_orders WHERE id
= 1002");
+ }
+ }
+
+ /** Reads the Iceberg table and compares every final business field by
primary key. */
+ private void assertRows(String warehouse, Map<Long, ExpectedOrder>
expectedRows)
+ throws IOException {
+ Map<Long, Record> actualRows =
+ loadRows(warehouse).stream()
+ .collect(Collectors.toMap(row -> (Long)
row.getField("id"), row -> row));
+ Assertions.assertEquals(expectedRows.keySet(), actualRows.keySet());
+ expectedRows.forEach(
+ (id, expected) -> {
+ Record actual = actualRows.get(id);
+ Assertions.assertEquals(
+ expected.customerName,
actual.getField("customer_name"));
+ Assertions.assertEquals(expected.amount,
actual.getField("amount"));
+ Assertions.assertEquals(expected.statusName,
actual.getField("status_name"));
+ Assertions.assertEquals("postgresql_cdc",
actual.getField("sync_source"));
+ });
+ }
+
+ /** Builds the two-row expectation used before and after incremental
changes. */
+ private Map<Long, ExpectedOrder> expectedRows(
+ long firstId,
+ String firstName,
+ String firstAmount,
+ String firstStatus,
+ long secondId,
+ String secondName,
+ String secondAmount,
+ String secondStatus) {
+ Map<Long, ExpectedOrder> expectedRows = new HashMap<>();
+ expectedRows.put(
+ firstId, new ExpectedOrder(firstName, new
BigDecimal(firstAmount), firstStatus));
+ expectedRows.put(
+ secondId,
+ new ExpectedOrder(secondName, new BigDecimal(secondAmount),
secondStatus));
+ return expectedRows;
+ }
+
+ /** Loads the current snapshot from the Hadoop catalog used by the recipe
job. */
+ private List<Record> loadRows(String warehouse) throws IOException {
+ Map<String, Object> catalogProps = new HashMap<>();
+ catalogProps.put("type", HADOOP.getType());
+ catalogProps.put("warehouse", "file://" + warehouse);
+
+ Map<String, Object> config = new HashMap<>();
+ config.put(IcebergCommonOptions.KEY_CATALOG_NAME.key(),
"recipe_catalog");
+ config.put(IcebergCommonOptions.KEY_NAMESPACE.key(), NAMESPACE);
+ config.put(IcebergCommonOptions.KEY_TABLE.key(), TABLE);
+ config.put(IcebergCommonOptions.CATALOG_PROPS.key(), catalogProps);
+
+ try (IcebergTableLoader tableLoader =
+ IcebergTableLoader.create(
+ new
IcebergSourceConfig(ReadonlyConfig.fromMap(config)))) {
+ tableLoader.open();
+ Table table = tableLoader.loadTable();
+ List<Record> rows = new ArrayList<>();
+ try (CloseableIterable<Record> records =
IcebergGenerics.read(table).build()) {
+ records.forEach(rows::add);
+ }
+ return rows;
+ }
+ }
+
+ /** Opens a JDBC connection to the PostgreSQL Testcontainer. */
+ private Connection getConnection() throws Exception {
+ return DriverManager.getConnection(
+ POSTGRES_CONTAINER.getJdbcUrl(), POSTGRES_USER,
POSTGRES_PASSWORD);
+ }
+
+ /** Fails immediately when a container setup command is unsuccessful. */
+ private static void assertCommandSucceeded(Container.ExecResult result) {
+ Assertions.assertEquals(0, result.getExitCode(), result.getStderr());
+ }
+
+ /** Expected transformed values for one Iceberg row. */
+ private static class ExpectedOrder {
+ private final String customerName;
+ private final BigDecimal amount;
+ private final String statusName;
+
+ private ExpectedOrder(String customerName, BigDecimal amount, String
statusName) {
+ this.customerName = customerName;
+ this.amount = amount;
+ this.statusName = statusName;
+ }
+ }
+}
diff --git
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-iceberg-e2e/src/test/resources/iceberg/postgres_cdc_to_iceberg_recipe.conf
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-iceberg-e2e/src/test/resources/iceberg/postgres_cdc_to_iceberg_recipe.conf
new file mode 100644
index 0000000000..a4345dab7f
--- /dev/null
+++
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-iceberg-e2e/src/test/resources/iceberg/postgres_cdc_to_iceberg_recipe.conf
@@ -0,0 +1,62 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+env {
+ parallelism = 1
+ job.mode = "STREAMING"
+ checkpoint.interval = 3000
+}
+
+source {
+ Postgres-CDC {
+ plugin_output = "postgres_orders_raw"
+ url = "jdbc:postgresql://postgres_iceberg_recipe:5432/sales"
+ username = "postgres"
+ password = "postgres"
+ database-names = ["sales"]
+ schema-names = ["inventory"]
+ table-names = ["sales.inventory.customer_orders"]
+ startup.mode = "initial"
+ decoding.plugin.name = "pgoutput"
+ slot.name = ${slot_name}
+ }
+}
+
+transform {
+ Sql {
+ plugin_input = "postgres_orders_raw"
+ plugin_output = "iceberg_customer_orders"
+ query = "select id, trim(customer_name) as customer_name, amount,
upper(status) as status_name, updated_at, 'postgresql_cdc' as sync_source from
dual"
+ }
+}
+
+sink {
+ Iceberg {
+ plugin_input = "iceberg_customer_orders"
+ catalog_name = "recipe_catalog"
+ iceberg.catalog.config = {
+ "type" = "hadoop"
+ "warehouse" = ${warehouse}
+ }
+ namespace = "sales_analytics"
+ table = "customer_orders"
+ iceberg.table.primary-keys = "id"
+ iceberg.table.upsert-mode-enabled = true
+ schema_save_mode = "CREATE_SCHEMA_WHEN_NOT_EXIST"
+ data_save_mode = "APPEND_DATA"
+ }
+}
