This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] 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 9ad05cd740 [Feature][Connector-V2] Introduce PG base CDC abstraction
for Postgres and OpenGauss (#11029)
9ad05cd740 is described below
commit 9ad05cd74037751ca309b2447dd352d76ebeeea3
Author: David Zollo <[email protected]>
AuthorDate: Fri Oct 9 08:25:18 2026 +0000
[Feature][Connector-V2] Introduce PG base CDC abstraction for Postgres and
OpenGauss (#11029)
Co-authored-by: DanielLeens <[email protected]>
Co-authored-by: danielnadean
<[email protected]>
Co-authored-by: Daniel Leens <[email protected]>
Co-authored-by: Daniel <[email protected]>
Co-authored-by: davidzollo <[email protected]>
---
docs/en/connectors/source/Opengauss-CDC.md | 7 +-
docs/zh/connectors/source/Opengauss-CDC.md | 7 +-
.../cdc/opengauss/OpengaussIncrementalSource.java | 64 ++++
.../OpengaussIncrementalSourceFactory.java | 21 +-
.../cdc/opengauss/OpengaussSourceOptions.java | 63 ++++
.../OpengaussIncrementalSourceFactoryTest.java | 353 +++++++++++++++++++++
.../connector-cdc/connector-cdc-pg-base/pom.xml | 47 +++
.../pgbase/config/PgBaseSourceConfigFactory.java | 145 +++++++++
.../cdc/pgbase/source/PgBaseIncrementalSource.java | 149 +++++++++
.../pgbase/source/PgBaseIncrementalSourceTest.java | 294 +++++++++++++++++
.../connector-cdc/connector-cdc-postgres/pom.xml | 10 +
.../config/PostgresSourceConfigFactory.java | 151 +++------
.../cdc/postgres/source/PostgresDialect.java | 6 +-
.../postgres/source/PostgresIncrementalSource.java | 92 +++---
.../config/PostgresSourceConfigFactoryTest.java | 154 ++++++++-
.../source/PostgresIncrementalSourceTest.java | 181 +++++++++++
seatunnel-connectors-v2/connector-cdc/pom.xml | 1 +
.../seatunnel/cdc/postgres/PostgresCDCIT.java | 3 +-
tools/check_connector_registration.py | 1 +
19 files changed, 1572 insertions(+), 177 deletions(-)
diff --git a/docs/en/connectors/source/Opengauss-CDC.md
b/docs/en/connectors/source/Opengauss-CDC.md
index c6fb489b08..83e2b4c8d8 100644
--- a/docs/en/connectors/source/Opengauss-CDC.md
+++ b/docs/en/connectors/source/Opengauss-CDC.md
@@ -34,7 +34,7 @@ ALTER SYSTEM SET wal_level TO 'logical';
SELECT pg_reload_conf();
```
-2. Change the REPLICA policy of the specified table to FULL
+2. Change the REPLICA policy of the specified table to FULL, unless
`require-replica-identity-full` is set to `false`. Jobs reading tables without
`REPLICA IDENTITY FULL` fail at submit (and at checkpoint restore) unless one
of the two is done.
```sql
ALTER TABLE your_table_name REPLICA IDENTITY FULL;
@@ -79,7 +79,7 @@ select 'ALTER TABLE ' || schemaname || '.' || tablename || '
REPLICA IDENTITY FU
| table-names | List | Yes, if
`table-pattern` is not used | - | Tables to monitor. Use the fully
qualified `database.schema.table` format, for example:
`opengauss_cdc.inventory.orders`. Use `database.schema.table` matching to scope
capture to a specific schema inside a database.
[...]
| table-pattern | String | Yes, if `table-names`
is not used | - | Regular expression for tables to monitor. Use the
fully qualified table name in the pattern, for example:
`opengauss_cdc\\.inventory\\..*`. `table-names` and `table-pattern` are
mutually exclusive.
[...]
| table-names-config | List | No | - |
Per-table config list. Example: `[{"table": "db1.schema1.table1","primaryKeys":
["key1"],"snapshotSplitColumn": "key2"}]`. Use `primaryKeys` for tables without
a physical primary key. `snapshotSplitColumn` must be a unique key; otherwise
SeaTunnel ignores it and selects a split column internally.
[...]
-| startup.mode | Enum | No | INITIAL |
Optional startup mode for Opengauss CDC consumer, valid enumerations are
`initial`, `snapshot-only`, `committed-offset`, `earliest` and `latest`. <br/>
`initial`: Synchronize historical data at startup, and then synchronize
incremental data.<br/> `snapshot-only`: Synchronize historical data at startup
and finish as a bounded job without entering WAL streaming.<br/>
`committed-offset`: Skip snapshot data and st [...]
+| startup.mode | Enum | No | INITIAL |
Optional startup mode for Opengauss CDC consumer, valid enumerations are
`initial`, `snapshot-only`, `earliest` and `latest`. <br/> `initial`:
Synchronize historical data at startup, and then synchronize incremental
data.<br/> `snapshot-only`: Synchronize historical data at startup and finish
as a bounded job without entering WAL streaming.<br/> `earliest`: Startup from
the earliest offset possible.<br/> `late [...]
| stop.mode | Enum | No | NEVER |
Optional stop mode for Opengauss CDC consumer. The only valid enumeration is
`never`: the source keeps streaming WAL changes and never stops on its own once
it reaches the incremental phase.
[...]
| snapshot.split.size | Integer | No | 8096 |
The split size (number of rows) of table snapshot, captured tables are split
into multiple splits when read the snapshot of table.
[...]
| snapshot.fetch.size | Integer | No | 1024 |
The maximum fetch size for per poll when read table snapshot.
[...]
@@ -87,6 +87,7 @@ select 'ALTER TABLE ' || schemaname || '.' || tablename || '
REPLICA IDENTITY FU
| enable_concurrent_read | Boolean | No | true |
Whether to enable concurrent read with split during the snapshot phase. When
set to false, the source reads the table as a single split without any split
analysis, which is useful for tables without indexes.
[...]
| slot.name | String | No | seatunnel
| The Opengauss logical decoding slot name. Use a different slot name for each
CDC job that reads from the same Opengauss instance.
[...]
| decoding.plugin.name | String | No | pgoutput |
The name of the Postgres logical decoding plug-in installed on the
server,Supported values are decoderbufs, wal2json, wal2json_rds,
wal2json_streaming,wal2json_rds_streaming and pgoutput.
[...]
+| require-replica-identity-full | Boolean | No | true |
Require the table to have REPLICA IDENTITY FULL. When set to false, allows
tables with other replica identity settings, but UPDATE/DELETE events may not
contain the previous state. This should only be used for append-only tables
(e.g., outbox pattern). Default is true for backward compatibility.
|
| server-time-zone | String | No | - |
The session time zone in database server. If not set, then
ZoneId.systemDefault() is used to determine the server time zone.
[...]
| connect.timeout.ms | Long | No | 30000 |
The maximum time that the connector should wait after trying to connect to the
database server before timing out.
[...]
| connect.max-retries | Integer | No | 3 |
The max retry times that the connector should retry to build database server
connection.
[...]
@@ -96,7 +97,7 @@ select 'ALTER TABLE ' || schemaname || '.' || tablename || '
REPLICA IDENTITY FU
| sample-sharding.threshold | Integer | No | 1000 |
This configuration specifies the threshold of estimated shard count to trigger
the sample sharding strategy. When the distribution factor is outside the
bounds specified by `chunk-key.even-distribution.factor.upper-bound` and
`chunk-key.even-distribution.factor.lower-bound`, and the estimated shard count
(calculated as approximate row count / chunk size) exceeds this threshold, the
sample sharding strategy wil [...]
| inverse-sampling.rate | Integer | No | 1000 |
The inverse of the sampling rate used in the sample sharding strategy. For
example, if this value is set to 1000, it means a 1/1000 sampling rate is
applied during the sampling process. This option provides flexibility in
controlling the granularity of the sampling, thus affecting the final number of
shards. It's especially useful when dealing with very large datasets where a
lower sampling rate is preferred. [...]
| split.allow-sampling | Boolean | No | true |
Whether to allow sampling-based sharding strategy. When set to false, the
system will fall back to unevenly-sized chunk splitting (iterative query
approach) regardless of the shard count. The default value is true. |
-| exactly_once | Boolean | No | false |
Enable exactly-once semantics during the initial snapshot phase. This option is
only available when `startup.mode` is `initial`.
[...]
+| exactly_once | Boolean | No | false |
Enable exactly-once semantics during the snapshot phase. This option is only
available when `startup.mode` is `initial` or `snapshot-only`.
[...]
| format | Enum | No | DEFAULT |
Optional output format for Opengauss CDC, valid enumerations are `DEFAULT`,
`COMPATIBLE_DEBEZIUM_JSON`.
[...]
| debezium | Config | No | - |
Pass-through [Debezium's
properties](https://github.com/debezium/debezium/blob/v1.9.8.Final/documentation/modules/ROOT/pages/connectors/postgresql.adoc#connector-configuration-properties)
to Debezium Embedded Engine which is used to capture data changes from
Opengauss server.
[...]
| common-options | | no | - |
Source plugin common parameters, please refer to [Source Common
Options](../common-options/source-common-options.md) for details
[...]
diff --git a/docs/zh/connectors/source/Opengauss-CDC.md
b/docs/zh/connectors/source/Opengauss-CDC.md
index 84a59eb90b..9f334c27e7 100644
--- a/docs/zh/connectors/source/Opengauss-CDC.md
+++ b/docs/zh/connectors/source/Opengauss-CDC.md
@@ -33,7 +33,7 @@ ALTER SYSTEM SET wal_level TO 'logical';
SELECT pg_reload_conf();
```
-2. 改变指定表的REPLICA策略为FULL
+2. 改变指定表的REPLICA策略为FULL，除非 `require-replica-identity-full` 设置为 `false`。读取未设置
`REPLICA IDENTITY FULL` 的表时，任务会在提交（以及从 checkpoint 恢复）时失败，除非完成上述两项之一。
```sql
ALTER TABLE your_table_name REPLICA IDENTITY FULL;
@@ -74,7 +74,7 @@ select 'ALTER TABLE ' || schemaname || '.' || tablename || '
REPLICA IDENTITY FU
| table-names | 列表 | 二选一 | - |
需要监控的表。请使用完整的 `database.schema.table` 格式，例如：`opengauss_cdc.inventory.orders`。
|
| table-pattern | 字符串 | 二选一 | - |
需要监控的表名正则表达式。正则需要匹配完整表名，例如：`opengauss_cdc\\.inventory\\..*`。`table-names` 和
`table-pattern` 互斥。
|
| table-names-config | 列表 | 否 | - |
表级配置列表。例如：`[{"table": "db1.schema1.table1","primaryKeys":
["key1"],"snapshotSplitColumn": "key2"}]`。无物理主键表可通过 `primaryKeys`
指定唯一键。`snapshotSplitColumn` 必须是唯一键，否则 SeaTunnel 会忽略该配置并自动选择拆分列。
|
-| startup.mode | 枚举 | 否 | INITIAL |
Opengauss CDC消费者的可选启动模式，有效枚举为
`initial`、`snapshot-only`、`committed-offset`、`earliest` 和 `latest`。<br/>
`initial`: 启动时同步历史数据，然后同步增量数据。<br/> `snapshot-only`: 仅同步启动时的历史数据，然后以有界任务结束，不进入
WAL 流读取。<br/> `committed-offset`: 跳过快照数据，从配置的复制槽已提交 LSN 开始读取 WAL。该模式要求显式配置
`slot.name`，如果复制槽不存在或没有可用的已提交 LSN，则启动失败。<br/> `earliest`: 从可能的最早偏移量启动。<br/>
`latest`: 从最新偏移量启动。 |
+| startup.mode | 枚举 | 否 | INITIAL |
Opengauss CDC消费者的可选启动模式，有效枚举为 `initial`、`snapshot-only`、`earliest` 和
`latest`。<br/> `initial`: 启动时同步历史数据，然后同步增量数据。<br/> `snapshot-only`:
仅同步启动时的历史数据，然后以有界任务结束，不进入 WAL 流读取。<br/> `earliest`: 从可能的最早偏移量启动。<br/> `latest`:
从最新偏移量启动。<br/> `committed-offset` 仅 PostgreSQL 支持（它读取 OpenGauss 不提供的
`pg_replication_slots` 列），该连接器会在配置校验阶段拒绝此值。
|
| stop.mode | 枚举 | 否 | NEVER |
Opengauss CDC 消费者的可选停止模式。唯一有效的枚举值是 `never`：一旦进入增量阶段，数据源会持续读取 WAL 变更，不会自行停止。
|
| snapshot.split.size | 整型 | 否 | 8096 |
表快照的分割大小（行数），在读取表的快照时，捕获的表被分割成多个split
|
| snapshot.fetch.size | 整型 | 否 | 1024 |
读取表快照时，每次轮询的最大读取大小
|
@@ -82,6 +82,7 @@ select 'ALTER TABLE ' || schemaname || '.' || tablename || '
REPLICA IDENTITY FU
| enable_concurrent_read | 布尔 | 否 | true |
快照阶段是否启用分片并发读取。设置为 false 时，源会以单个分片读取表，不做分片分析，适用于没有索引的表
|
| slot.name | 字符串 | 否 | seatunnel |
Opengauss 逻辑解码槽名称。同一个 Opengauss 实例上如果有多个 CDC 任务，请为每个任务配置不同的 `slot.name`。
|
| decoding.plugin.name | 字符串 | 否 | pgoutput |
安装在服务器上的Postgres逻辑解码插件的名称，支持的值是decoderbufs、wal2json、wal2json_rds、wal2json_streaming、wal2json_rds_streaming和pgoutput
|
+| require-replica-identity-full | 布尔 | 否 | true |
要求表设置 REPLICA IDENTITY FULL。设置为 false 时允许其他 replica identity 的表，但 UPDATE/DELETE
事件可能不包含变更前的数据，仅建议用于只追加写入的表（例如 outbox 模式）。默认值 true 以保持向后兼容。
|
| server-time-zone | 字符串 | 否 | - |
数据库服务器中的会话时区。如果没有设置，则使用ZoneId.systemDefault()来确定服务器的时区
|
| connect.timeout.ms | Long | 否 | 30000 |
在尝试连接数据库服务器之后，连接器在超时之前应该等待的最大时间
|
| connect.max-retries | 整型 | 否 | 3 |
连接器在建立数据库服务器连接时应该重试的最大次数
|
@@ -91,7 +92,7 @@ select 'ALTER TABLE ' || schemaname || '.' || tablename || '
REPLICA IDENTITY FU
| sample-sharding.threshold | 整型 | 否 | 1000 |
此配置指定了用于触发采样分片策略的估计分片数的阈值。当分布因子超出了由`chunk-key.even-distribution.factor.upper-bound
`和`chunk-key.even-distribution.factor.lower-bound`，并且估计的分片计数(以近似的行数/块大小计算)超过此阈值，则将使用样本分片策略。这有助于更有效地处理大型数据集。默认值为1000个分片。
|
| inverse-sampling.rate | 整型 | 否 | 1000 |
采样分片策略中使用的采样率的倒数。例如，如果该值设置为1000，则意味着在采样过程中应用了1/1000的采样率。该选项提供了控制采样粒度的灵活性，从而影响最终的分片数量。当处理非常大的数据集时，它特别有用，其中首选较低的采样率。缺省值为1000。
|
| split.allow-sampling | 布尔 | 否 | true |
是否允许使用基于采样的分片策略。当设置为 false 时，无论预估分片数是否超过阈值，系统都会回退到非均匀分片方式（迭代查询方式）。默认值为 true。
|
-| exactly_once | 布尔 | 否 | false |
在初始快照阶段启用精确一次语义。仅当 `startup.mode` 为 `initial` 时可用。
|
+| exactly_once | 布尔 | 否 | false |
在快照阶段启用精确一次语义。仅当 `startup.mode` 为 `initial` 或 `snapshot-only` 时可用。
|
| format | 枚举 | 否 | DEFAULT |
Opengauss CDC可选的输出格式, 有效的枚举是`DEFAULT`, `COMPATIBLE_DEBEZIUM_JSON`.
|
| debezium | 配置 | 否 | - | 将
[Debezium的属性](https://github.com/debezium/debezium/blob/v1.9.8.Final/documentation/modules/ROOT/pages/connectors/postgresql.adoc#connector-configuration-properties)
传递到Debezium嵌入式引擎，该引擎用于捕获来自Opengauss服务的数据更改 |
| common-options | | 否 | - |
源码插件通用参数, 请参考[Source Common
Options](../common-options/source-common-options.md)获取详情
|
diff --git
a/seatunnel-connectors-v2/connector-cdc/connector-cdc-opengauss/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/opengauss/OpengaussIncrementalSource.java
b/seatunnel-connectors-v2/connector-cdc/connector-cdc-opengauss/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/opengauss/OpengaussIncrementalSource.java
new file mode 100644
index 0000000000..a190f9473f
--- /dev/null
+++
b/seatunnel-connectors-v2/connector-cdc/connector-cdc-opengauss/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/opengauss/OpengaussIncrementalSource.java
@@ -0,0 +1,64 @@
+/*
+ * 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.cdc.opengauss;
+
+import org.apache.seatunnel.api.configuration.Option;
+import org.apache.seatunnel.api.configuration.ReadonlyConfig;
+import org.apache.seatunnel.api.table.catalog.CatalogTable;
+import org.apache.seatunnel.connectors.cdc.base.option.StartupMode;
+import
org.apache.seatunnel.connectors.seatunnel.cdc.postgres.source.PostgresIncrementalSource;
+
+import java.util.List;
+
+/**
+ * OpenGauss source entry point that reuses the shared PG-base PostgreSQL
implementation while
+ * preserving the existing OpenGauss runtime behavior.
+ */
+public class OpengaussIncrementalSource<T> extends
PostgresIncrementalSource<T> {
+
+ // Same reason as PgBaseIncrementalSource: this source is Java-serialized
into the job DAG,
+ // so the UID must not drift when this class is edited later.
+ private static final long serialVersionUID = 1L;
+
+ private static final String IDENTIFIER = "Opengauss-CDC";
+
+ public OpengaussIncrementalSource(ReadonlyConfig options,
List<CatalogTable> catalogTables) {
+ super(options, catalogTables);
+ }
+
+ @Override
+ public String getPluginName() {
+ return IDENTIFIER;
+ }
+
+ /**
+ * Resolves startup mode against the OpenGauss option instead of the
PostgreSQL one, so this
+ * connector does not inherit PostgreSQL's replication-slot-specific
{@code committed-offset}
+ * mode. See {@link OpengaussSourceOptions#STARTUP_MODE}.
+ *
+ * <p>Initialization-order constraint: {@code IncrementalSource}'s
constructor calls this hook
+ * while resolving the startup config, before this class's own constructor
body has run. It is
+ * safe only because it returns a stateless static constant. It must never
read instance state;
+ * doing so would silently resolve the wrong option, the same class of bug
that once disabled
+ * {@code require-replica-identity-full} in the PostgreSQL source.
+ */
+ @Override
+ public Option<StartupMode> getStartupModeOption() {
+ return OpengaussSourceOptions.STARTUP_MODE;
+ }
+}
diff --git
a/seatunnel-connectors-v2/connector-cdc/connector-cdc-opengauss/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/opengauss/OpengaussIncrementalSourceFactory.java
b/seatunnel-connectors-v2/connector-cdc/connector-cdc-opengauss/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/opengauss/OpengaussIncrementalSourceFactory.java
index 3b4cf66cd7..85c01eb75d 100644
---
a/seatunnel-connectors-v2/connector-cdc/connector-cdc-opengauss/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/opengauss/OpengaussIncrementalSourceFactory.java
+++
b/seatunnel-connectors-v2/connector-cdc/connector-cdc-opengauss/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/opengauss/OpengaussIncrementalSourceFactory.java
@@ -69,24 +69,34 @@ public class OpengaussIncrementalSourceFactory implements
TableSourceFactory {
JdbcSourceOptions.CONNECTION_POOL_SIZE,
PostgresIncrementalSourceOptions.DECODING_PLUGIN_NAME,
PostgresIncrementalSourceOptions.SLOT_NAME,
+ // OpenGauss inherits the PostgreSQL replica-identity
enforcement, so it
+ // must also advertise the opt-out PostgreSQL already
declares.
+
PostgresIncrementalSourceOptions.REQUIRE_REPLICA_IDENTITY_FULL,
JdbcSourceOptions.CHUNK_KEY_EVEN_DISTRIBUTION_FACTOR_LOWER_BOUND,
JdbcSourceOptions.CHUNK_KEY_EVEN_DISTRIBUTION_FACTOR_UPPER_BOUND,
JdbcSourceOptions.SAMPLE_SHARDING_THRESHOLD,
JdbcSourceOptions.INVERSE_SAMPLING_RATE,
JdbcSourceOptions.SPLIT_ALLOW_SAMPLING,
JdbcSourceOptions.TABLE_NAMES_CONFIG)
- .optional(PostgresSourceOptions.STARTUP_MODE,
PostgresSourceOptions.STOP_MODE)
+ // startup.mode is OpenGauss-owned so the PostgreSQL-only
committed-offset mode
+ // cannot leak in here; stop.mode stays shared because "never"
is its only legal
+ // value. exactly_once applies to the snapshot phase, so it is
offered for both
+ // modes that run one, mirroring the PostgreSQL rule.
+ .optional(OpengaussSourceOptions.STARTUP_MODE,
PostgresSourceOptions.STOP_MODE)
.conditional(
- PostgresSourceOptions.STARTUP_MODE,
+ OpengaussSourceOptions.STARTUP_MODE,
StartupMode.INITIAL,
JdbcSourceOptions.EXACTLY_ONCE)
+ .conditional(
+ OpengaussSourceOptions.STARTUP_MODE,
+ StartupMode.SNAPSHOT_ONLY,
+ JdbcSourceOptions.EXACTLY_ONCE)
.build();
}
@Override
public Class<? extends SeaTunnelSource> getSourceClass() {
- return org.apache.seatunnel.connectors.seatunnel.cdc.postgres.source
- .PostgresIncrementalSource.class;
+ return OpengaussIncrementalSource.class;
}
@Override
@@ -110,8 +120,7 @@ public class OpengaussIncrementalSourceFactory implements
TableSourceFactory {
catalogTables, tableConfigs.get(), s ->
TablePath.of(s, true));
}
return (SeaTunnelSource<T, SplitT, StateT>)
- new
org.apache.seatunnel.connectors.seatunnel.cdc.postgres.source
- .PostgresIncrementalSource<>(context.getOptions(),
catalogTables);
+ new OpengaussIncrementalSource<>(context.getOptions(),
catalogTables);
};
}
}
diff --git
a/seatunnel-connectors-v2/connector-cdc/connector-cdc-opengauss/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/opengauss/OpengaussSourceOptions.java
b/seatunnel-connectors-v2/connector-cdc/connector-cdc-opengauss/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/opengauss/OpengaussSourceOptions.java
new file mode 100644
index 0000000000..233ad56741
--- /dev/null
+++
b/seatunnel-connectors-v2/connector-cdc/connector-cdc-opengauss/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/opengauss/OpengaussSourceOptions.java
@@ -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.
+ */
+
+package org.apache.seatunnel.connectors.seatunnel.cdc.opengauss;
+
+import org.apache.seatunnel.api.configuration.Options;
+import org.apache.seatunnel.api.configuration.SingleChoiceOption;
+import org.apache.seatunnel.connectors.cdc.base.option.SourceOptions;
+import org.apache.seatunnel.connectors.cdc.base.option.StartupMode;
+
+import java.util.Arrays;
+
+/**
+ * Startup-mode contract owned by the OpenGauss CDC connector.
+ *
+ * <p>OpenGauss reuses the PostgreSQL runtime through the PG base, but it must
not inherit
+ * PostgreSQL's startup-mode surface wholesale. {@code committed-offset} is
excluded: it resolves
+ * its start position through {@code LsnOffsetFactory#committedOffset()},
which reads {@code
+ * confirmed_flush_lsn} and {@code active_pid} from {@code
pg_replication_slots}, columns OpenGauss
+ * is not verified to expose. {@code snapshot-only} is kept: it is served
entirely by the
+ * dialect-agnostic incremental framework ({@code SnapshotOnlySplitAssigner},
gated only on {@code
+ * StartupMode} in {@code IncrementalSource}), touches nothing
PostgreSQL-specific, and was already
+ * accepted by this connector through the shared PostgreSQL option. Owning the
option here means a
+ * later PostgreSQL-only addition cannot silently widen this connector again.
+ *
+ * <p>{@code stop.mode} is deliberately still taken from the PostgreSQL
options: it has a single
+ * legal value ({@code never}) with no dialect-specific behavior behind it.
+ */
+public class OpengaussSourceOptions {
+
+ /**
+ * Startup modes OpenGauss CDC accepts: everything PostgreSQL CDC accepts
except {@code
+ * committed-offset}. No cast is needed here because the single-choice
builder already returns a
+ * {@link SingleChoiceOption} of the chosen enum type.
+ */
+ public static final SingleChoiceOption<StartupMode> STARTUP_MODE =
+ Options.key(SourceOptions.STARTUP_MODE_KEY)
+ .singleChoice(
+ StartupMode.class,
+ Arrays.asList(
+ StartupMode.INITIAL,
+ StartupMode.SNAPSHOT_ONLY,
+ StartupMode.EARLIEST,
+ StartupMode.LATEST))
+ .defaultValue(StartupMode.INITIAL)
+ .withDescription(
+ "Optional startup mode for Opengauss CDC source,
valid enumerations are "
+ + "\"initial\", \"snapshot-only\",
\"earliest\", \"latest\"");
+}
diff --git
a/seatunnel-connectors-v2/connector-cdc/connector-cdc-opengauss/src/test/java/org/apache/seatunnel/connectors/seatunnel/cdc/opengauss/OpengaussIncrementalSourceFactoryTest.java
b/seatunnel-connectors-v2/connector-cdc/connector-cdc-opengauss/src/test/java/org/apache/seatunnel/connectors/seatunnel/cdc/opengauss/OpengaussIncrementalSourceFactoryTest.java
new file mode 100644
index 0000000000..5696b31733
--- /dev/null
+++
b/seatunnel-connectors-v2/connector-cdc/connector-cdc-opengauss/src/test/java/org/apache/seatunnel/connectors/seatunnel/cdc/opengauss/OpengaussIncrementalSourceFactoryTest.java
@@ -0,0 +1,353 @@
+/*
+ * 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.cdc.opengauss;
+
+import org.apache.seatunnel.api.configuration.Option;
+import org.apache.seatunnel.api.configuration.ReadonlyConfig;
+import org.apache.seatunnel.api.configuration.SingleChoiceOption;
+import org.apache.seatunnel.api.configuration.util.Condition;
+import org.apache.seatunnel.api.configuration.util.OptionRule;
+import org.apache.seatunnel.api.configuration.util.RequiredOption;
+import org.apache.seatunnel.api.options.ConnectorCommonOptions;
+import org.apache.seatunnel.api.table.catalog.CatalogTable;
+import org.apache.seatunnel.api.table.catalog.CatalogTableUtil;
+import org.apache.seatunnel.api.table.type.BasicType;
+import org.apache.seatunnel.api.table.type.SeaTunnelDataType;
+import org.apache.seatunnel.api.table.type.SeaTunnelRowType;
+import org.apache.seatunnel.connectors.cdc.base.config.JdbcSourceConfig;
+import org.apache.seatunnel.connectors.cdc.base.config.SourceConfig;
+import org.apache.seatunnel.connectors.cdc.base.dialect.DataSourceDialect;
+import org.apache.seatunnel.connectors.cdc.base.option.JdbcSourceOptions;
+import org.apache.seatunnel.connectors.cdc.base.option.SourceOptions;
+import org.apache.seatunnel.connectors.cdc.base.option.StartupMode;
+import org.apache.seatunnel.connectors.cdc.base.source.offset.Offset;
+import org.apache.seatunnel.connectors.cdc.base.source.offset.OffsetFactory;
+import
org.apache.seatunnel.connectors.cdc.debezium.DebeziumDeserializationSchema;
+import
org.apache.seatunnel.connectors.seatunnel.cdc.postgres.config.PostgresIncrementalSourceOptions;
+import
org.apache.seatunnel.connectors.seatunnel.cdc.postgres.source.PostgresSourceOptions;
+import org.apache.seatunnel.connectors.seatunnel.jdbc.config.JdbcCommonOptions;
+
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+import io.debezium.config.Configuration;
+
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import static org.mockito.Mockito.mock;
+
+/**
+ * Tests the OpenGauss CDC source entry point introduced by the PG-base
refactor so the dedicated
+ * factory/source wiring stays compatible.
+ */
+public class OpengaussIncrementalSourceFactoryTest {
+
+ @Test
+ public void testGetSourceClassUsesDedicatedOpengaussSource() {
+ OpengaussIncrementalSourceFactory factory = new
OpengaussIncrementalSourceFactory();
+
+ Assertions.assertEquals(OpengaussIncrementalSource.class,
factory.getSourceClass());
+ }
+
+ /**
+ * OpenGauss shares the PostgreSQL runtime but owns its startup-mode
surface. committed-offset
+ * is PostgreSQL-specific (it reads confirmed_flush_lsn and active_pid from
+ * pg_replication_slots) and must not become selectable here just because
the PostgreSQL option
+ * has it. snapshot-only is served by the dialect-agnostic framework and
stays available, so an
+ * existing OpenGauss job configured with it keeps passing validation.
+ */
+ @Test
+ public void testOptionRuleExposesOnlyOpengaussStartupModes() {
+ SingleChoiceOption<StartupMode> startupMode = startupModeFromRule();
+
+ Assertions.assertEquals(
+ Arrays.asList(
+ StartupMode.INITIAL,
+ StartupMode.SNAPSHOT_ONLY,
+ StartupMode.EARLIEST,
+ StartupMode.LATEST),
+ startupMode.getOptionValues());
+ Assertions.assertFalse(
+
startupMode.getOptionValues().contains(StartupMode.COMMITTED_OFFSET));
+ Assertions.assertEquals(StartupMode.INITIAL,
startupMode.defaultValue());
+ }
+
+ /** Locates the startup.mode option the OpenGauss factory actually
advertises in its rule. */
+ private static SingleChoiceOption<StartupMode> startupModeFromRule() {
+ return (SingleChoiceOption<StartupMode>)
+ new OpengaussIncrementalSourceFactory()
+ .optionRule().getOptionalOptions().stream()
+ .filter(
+ option ->
+
SourceOptions.STARTUP_MODE_KEY.equals(option.key()))
+ .findFirst()
+ .orElseThrow(
+ () ->
+ new AssertionError(
+ "startup.mode missing
from the Opengauss option rule"));
+ }
+
+ /**
+ * exactly_once governs the snapshot phase, and both initial and
snapshot-only run one, so the
+ * OpenGauss rule must offer it under both modes exactly as the PostgreSQL
rule does. Pinning
+ * this keeps a future edit from restoring snapshot-only in the choice
list but forgetting the
+ * conditional that makes exactly_once configurable with it.
+ */
+ @Test
+ public void testOptionRuleOffersExactlyOnceForBothSnapshotModes() {
+ OptionRule rule = new OpengaussIncrementalSourceFactory().optionRule();
+
+ for (StartupMode snapshotMode :
+ Arrays.asList(StartupMode.INITIAL, StartupMode.SNAPSHOT_ONLY))
{
+ Assertions.assertTrue(
+ rule.getRequiredOptions()
+ .contains(
+
RequiredOption.ConditionalRequiredOptions.of(
+ Condition.of(
+
OpengaussSourceOptions.STARTUP_MODE,
+ snapshotMode),
+
Collections.<Option<?>>singletonList(
+
JdbcSourceOptions.EXACTLY_ONCE))),
+ "exactly_once conditional missing for startup.mode " +
snapshotMode);
+ }
+ }
+
+ /**
+ * The source must resolve startup mode through the same narrowed option
the rule advertises.
+ */
+ @Test
+ public void testSourceResolvesStartupModeFromOpengaussOption() {
+ OpengaussIncrementalSource<Object> source =
+ new TestingOpengaussIncrementalSource(
+ ReadonlyConfig.fromMap(Collections.emptyMap()),
createCatalogTables());
+
+ Assertions.assertSame(OpengaussSourceOptions.STARTUP_MODE,
source.getStartupModeOption());
+ }
+
+ /** Guards the other direction: narrowing OpenGauss must not narrow
PostgreSQL. */
+ @Test
+ public void testPostgresStartupModesRemainUnchanged() {
+ Assertions.assertEquals(
+ Arrays.asList(
+ StartupMode.INITIAL,
+ StartupMode.SNAPSHOT_ONLY,
+ StartupMode.COMMITTED_OFFSET,
+ StartupMode.EARLIEST,
+ StartupMode.LATEST),
+ PostgresSourceOptions.STARTUP_MODE.getOptionValues());
+ }
+
+ @Test
+ public void testSourceKeepsOpengaussPluginName() {
+ OpengaussIncrementalSource<Object> source =
+ new TestingOpengaussIncrementalSource(
+ ReadonlyConfig.fromMap(Collections.emptyMap()),
createCatalogTables());
+
+ Assertions.assertEquals("Opengauss-CDC", source.getPluginName());
+ }
+
+ /**
+ * OpenGauss reuses the PostgreSQL config factory through the PG base.
This pins the Debezium
+ * properties that assembly produces, which is what the refactor actually
moved.
+ */
+ @Test
+ public void testSourceKeepsOpengaussDebeziumPropertyAssembly() {
+ JdbcSourceConfig sourceConfig =
+ new TestingOpengaussConfigSource(opengaussConfig(),
createCatalogTables())
+ .buildSourceConfig();
+ Configuration dbzConfiguration = sourceConfig.getDbzConfiguration();
+
+ Assertions.assertEquals(
+ "io.debezium.connector.postgresql.PostgresConnector",
+ dbzConfiguration.getString("connector.class"));
+ Assertions.assertEquals(
+ "postgres_cdc_source",
dbzConfiguration.getString("database.server.name"));
+ Assertions.assertEquals("opengauss-host",
dbzConfiguration.getString("database.hostname"));
+ Assertions.assertEquals("5432",
dbzConfiguration.getString("database.port"));
+ Assertions.assertEquals("inventory",
dbzConfiguration.getString("database.dbname"));
+ Assertions.assertEquals("pgoutput",
dbzConfiguration.getString("plugin.name"));
+ Assertions.assertEquals("opengauss_slot",
dbzConfiguration.getString("slot.name"));
+ Assertions.assertEquals("public",
dbzConfiguration.getString("schema.include.list"));
+ // db.schema.table must still collapse to schema.table for the
PG-compatible connector.
+ Assertions.assertEquals("public.orders",
dbzConfiguration.getString("table.include.list"));
+ Assertions.assertEquals("false",
dbzConfiguration.getString("include.schema.changes"));
+ Assertions.assertEquals("org.postgresql.Driver",
sourceConfig.getDriverClassName());
+ }
+
+ /** Config mirroring a minimal OpenGauss CDC job so the real config
factory can run. */
+ private static ReadonlyConfig opengaussConfig() {
+ Map<String, Object> options = new HashMap<>();
+ options.put(JdbcCommonOptions.URL.key(),
"jdbc:postgresql://opengauss-host:5432/inventory");
+ options.put(JdbcSourceOptions.USERNAME.key(), "og_user");
+ options.put(JdbcSourceOptions.PASSWORD.key(), "og_pwd");
+ options.put(JdbcSourceOptions.DATABASE_NAMES.key(),
Collections.singletonList("inventory"));
+ options.put(
+ ConnectorCommonOptions.TABLE_NAMES.key(),
+ Collections.singletonList("inventory.public.orders"));
+ options.put(
+ PostgresIncrementalSourceOptions.SCHEMA_NAME.key(),
+ Collections.singletonList("public"));
+ options.put(PostgresIncrementalSourceOptions.SLOT_NAME.key(),
"opengauss_slot");
+ return ReadonlyConfig.fromMap(options);
+ }
+
+ /** Builds a minimal catalog table list so the source constructor can
initialize metadata. */
+ private List<CatalogTable> createCatalogTables() {
+ SeaTunnelRowType rowType =
+ new SeaTunnelRowType(
+ new String[] {"id"}, new SeaTunnelDataType[]
{BasicType.INT_TYPE});
+ return Collections.singletonList(
+ CatalogTableUtil.getCatalogTable(
+ "opengauss", "test_db", "public", "orders", rowType));
+ }
+
+ /**
+ * Lightweight test source that bypasses the real JDBC bootstrap so the
regression test can
+ * focus on the entry-path contract only.
+ */
+ private static final class TestingOpengaussIncrementalSource
+ extends OpengaussIncrementalSource<Object> {
+
+ private TestingOpengaussIncrementalSource(
+ ReadonlyConfig options, List<CatalogTable> catalogTables) {
+ super(options, catalogTables);
+ }
+
+ /**
+ * Returns a mock config factory because this regression test does not
need real JDBC setup.
+ */
+ @Override
+ public SourceConfig.Factory<JdbcSourceConfig>
createSourceConfigFactory(
+ ReadonlyConfig config) {
+ return subtask -> mock(JdbcSourceConfig.class);
+ }
+
+ /**
+ * Returns a mock dialect because the constructor should not hit an
external database here.
+ */
+ @Override
+ public DataSourceDialect<JdbcSourceConfig>
createDataSourceDialect(ReadonlyConfig config) {
+ return mockDataSourceDialect();
+ }
+
+ /** Returns a mock deserializer because schema loading is outside this
regression scope. */
+ @Override
+ public DebeziumDeserializationSchema<Object>
createDebeziumDeserializationSchema(
+ ReadonlyConfig config) {
+ return mockDebeziumDeserializationSchema();
+ }
+
+ /** Returns a no-op offset factory so the constructor can finish
without connector state. */
+ @Override
+ public OffsetFactory createOffsetFactory(ReadonlyConfig config) {
+ return new TestingOffsetFactory();
+ }
+
+ /** Creates a typed dialect mock for the minimal constructor-only
regression scenario. */
+ @SuppressWarnings("unchecked")
+ private static DataSourceDialect<JdbcSourceConfig>
mockDataSourceDialect() {
+ return (DataSourceDialect<JdbcSourceConfig>)
mock(DataSourceDialect.class);
+ }
+
+ /**
+ * Creates a typed deserializer mock for the minimal constructor-only
regression scenario.
+ */
+ @SuppressWarnings("unchecked")
+ private static DebeziumDeserializationSchema<Object>
mockDebeziumDeserializationSchema() {
+ return (DebeziumDeserializationSchema<Object>)
+ mock(DebeziumDeserializationSchema.class);
+ }
+ }
+
+ /**
+ * Test source that keeps the real inherited config factory so the
Debezium property assembly is
+ * exercised, and mocks only the parts the superclass constructor builds
afterwards.
+ */
+ private static final class TestingOpengaussConfigSource
+ extends OpengaussIncrementalSource<Object> {
+
+ private TestingOpengaussConfigSource(
+ ReadonlyConfig options, List<CatalogTable> catalogTables) {
+ super(options, catalogTables);
+ }
+
+ /**
+ * Returns a mock dialect because the constructor should not hit an
external database here.
+ */
+ @Override
+ public DataSourceDialect<JdbcSourceConfig>
createDataSourceDialect(ReadonlyConfig config) {
+ return TestingOpengaussIncrementalSource.mockDataSourceDialect();
+ }
+
+ /** Returns a mock deserializer because schema loading is outside this
regression scope. */
+ @Override
+ public DebeziumDeserializationSchema<Object>
createDebeziumDeserializationSchema(
+ ReadonlyConfig config) {
+ return
TestingOpengaussIncrementalSource.mockDebeziumDeserializationSchema();
+ }
+
+ /** Returns a no-op offset factory so the constructor can finish
without connector state. */
+ @Override
+ public OffsetFactory createOffsetFactory(ReadonlyConfig config) {
+ return new TestingOffsetFactory();
+ }
+
+ /** Builds the source config the inherited PostgreSQL factory
assembles for OpenGauss. */
+ private JdbcSourceConfig buildSourceConfig() {
+ return configFactory.create(0);
+ }
+ }
+
+ /** Minimal offset factory used to satisfy the source constructor in this
regression test. */
+ private static final class TestingOffsetFactory extends OffsetFactory {
+
+ @Override
+ public Offset earliest() {
+ return null;
+ }
+
+ @Override
+ public Offset neverStop() {
+ return null;
+ }
+
+ @Override
+ public Offset latest() {
+ return null;
+ }
+
+ @Override
+ public Offset specific(Map<String, String> offset) {
+ return null;
+ }
+
+ @Override
+ public Offset specific(String filename, Long position) {
+ return null;
+ }
+
+ @Override
+ public Offset timestamp(long timestamp) {
+ return null;
+ }
+ }
+}
diff --git
a/seatunnel-connectors-v2/connector-cdc/connector-cdc-pg-base/pom.xml
b/seatunnel-connectors-v2/connector-cdc/connector-cdc-pg-base/pom.xml
new file mode 100644
index 0000000000..d6caccf2a1
--- /dev/null
+++ b/seatunnel-connectors-v2/connector-cdc/connector-cdc-pg-base/pom.xml
@@ -0,0 +1,47 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+
+ Licensed to the Apache Software Foundation (ASF) under one or more
+ contributor license agreements. See the NOTICE file distributed with
+ this work for additional information regarding copyright ownership.
+ The ASF licenses this file to You under the Apache License, Version 2.0
+ (the "License"); you may not use this file except in compliance with
+ the License. You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+ Unless required by applicable law or agreed to in writing, software
+ distributed under the License is distributed on an "AS IS" BASIS,
+ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ See the License for the specific language governing permissions and
+ limitations under the License.
+
+-->
+<project xmlns="http://maven.apache.org/POM/4.0.0";
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance";
+ xsi:schemaLocation="http://maven.apache.org/POM/4.0.0
http://maven.apache.org/xsd/maven-4.0.0.xsd";>
+ <modelVersion>4.0.0</modelVersion>
+ <parent>
+ <groupId>org.apache.seatunnel</groupId>
+ <artifactId>connector-cdc</artifactId>
+ <version>${revision}</version>
+ </parent>
+
+ <artifactId>connector-cdc-pg-base</artifactId>
+ <name>SeaTunnel : Connectors V2 : CDC : PG Base</name>
+
+ <dependencies>
+ <!--
+ Must stay 'provided', like every other connector-cdc-* module.
connector-cdc-base is
+ shipped as its own connectors/connector-cdc-base jar; a compile
scope here would make
+ the shade plugin bundle it (and the Debezium embedded runtime)
into this jar, and then
+ into connector-cdc-postgres and connector-cdc-opengauss, giving
the plugin classloader
+ a second copy of
IncrementalSource/SourceSplitBase/EmbeddedDatabaseHistory.
+ -->
+ <dependency>
+ <groupId>org.apache.seatunnel</groupId>
+ <artifactId>connector-cdc-base</artifactId>
+ <version>${project.version}</version>
+ <scope>provided</scope>
+ </dependency>
+ </dependencies>
+</project>
diff --git
a/seatunnel-connectors-v2/connector-cdc/connector-cdc-pg-base/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/pgbase/config/PgBaseSourceConfigFactory.java
b/seatunnel-connectors-v2/connector-cdc/connector-cdc-pg-base/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/pgbase/config/PgBaseSourceConfigFactory.java
new file mode 100644
index 0000000000..76ce7f8eb5
--- /dev/null
+++
b/seatunnel-connectors-v2/connector-cdc/connector-cdc-pg-base/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/pgbase/config/PgBaseSourceConfigFactory.java
@@ -0,0 +1,145 @@
+/*
+ * 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.cdc.pgbase.config;
+
+import org.apache.seatunnel.connectors.cdc.base.config.JdbcSourceConfig;
+import org.apache.seatunnel.connectors.cdc.base.config.JdbcSourceConfigFactory;
+import org.apache.seatunnel.connectors.cdc.base.option.StartupMode;
+import org.apache.seatunnel.connectors.cdc.debezium.EmbeddedDatabaseHistory;
+
+import java.util.Properties;
+import java.util.UUID;
+import java.util.stream.Collectors;
+
+import static
org.apache.seatunnel.shade.com.google.common.base.Preconditions.checkArgument;
+import static
org.apache.seatunnel.shade.com.google.common.base.Preconditions.checkNotNull;
+
+/**
+ * Base JDBC source config factory for PostgreSQL-compatible CDC connectors.
+ *
+ * <p>This factory centralizes the common Debezium property assembly shared by
PG-base connectors
+ * while leaving connector-specific identifiers and extra properties to
subclasses.
+ */
+public abstract class PgBaseSourceConfigFactory<C extends JdbcSourceConfig>
+ extends JdbcSourceConfigFactory {
+
+ // Pinned because this class sits in the serialization hierarchy of the
concrete factories,
+ // which are shipped inside the job DAG. A computed UID would drift on
every edit here and
+ // break rolling upgrades (jobs submitted on the prior version fail to
deserialize).
+ private static final long serialVersionUID = 1L;
+
+ @Override
+ public C create(int subtask) {
+ Properties props = new Properties();
+ props.setProperty("connector.class", connectorClassName());
+ props.setProperty("database.server.name", databaseServerName());
+ props.setProperty("database.hostname", checkNotNull(hostname));
+ props.setProperty("database.user", checkNotNull(username));
+ props.setProperty("database.password", checkNotNull(password));
+ props.setProperty("database.port", String.valueOf(port));
+ // Validate before indexing: databaseList.get(0) would otherwise throw
a bare NPE or
+ // IndexOutOfBoundsException that gives no hint the 'database-names'
option is missing.
+ checkNotNull(databaseList, "The 'database-names' option is required.");
+ checkArgument(!databaseList.isEmpty(), "The 'database-names' option
must not be empty.");
+ props.setProperty("database.dbname",
checkNotNull(databaseList.get(0)));
+ // Deliberately do NOT set "database.include.list": Debezium folds it
into
+ // dataCollectionFilter() as a predicate on TableId#catalog, but
PostgreSQL table ids are
+ // catalog-less (see PostgresSchema#readTableSchema and the event
dispatchers), so every
+ // snapshot schema read and streaming event would be filtered out.
Database scoping for
+ // discovery is applied in TableDiscoveryUtils#listTables via an
explicit predicate.
+
+ // Keep the current in-memory history wiring unchanged to avoid
restore drift in phase 1.
+ props.setProperty("database.history",
EmbeddedDatabaseHistory.class.getCanonicalName());
+ props.setProperty("database.history.instance.name", UUID.randomUUID()
+ "_" + subtask);
+ props.setProperty("database.history.skip.unparseable.ddl",
String.valueOf(true));
+ props.setProperty("database.history.refer.ddl", String.valueOf(true));
+
+ props.setProperty("database.tcpKeepAlive", String.valueOf(true));
+
+ configureConnectorProperties(props, subtask);
+
+ if (tableList != null) {
+ props.setProperty(
+ "table.include.list",
+ tableList.stream()
+ .map(this::formatTableIdentifier)
+ .collect(Collectors.joining(",")));
+ }
+
+ if (dbzProperties != null) {
+ props.putAll(dbzProperties);
+ }
+ // Debezium PostgreSQL-compatible connectors do not emit DDL records,
but SeaTunnel uses
+ // this flag to enable synthetic schema records produced from pgoutput
RELATION messages.
+ // Apply it after the Debezium pass-through properties so the
SeaTunnel option remains
+ // authoritative even if a user's raw debezium.* passthrough also sets
this key.
+ props.setProperty("include.schema.changes",
String.valueOf(schemaChangeEnabled));
+
+ if (startupConfig != null && startupConfig.getStartupMode() ==
StartupMode.SNAPSHOT_ONLY) {
+ props.setProperty("snapshot.mode", "initial_only");
+ } else if (startupConfig != null
+ && startupConfig.getStartupMode() ==
StartupMode.COMMITTED_OFFSET) {
+ props.setProperty("snapshot.mode", "never");
+ }
+
+ C config = createSourceConfig(props, driverClassName());
+ // Keep the concurrent-read flag wired after moving concrete
connectors onto PG-base.
+ config.setEnableConcurrentRead(this.enableConcurrentRead);
+ return config;
+ }
+
+ /** Returns the Debezium connector class name used by the concrete PG-base
connector. */
+ protected abstract String connectorClassName();
+
+ /** Returns the logical server name used as Debezium topic namespace. */
+ protected abstract String databaseServerName();
+
+ /** Returns the JDBC driver class used by the concrete PG-base connector.
*/
+ protected abstract String driverClassName();
+
+ /**
+ * Allows subclasses to inject connector-specific Debezium properties such
as slot or plugin
+ * settings before user-supplied Debezium overrides are merged.
+ */
+ protected void configureConnectorProperties(Properties props, int subtask)
{}
+
+ /**
+ * Normalizes table identifiers to the schema.table form expected by
PostgreSQL-compatible
+ * Debezium connectors.
+ *
+ * <p>The rejection message keeps the pre-refactor wording verbatim: both
Postgres-CDC and
+ * Opengauss-CDC already surfaced it to users, and the internal module
name would mean nothing
+ * to someone reading the failure.
+ */
+ protected String formatTableIdentifier(String tableIdentifier) {
+ String[] splits = tableIdentifier.split("\\.");
+ if (splits.length == 2) {
+ return tableIdentifier;
+ }
+ if (splits.length == 3) {
+ return String.join(".", splits[1], splits[2]);
+ }
+ throw new IllegalArgumentException(
+ "Invalid table name: "
+ + tableIdentifier
+ + " ,Postgres identifier is of the form
schemaName.tableName");
+ }
+
+ /** Creates the concrete JDBC source config after the common Debezium
properties are built. */
+ protected abstract C createSourceConfig(Properties props, String
driverClassName);
+}
diff --git
a/seatunnel-connectors-v2/connector-cdc/connector-cdc-pg-base/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/pgbase/source/PgBaseIncrementalSource.java
b/seatunnel-connectors-v2/connector-cdc/connector-cdc-pg-base/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/pgbase/source/PgBaseIncrementalSource.java
new file mode 100644
index 0000000000..322973fd08
--- /dev/null
+++
b/seatunnel-connectors-v2/connector-cdc/connector-cdc-pg-base/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/pgbase/source/PgBaseIncrementalSource.java
@@ -0,0 +1,149 @@
+/*
+ * 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.cdc.pgbase.source;
+
+import org.apache.seatunnel.api.configuration.ReadonlyConfig;
+import org.apache.seatunnel.api.source.SupportParallelism;
+import org.apache.seatunnel.api.table.catalog.CatalogTable;
+import org.apache.seatunnel.common.utils.SeaTunnelException;
+import org.apache.seatunnel.connectors.cdc.base.config.JdbcSourceConfig;
+import org.apache.seatunnel.connectors.cdc.base.dialect.JdbcDataSourceDialect;
+import org.apache.seatunnel.connectors.cdc.base.option.JdbcSourceOptions;
+import org.apache.seatunnel.connectors.cdc.base.source.IncrementalSource;
+import
org.apache.seatunnel.connectors.cdc.debezium.DebeziumDeserializationConverterFactory;
+import
org.apache.seatunnel.connectors.cdc.debezium.DebeziumDeserializationSchema;
+import org.apache.seatunnel.connectors.cdc.debezium.DeserializeFormat;
+import
org.apache.seatunnel.connectors.cdc.debezium.row.DebeziumJsonDeserializeSchema;
+import
org.apache.seatunnel.connectors.cdc.debezium.row.SeaTunnelRowDebeziumDeserializeSchema;
+
+import org.apache.kafka.connect.data.Struct;
+
+import io.debezium.jdbc.JdbcConnection;
+import io.debezium.relational.TableId;
+import io.debezium.relational.history.ConnectTableChangeSerializer;
+import io.debezium.relational.history.TableChanges;
+import io.debezium.util.SchemaNameAdjuster;
+
+import java.time.ZoneId;
+import java.util.List;
+import java.util.Map;
+import java.util.function.Function;
+import java.util.stream.Collectors;
+
+/**
+ * Shared incremental source logic for PostgreSQL-compatible CDC connectors.
+ *
+ * <p>This phase intentionally keeps the dialect, fetch-task, and offset
behavior in the concrete
+ * connector modules while deduplicating the common deserialization and
table-change discovery
+ * logic.
+ */
+public abstract class PgBaseIncrementalSource<T, C extends JdbcSourceConfig>
+ extends IncrementalSource<T, C> implements SupportParallelism {
+
+ // The SeaTunnelSource is Java-serialized into the job DAG, and this class
sits in that
+ // hierarchy. Pinning the UID keeps a later edit here from breaking source
deserialization
+ // between mixed-build client/server submissions.
+ private static final long serialVersionUID = 1L;
+
+ protected PgBaseIncrementalSource(ReadonlyConfig options,
List<CatalogTable> catalogTables) {
+ super(options, catalogTables);
+ }
+
+ @SuppressWarnings("unchecked")
+ @Override
+ public DebeziumDeserializationSchema<T>
createDebeziumDeserializationSchema(
+ ReadonlyConfig config) {
+ Map<TableId, Struct> tableIdTableChangeMap = loadTableChanges();
+ if (DeserializeFormat.COMPATIBLE_DEBEZIUM_JSON.equals(
+ config.get(JdbcSourceOptions.FORMAT))) {
+ return (DebeziumDeserializationSchema<T>)
+ new DebeziumJsonDeserializeSchema(
+ config.get(JdbcSourceOptions.DEBEZIUM_PROPERTIES),
+ tableIdTableChangeMap);
+ }
+
+ return (DebeziumDeserializationSchema<T>)
+ SeaTunnelRowDebeziumDeserializeSchema.builder()
+ .setTables(catalogTables)
+ .setServerTimeZone(
+
ZoneId.of(config.get(JdbcSourceOptions.SERVER_TIME_ZONE)))
+ .setTableIdTableChangeMap(tableIdTableChangeMap)
+
.setUserDefinedConverterFactory(getUserDefinedConverterFactory())
+ .build();
+ }
+
+ /**
+ * Returns the Debezium converter factory used by the concrete PG-base
connector.
+ *
+ * <p>Defaults to the same factory {@link
SeaTunnelRowDebeziumDeserializeSchema.Builder} already
+ * applies, so a dialect only overrides this when it genuinely needs
custom conversion.
+ */
+ protected DebeziumDeserializationConverterFactory
getUserDefinedConverterFactory() {
+ return DebeziumDeserializationConverterFactory.DEFAULT;
+ }
+
+ /**
+ * Discovers the captured tables and serializes their schemas into the
Debezium table-change
+ * payload used by the row deserializer.
+ *
+ * <p>Reuses the dialect built by the constructor instead of instantiating
a second one, so the
+ * discovery here shares the connector's existing connection
configuration. Table discovery must
+ * finish before the schema reader opens because discovery owns its JDBC
connection; this keeps
+ * initialization at the historical peak of one active connection.
+ */
+ protected Map<TableId, Struct> loadTableChanges() {
+ C sourceConfig = configFactory.create(0);
+ JdbcDataSourceDialect jdbcDataSourceDialect = (JdbcDataSourceDialect)
dataSourceDialect;
+ ConnectTableChangeSerializer serializer =
+ new ConnectTableChangeSerializer(SchemaNameAdjuster.create());
+ try {
+ // Discovery owns its JDBC connection, so finish it before opening
the schema reader.
+ List<TableId> tableIds =
jdbcDataSourceDialect.discoverDataCollections(sourceConfig);
+ try (JdbcConnection jdbcConnection =
+ jdbcDataSourceDialect.openJdbcConnection(sourceConfig)) {
+ return tableIds.stream()
+ .collect(
+ Collectors.toMap(
+ Function.identity(),
+ tableId ->
+ serializeTableChange(
+ jdbcDataSourceDialect,
+ jdbcConnection,
+ tableId,
+ serializer)));
+ }
+ } catch (Exception e) {
+ // Shared by every PG-base connector now, so name the phase: a
bare wrapped
+ // exception here is indistinguishable between discovery, schema
query and
+ // serialization failures.
+ throw new SeaTunnelException(
+ "Failed to load table changes for CDC schema
initialization", e);
+ }
+ }
+
+ private Struct serializeTableChange(
+ JdbcDataSourceDialect jdbcDataSourceDialect,
+ JdbcConnection jdbcConnection,
+ TableId tableId,
+ ConnectTableChangeSerializer serializer) {
+ TableChanges tableChanges = new TableChanges();
+ tableChanges.create(
+ jdbcDataSourceDialect.queryTableSchema(jdbcConnection,
tableId).getTable());
+ return serializer.serialize(tableChanges).get(0);
+ }
+}
diff --git
a/seatunnel-connectors-v2/connector-cdc/connector-cdc-pg-base/src/test/java/org/apache/seatunnel/connectors/seatunnel/cdc/pgbase/source/PgBaseIncrementalSourceTest.java
b/seatunnel-connectors-v2/connector-cdc/connector-cdc-pg-base/src/test/java/org/apache/seatunnel/connectors/seatunnel/cdc/pgbase/source/PgBaseIncrementalSourceTest.java
new file mode 100644
index 0000000000..239c88d1d9
--- /dev/null
+++
b/seatunnel-connectors-v2/connector-cdc/connector-cdc-pg-base/src/test/java/org/apache/seatunnel/connectors/seatunnel/cdc/pgbase/source/PgBaseIncrementalSourceTest.java
@@ -0,0 +1,294 @@
+/*
+ * 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.cdc.pgbase.source;
+
+import org.apache.seatunnel.api.configuration.Option;
+import org.apache.seatunnel.api.configuration.Options;
+import org.apache.seatunnel.api.configuration.ReadonlyConfig;
+import org.apache.seatunnel.api.table.catalog.CatalogTable;
+import org.apache.seatunnel.api.table.catalog.CatalogTableUtil;
+import org.apache.seatunnel.api.table.type.BasicType;
+import org.apache.seatunnel.api.table.type.SeaTunnelDataType;
+import org.apache.seatunnel.api.table.type.SeaTunnelRow;
+import org.apache.seatunnel.api.table.type.SeaTunnelRowType;
+import org.apache.seatunnel.connectors.cdc.base.config.JdbcSourceConfig;
+import org.apache.seatunnel.connectors.cdc.base.config.SourceConfig;
+import org.apache.seatunnel.connectors.cdc.base.dialect.DataSourceDialect;
+import org.apache.seatunnel.connectors.cdc.base.dialect.JdbcDataSourceDialect;
+import org.apache.seatunnel.connectors.cdc.base.option.SourceOptions;
+import org.apache.seatunnel.connectors.cdc.base.option.StartupMode;
+import org.apache.seatunnel.connectors.cdc.base.option.StopMode;
+import org.apache.seatunnel.connectors.cdc.base.source.offset.Offset;
+import org.apache.seatunnel.connectors.cdc.base.source.offset.OffsetFactory;
+import
org.apache.seatunnel.connectors.cdc.debezium.DebeziumDeserializationConverterFactory;
+import org.apache.seatunnel.connectors.cdc.debezium.DeserializeFormat;
+import
org.apache.seatunnel.connectors.cdc.debezium.row.DebeziumJsonDeserializeSchema;
+import
org.apache.seatunnel.connectors.cdc.debezium.row.SeaTunnelRowDebeziumDeserializeSchema;
+
+import org.apache.kafka.connect.data.Struct;
+
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+import org.mockito.InOrder;
+
+import io.debezium.jdbc.JdbcConnection;
+import io.debezium.relational.Column;
+import io.debezium.relational.Table;
+import io.debezium.relational.TableId;
+import io.debezium.relational.history.TableChanges;
+
+import java.sql.Types;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.clearInvocations;
+import static org.mockito.Mockito.inOrder;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.verifyNoMoreInteractions;
+import static org.mockito.Mockito.when;
+
+/**
+ * Covers the source behavior the PG-base extraction moved out of the concrete
PostgreSQL connector:
+ * table-change discovery through the already-built dialect, and the
format-dependent deserializer
+ * selection. A regression here silently changes the schema payload every
PG-compatible CDC job
+ * hands to its row deserializer.
+ */
+public class PgBaseIncrementalSourceTest {
+
+ private static final TableId ORDERS = new TableId(null, "public",
"orders");
+
+ private static final TableId CUSTOMERS = new TableId(null, "public",
"customers");
+
+ @Test
+ public void testLoadTableChangesSerializesEveryDiscoveredTable() {
+ TestingPgBaseIncrementalSource source = new
TestingPgBaseIncrementalSource(defaultConfig());
+
+ Map<TableId, Struct> tableChanges = source.loadTableChanges();
+
+ Assertions.assertEquals(2, tableChanges.size());
+
Assertions.assertTrue(tableChanges.keySet().containsAll(Arrays.asList(ORDERS,
CUSTOMERS)));
+ tableChanges.values().forEach(Assertions::assertNotNull);
+ }
+
+ @Test
+ public void testLoadTableChangesReusesTheDialectBuiltByTheConstructor() {
+ TestingPgBaseIncrementalSource source = new
TestingPgBaseIncrementalSource(defaultConfig());
+ JdbcDataSourceDialect constructorDialect = source.dialect();
+
+ clearInvocations(constructorDialect);
+ source.loadTableChanges();
+
+ InOrder connectionOrder = inOrder(constructorDialect);
+
connectionOrder.verify(constructorDialect).discoverDataCollections(any());
+ connectionOrder.verify(constructorDialect).openJdbcConnection(any());
+ connectionOrder.verify(constructorDialect,
times(2)).queryTableSchema(any(), any());
+ verifyNoMoreInteractions(constructorDialect);
+ }
+
+ @Test
+ public void testLoadTableChangesClosesJdbcConnection() throws Exception {
+ TestingPgBaseIncrementalSource source = new
TestingPgBaseIncrementalSource(defaultConfig());
+
+ source.loadTableChanges();
+
+ verify(source.connection(), times(2)).close();
+ }
+
+ @Test
+ public void
testCreateDebeziumDeserializationSchemaUsesRowSchemaByDefault() {
+ ReadonlyConfig config = defaultConfig();
+ TestingPgBaseIncrementalSource source = new
TestingPgBaseIncrementalSource(config);
+
+ Assertions.assertInstanceOf(
+ SeaTunnelRowDebeziumDeserializeSchema.class,
+ source.createDebeziumDeserializationSchema(config));
+ }
+
+ @Test
+ public void
testCreateDebeziumDeserializationSchemaHonorsCompatibleDebeziumJsonFormat() {
+ Map<String, Object> options = new HashMap<>();
+ options.put(SourceOptions.FORMAT.key(),
DeserializeFormat.COMPATIBLE_DEBEZIUM_JSON.name());
+ options.put(SourceOptions.DEBEZIUM_PROPERTIES.key(), new
HashMap<String, String>());
+ ReadonlyConfig config = ReadonlyConfig.fromMap(options);
+
+ TestingPgBaseIncrementalSource source = new
TestingPgBaseIncrementalSource(config);
+
+ Assertions.assertInstanceOf(
+ DebeziumJsonDeserializeSchema.class,
+ source.createDebeziumDeserializationSchema(config));
+ }
+
+ @Test
+ public void testDefaultConverterFactoryIsTheDebeziumDefault() {
+ TestingPgBaseIncrementalSource source = new
TestingPgBaseIncrementalSource(defaultConfig());
+
+ Assertions.assertSame(
+ DebeziumDeserializationConverterFactory.DEFAULT,
+ source.getUserDefinedConverterFactory());
+ }
+
+ private static ReadonlyConfig defaultConfig() {
+ return ReadonlyConfig.fromMap(new HashMap<>());
+ }
+
+ private static List<CatalogTable> catalogTables() {
+ SeaTunnelRowType rowType =
+ new SeaTunnelRowType(
+ new String[] {"id"}, new SeaTunnelDataType[]
{BasicType.INT_TYPE});
+ return Collections.singletonList(
+ CatalogTableUtil.getCatalogTable("pg", "test_db", "public",
"orders", rowType));
+ }
+
+ /** Builds a minimal real Debezium table so the change serializer produces
a usable Struct. */
+ private static TableChanges.TableChange tableChange(TableId tableId) {
+ Table table =
+ Table.editor()
+ .tableId(tableId)
+ .addColumn(
+ Column.editor()
+ .name("id")
+ .type("int4")
+ .jdbcType(Types.INTEGER)
+ .optional(false)
+ .create())
+ .setPrimaryKeyNames("id")
+ .create();
+ return new TableChanges().create(table).iterator().next();
+ }
+
+ /**
+ * Test source that swaps the JDBC bootstrap for mocks while leaving the
PG-base logic under
+ * test untouched. The mocks are created inside the overridden factory
methods because the
+ * superclass constructor invokes them before any subclass field
initializer runs.
+ */
+ private static final class TestingPgBaseIncrementalSource
+ extends PgBaseIncrementalSource<SeaTunnelRow, JdbcSourceConfig> {
+
+ private static final Option<StartupMode> STARTUP_MODE =
+ Options.key(SourceOptions.STARTUP_MODE_KEY)
+ .enumType(StartupMode.class)
+ .defaultValue(StartupMode.INITIAL)
+ .withDescription("Test-only startup mode option.");
+
+ private static final Option<StopMode> STOP_MODE =
+ Options.key(SourceOptions.STOP_MODE_KEY)
+ .enumType(StopMode.class)
+ .defaultValue(StopMode.NEVER)
+ .withDescription("Test-only stop mode option.");
+
+ private JdbcConnection jdbcConnection;
+
+ private TestingPgBaseIncrementalSource(ReadonlyConfig options) {
+ super(options, catalogTables());
+ }
+
+ @Override
+ public String getPluginName() {
+ return "PgBase-Test";
+ }
+
+ @Override
+ public Option<StartupMode> getStartupModeOption() {
+ return STARTUP_MODE;
+ }
+
+ @Override
+ public Option<StopMode> getStopModeOption() {
+ return STOP_MODE;
+ }
+
+ @Override
+ public SourceConfig.Factory<JdbcSourceConfig>
createSourceConfigFactory(
+ ReadonlyConfig config) {
+ return subtask -> mock(JdbcSourceConfig.class);
+ }
+
+ @Override
+ public DataSourceDialect<JdbcSourceConfig>
createDataSourceDialect(ReadonlyConfig config) {
+ jdbcConnection = mock(JdbcConnection.class);
+ JdbcDataSourceDialect dialect = mock(JdbcDataSourceDialect.class);
+ when(dialect.openJdbcConnection(any())).thenReturn(jdbcConnection);
+ when(dialect.discoverDataCollections(any()))
+ .thenReturn(Arrays.asList(ORDERS, CUSTOMERS));
+ when(dialect.queryTableSchema(any(), any()))
+ .thenAnswer(invocation ->
tableChange(invocation.getArgument(1)));
+ return dialect;
+ }
+
+ @Override
+ public OffsetFactory createOffsetFactory(ReadonlyConfig config) {
+ return new TestingOffsetFactory();
+ }
+
+ @Override
+ public Optional<String> driverName() {
+ return Optional.of("org.postgresql.Driver");
+ }
+
+ /** Exposes the dialect built by the superclass constructor for
verification. */
+ private JdbcDataSourceDialect dialect() {
+ return (JdbcDataSourceDialect) dataSourceDialect;
+ }
+
+ /** Exposes the connection handed out by the mocked dialect for close
verification. */
+ private JdbcConnection connection() {
+ return jdbcConnection;
+ }
+ }
+
+ /** Minimal offset factory so the source constructor can finish without
connector state. */
+ private static final class TestingOffsetFactory extends OffsetFactory {
+
+ @Override
+ public Offset earliest() {
+ return null;
+ }
+
+ @Override
+ public Offset neverStop() {
+ return null;
+ }
+
+ @Override
+ public Offset latest() {
+ return null;
+ }
+
+ @Override
+ public Offset specific(Map<String, String> offset) {
+ return null;
+ }
+
+ @Override
+ public Offset specific(String filename, Long position) {
+ return null;
+ }
+
+ @Override
+ public Offset timestamp(long timestamp) {
+ return null;
+ }
+ }
+}
diff --git
a/seatunnel-connectors-v2/connector-cdc/connector-cdc-postgres/pom.xml
b/seatunnel-connectors-v2/connector-cdc/connector-cdc-postgres/pom.xml
index 3d16cbe8f8..5e71097a2d 100644
--- a/seatunnel-connectors-v2/connector-cdc/connector-cdc-postgres/pom.xml
+++ b/seatunnel-connectors-v2/connector-cdc/connector-cdc-postgres/pom.xml
@@ -48,6 +48,12 @@
<version>${project.version}</version>
<scope>compile</scope>
</dependency>
+ <dependency>
+ <groupId>org.apache.seatunnel</groupId>
+ <artifactId>connector-cdc-pg-base</artifactId>
+ <version>${project.version}</version>
+ <scope>compile</scope>
+ </dependency>
<dependency>
<groupId>io.debezium</groupId>
@@ -65,6 +71,10 @@
<artifactId>connector-cdc-base</artifactId>
<scope>provided</scope>
</dependency>
+ <dependency>
+ <groupId>org.apache.seatunnel</groupId>
+ <artifactId>connector-cdc-pg-base</artifactId>
+ </dependency>
<dependency>
<groupId>io.debezium</groupId>
diff --git
a/seatunnel-connectors-v2/connector-cdc/connector-cdc-postgres/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/postgres/config/PostgresSourceConfigFactory.java
b/seatunnel-connectors-v2/connector-cdc/connector-cdc-postgres/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/postgres/config/PostgresSourceConfigFactory.java
index 0a03e247e1..99e273de9f 100644
---
a/seatunnel-connectors-v2/connector-cdc/connector-cdc-postgres/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/postgres/config/PostgresSourceConfigFactory.java
+++
b/seatunnel-connectors-v2/connector-cdc/connector-cdc-postgres/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/postgres/config/PostgresSourceConfigFactory.java
@@ -18,20 +18,17 @@
package org.apache.seatunnel.connectors.seatunnel.cdc.postgres.config;
import org.apache.seatunnel.api.configuration.ReadonlyConfig;
-import org.apache.seatunnel.connectors.cdc.base.config.JdbcSourceConfigFactory;
-import org.apache.seatunnel.connectors.cdc.base.option.StartupMode;
-import org.apache.seatunnel.connectors.cdc.debezium.EmbeddedDatabaseHistory;
+import
org.apache.seatunnel.connectors.seatunnel.cdc.pgbase.config.PgBaseSourceConfigFactory;
import io.debezium.connector.postgresql.PostgresConnector;
import java.util.List;
import java.util.Properties;
-import java.util.UUID;
-import java.util.stream.Collectors;
-import static
org.apache.seatunnel.shade.com.google.common.base.Preconditions.checkNotNull;
-
-public class PostgresSourceConfigFactory extends JdbcSourceConfigFactory {
+/**
+ * PostgreSQL-specific source config factory built on top of the PG-base
Debezium property assembly.
+ */
+public class PostgresSourceConfigFactory extends
PgBaseSourceConfigFactory<PostgresSourceConfig> {
private static final long serialVersionUID = 1L;
@@ -47,7 +44,7 @@ public class PostgresSourceConfigFactory extends
JdbcSourceConfigFactory {
private List<String> schemaList;
@Override
- public JdbcSourceConfigFactory fromReadonlyConfig(ReadonlyConfig config) {
+ public PostgresSourceConfigFactory fromReadonlyConfig(ReadonlyConfig
config) {
super.fromReadonlyConfig(config);
this.decodingPluginName =
config.get(PostgresIncrementalSourceOptions.DECODING_PLUGIN_NAME);
this.slotName = config.get(PostgresIncrementalSourceOptions.SLOT_NAME);
@@ -56,103 +53,55 @@ public class PostgresSourceConfigFactory extends
JdbcSourceConfigFactory {
}
@Override
- public PostgresSourceConfig create(int subtask) {
- Properties props = new Properties();
- props.setProperty("connector.class",
PostgresConnector.class.getCanonicalName());
- // hard code server name, because we don't need to distinguish it,
docs:
- // Logical name that identifies and provides a namespace for the
particular PostgreSQL
- // database server/cluster being monitored. The logical name should be
unique across
- // all other connectors, since it is used as a prefix for all Kafka
topic names coming
- // from this connector. Only alphanumeric characters and underscores
should be used.
- props.setProperty("database.server.name", DATABASE_SERVER_NAME);
- props.setProperty("database.hostname", checkNotNull(hostname));
- props.setProperty("database.user", checkNotNull(username));
- props.setProperty("database.password", checkNotNull(password));
- props.setProperty("database.port", String.valueOf(port));
- props.setProperty("database.dbname",
checkNotNull(databaseList.get(0)));
- // Deliberately do NOT set "database.include.list": Debezium folds it
into
- // dataCollectionFilter() as a predicate on TableId#catalog, but
PostgreSQL table ids are
- // catalog-less (see PostgresSchema#readTableSchema and the event
dispatchers), so every
- // snapshot schema read and streaming event would be filtered out.
Database scoping for
- // discovery is applied in TableDiscoveryUtils#listTables via an
explicit predicate.
- props.setProperty("plugin.name", decodingPluginName);
- props.setProperty("slot.name", slotName);
+ protected String connectorClassName() {
+ return PostgresConnector.class.getCanonicalName();
+ }
- // database history
- props.setProperty("database.history",
EmbeddedDatabaseHistory.class.getCanonicalName());
- props.setProperty("database.history.instance.name", UUID.randomUUID()
+ "_" + subtask);
- props.setProperty("database.history.skip.unparseable.ddl",
String.valueOf(true));
- props.setProperty("database.history.refer.ddl", String.valueOf(true));
+ @Override
+ protected String databaseServerName() {
+ return DATABASE_SERVER_NAME;
+ }
+
+ @Override
+ protected String driverClassName() {
+ return DRIVER_CLASS_NAME;
+ }
- props.setProperty("database.tcpKeepAlive", String.valueOf(true));
+ @Override
+ protected void configureConnectorProperties(Properties props, int subtask)
{
+ props.setProperty("plugin.name", decodingPluginName);
+ props.setProperty("slot.name", slotName);
if (schemaList != null) {
props.setProperty("schema.include.list", String.join(",",
schemaList));
}
+ }
- if (tableList != null) {
- // pg identifier is of the form schemaName.tableName
- props.setProperty(
- "table.include.list",
- tableList.stream()
- .map(
- tableStr -> {
- String[] splits =
tableStr.split("\\.");
- if (splits.length == 2) {
- return tableStr;
- }
- if (splits.length == 3) {
- return String.join(".", splits[1],
splits[2]);
- }
- throw new IllegalArgumentException(
- "Invalid table name: "
- + tableStr
- + " ,Postgres
identifier is of the form schemaName.tableName");
- })
- .collect(Collectors.joining(",")));
- }
-
- if (dbzProperties != null) {
- props.putAll(dbzProperties);
- }
- // Debezium PostgreSQL does not emit DDL records, but SeaTunnel uses
this flag to enable
- // synthetic schema records produced from pgoutput RELATION messages.
Apply it after the
- // Debezium pass-through properties so the SeaTunnel option remains
authoritative.
- props.setProperty("include.schema.changes",
String.valueOf(schemaChangeEnabled));
- if (startupConfig != null && startupConfig.getStartupMode() ==
StartupMode.SNAPSHOT_ONLY) {
- props.setProperty("snapshot.mode", "initial_only");
- } else if (startupConfig != null
- && startupConfig.getStartupMode() ==
StartupMode.COMMITTED_OFFSET) {
- props.setProperty("snapshot.mode", "never");
- }
-
- PostgresSourceConfig config =
- new PostgresSourceConfig(
- startupConfig,
- stopConfig,
- databaseList,
- tableList,
- splitSize,
- splitColumn,
- distributionFactorUpper,
- distributionFactorLower,
- sampleShardingThreshold,
- inverseSamplingRate,
- sampleShardingAllow,
- props,
- DRIVER_CLASS_NAME,
- hostname,
- port,
- username,
- password,
- originUrl,
- fetchSize,
- serverTimeZone,
- connectTimeoutMillis,
- connectMaxRetries,
- connectionPoolSize,
- exactlyOnce);
- // Propagate the enableConcurrentRead flag so the chunk splitter can
skip split analysis.
- config.setEnableConcurrentRead(this.enableConcurrentRead);
- return config;
+ @Override
+ protected PostgresSourceConfig createSourceConfig(Properties props, String
driverClassName) {
+ return new PostgresSourceConfig(
+ startupConfig,
+ stopConfig,
+ databaseList,
+ tableList,
+ splitSize,
+ splitColumn,
+ distributionFactorUpper,
+ distributionFactorLower,
+ sampleShardingThreshold,
+ inverseSamplingRate,
+ sampleShardingAllow,
+ props,
+ driverClassName,
+ hostname,
+ port,
+ username,
+ password,
+ originUrl,
+ fetchSize,
+ serverTimeZone,
+ connectTimeoutMillis,
+ connectMaxRetries,
+ connectionPoolSize,
+ exactlyOnce);
}
}
diff --git
a/seatunnel-connectors-v2/connector-cdc/connector-cdc-postgres/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/postgres/source/PostgresDialect.java
b/seatunnel-connectors-v2/connector-cdc/connector-cdc-postgres/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/postgres/source/PostgresDialect.java
index 09bf8e6b60..8f3e6a0872 100644
---
a/seatunnel-connectors-v2/connector-cdc/connector-cdc-postgres/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/postgres/source/PostgresDialect.java
+++
b/seatunnel-connectors-v2/connector-cdc/connector-cdc-postgres/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/postgres/source/PostgresDialect.java
@@ -138,9 +138,13 @@ public class PostgresDialect implements
JdbcDataSourceDialect {
postgresConnection.readReplicaIdentityInfo(tableId);
if (requireReplicaIdentityFull
&&
!ServerInfo.ReplicaIdentity.FULL.equals(replicaIdentity)) {
+ // Name both remediations: this also fires on enumerator
restore, where the job
+ // config is already baked into the persisted DAG and can only
be changed by
+ // cancelling and resubmitting, so the message must be
actionable on its own.
throw new SeaTunnelException(
String.format(
- "Table %s does not have a full replica
identity, please execute: ALTER TABLE %s REPLICA IDENTITY FULL;",
+ "Table %s does not have a full replica
identity, please execute: ALTER TABLE %s REPLICA IDENTITY FULL; "
+ + "or set
require-replica-identity-full = false to accept UPDATE/DELETE events without
the previous row state.",
tableId, tableId));
}
}
diff --git
a/seatunnel-connectors-v2/connector-cdc/connector-cdc-postgres/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/postgres/source/PostgresIncrementalSource.java
b/seatunnel-connectors-v2/connector-cdc/connector-cdc-postgres/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/postgres/source/PostgresIncrementalSource.java
index 8f19c01925..ae165073b4 100644
---
a/seatunnel-connectors-v2/connector-cdc/connector-cdc-postgres/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/postgres/source/PostgresIncrementalSource.java
+++
b/seatunnel-connectors-v2/connector-cdc/connector-cdc-postgres/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/postgres/source/PostgresIncrementalSource.java
@@ -19,7 +19,6 @@ package
org.apache.seatunnel.connectors.seatunnel.cdc.postgres.source;
import org.apache.seatunnel.api.configuration.Option;
import org.apache.seatunnel.api.configuration.ReadonlyConfig;
-import org.apache.seatunnel.api.source.SupportParallelism;
import org.apache.seatunnel.api.source.SupportSchemaEvolution;
import org.apache.seatunnel.api.table.catalog.CatalogTable;
import org.apache.seatunnel.api.table.schema.SchemaChangeType;
@@ -35,12 +34,12 @@ import
org.apache.seatunnel.connectors.cdc.base.option.StartupMode;
import org.apache.seatunnel.connectors.cdc.base.option.StopMode;
import org.apache.seatunnel.connectors.cdc.base.schema.SchemaChangeEventFilter;
import org.apache.seatunnel.connectors.cdc.base.schema.SchemaChangeResolver;
-import org.apache.seatunnel.connectors.cdc.base.source.IncrementalSource;
import org.apache.seatunnel.connectors.cdc.base.source.offset.OffsetFactory;
import
org.apache.seatunnel.connectors.cdc.debezium.DebeziumDeserializationSchema;
import org.apache.seatunnel.connectors.cdc.debezium.DeserializeFormat;
import
org.apache.seatunnel.connectors.cdc.debezium.row.DebeziumJsonDeserializeSchema;
import
org.apache.seatunnel.connectors.cdc.debezium.row.SeaTunnelRowDebeziumDeserializeSchema;
+import
org.apache.seatunnel.connectors.seatunnel.cdc.pgbase.source.PgBaseIncrementalSource;
import
org.apache.seatunnel.connectors.seatunnel.cdc.postgres.config.PostgresIncrementalSourceOptions;
import
org.apache.seatunnel.connectors.seatunnel.cdc.postgres.config.PostgresSourceConfigFactory;
import
org.apache.seatunnel.connectors.seatunnel.cdc.postgres.source.offset.LsnOffsetFactory;
@@ -48,31 +47,38 @@ import
org.apache.seatunnel.connectors.seatunnel.jdbc.config.JdbcCommonOptions;
import org.apache.kafka.connect.data.Struct;
-import io.debezium.jdbc.JdbcConnection;
import io.debezium.relational.TableId;
-import io.debezium.relational.history.ConnectTableChangeSerializer;
-import io.debezium.relational.history.TableChanges;
-import io.debezium.util.SchemaNameAdjuster;
import java.time.ZoneId;
import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.Optional;
-import java.util.function.Function;
-import java.util.stream.Collectors;
-public class PostgresIncrementalSource<T> extends IncrementalSource<T,
JdbcSourceConfig>
- implements SupportParallelism, SupportSchemaEvolution {
+/**
+ * PostgreSQL incremental source backed by the shared PG-base source behavior.
+ *
+ * <p>Keeps PostgreSQL-specific options, dialect selection, offset handling
and RELATION-message
+ * schema evolution in this connector.
+ */
+public class PostgresIncrementalSource<T> extends PgBaseIncrementalSource<T,
JdbcSourceConfig>
+ implements SupportSchemaEvolution {
+
+ /**
+ * Preserves serialized job DAG compatibility with the Postgres source
released in 2.3.13.
+ *
+ * <p>Cross-checked with {@code serialver} against the real {@code
PostgresIncrementalSource}
+ * class shipped in the {@code connector-cdc-postgres} 2.3.12 and 2.3.13
Maven Central
+ * artifacts, where both versions compute this same default UID (2.3.11
computes a different
+ * one, so it was already incompatible with 2.3.12 before this PR). This
constant is not a
+ * guess; do not regenerate it when the class hierarchy or implementation
changes.
+ */
+ private static final long serialVersionUID = -9086519839702872016L;
static final String IDENTIFIER = "Postgres-CDC";
- private final boolean requireReplicaIdentityFull;
-
public PostgresIncrementalSource(ReadonlyConfig options,
List<CatalogTable> catalogTables) {
super(options, catalogTables);
- this.requireReplicaIdentityFull =
-
options.get(PostgresIncrementalSourceOptions.REQUIRE_REPLICA_IDENTITY_FULL);
validateSchemaEvolutionOptions(options);
}
@@ -111,11 +117,19 @@ public class PostgresIncrementalSource<T> extends
IncrementalSource<T, JdbcSourc
return configFactory;
}
+ /**
+ * Adds the PostgreSQL RELATION-message schema-change resolver and filter
on top of the shared
+ * PG-base deserialization schema.
+ *
+ * <p>These two are kept here rather than in {@link
PgBaseIncrementalSource} because they are
+ * specific to how PostgreSQL's {@code pgoutput} plugin reports schema
changes; a sibling
+ * PG-base connector is not guaranteed to share the same schema-change
wire format.
+ */
@SuppressWarnings("unchecked")
@Override
public DebeziumDeserializationSchema<T>
createDebeziumDeserializationSchema(
ReadonlyConfig config) {
- Map<TableId, Struct> tableIdTableChangeMap = tableChanges();
+ Map<TableId, Struct> tableIdTableChangeMap = loadTableChanges();
if (DeserializeFormat.COMPATIBLE_DEBEZIUM_JSON.equals(
config.get(JdbcSourceOptions.FORMAT))) {
return (DebeziumDeserializationSchema<T>)
@@ -124,11 +138,11 @@ public class PostgresIncrementalSource<T> extends
IncrementalSource<T, JdbcSourc
tableIdTableChangeMap);
}
- String zoneId = config.get(JdbcSourceOptions.SERVER_TIME_ZONE);
return (DebeziumDeserializationSchema<T>)
SeaTunnelRowDebeziumDeserializeSchema.builder()
.setTables(catalogTables)
- .setServerTimeZone(ZoneId.of(zoneId))
+ .setServerTimeZone(
+
ZoneId.of(config.get(JdbcSourceOptions.SERVER_TIME_ZONE)))
.setTableIdTableChangeMap(tableIdTableChangeMap)
.setSchemaChangeResolver(createSchemaChangeResolver(config))
.setSchemaChangeEventFilter(SchemaChangeEventFilter.fromConfig(config))
@@ -146,7 +160,7 @@ public class PostgresIncrementalSource<T> extends
IncrementalSource<T, JdbcSourc
return new PostgresDialect(
(PostgresSourceConfigFactory) configFactory,
catalogTables,
- requireReplicaIdentityFull);
+
config.get(PostgresIncrementalSourceOptions.REQUIRE_REPLICA_IDENTITY_FULL));
}
@Override
@@ -165,6 +179,12 @@ public class PostgresIncrementalSource<T> extends
IncrementalSource<T, JdbcSourc
return Collections.singletonList(SchemaChangeType.ADD_COLUMN);
}
+ /**
+ * Validates PostgreSQL committed-offset startup before source config
creation.
+ *
+ * <p>Debezium can only resume from a committed LSN when the replication
slot is explicit and
+ * stable across job attempts.
+ */
private void validateStartupOptions(ReadonlyConfig options, StartupConfig
startupConfig) {
if (startupConfig.getStartupMode() != StartupMode.COMMITTED_OFFSET) {
return;
@@ -182,6 +202,12 @@ public class PostgresIncrementalSource<T> extends
IncrementalSource<T, JdbcSourc
}
}
+ /**
+ * Rejects schema evolution configured with a decoding plugin that cannot
report schema changes.
+ *
+ * <p>Only {@code pgoutput} emits Debezium RELATION messages; the other
supported plugins give
+ * this connector no signal that a table's schema changed.
+ */
private void validateSchemaEvolutionOptions(ReadonlyConfig options) {
if (options.get(SourceOptions.SCHEMA_CHANGES_ENABLED)
&& !"pgoutput"
@@ -194,34 +220,4 @@ public class PostgresIncrementalSource<T> extends
IncrementalSource<T, JdbcSourc
PostgresIncrementalSourceOptions.DECODING_PLUGIN_NAME.key()));
}
}
-
- private Map<TableId, Struct> tableChanges() {
- JdbcSourceConfig jdbcSourceConfig = configFactory.create(0);
- PostgresDialect dialect =
- new PostgresDialect(
- (PostgresSourceConfigFactory) configFactory,
- catalogTables,
- requireReplicaIdentityFull);
- List<TableId> discoverTables =
dialect.discoverDataCollections(jdbcSourceConfig);
- SchemaNameAdjuster adjuster = SchemaNameAdjuster.create();
- ConnectTableChangeSerializer connectTableChangeSerializer =
- new ConnectTableChangeSerializer(adjuster);
- try (JdbcConnection jdbcConnection =
dialect.openJdbcConnection(jdbcSourceConfig)) {
- return discoverTables.stream()
- .collect(
- Collectors.toMap(
- Function.identity(),
- (tableId) -> {
- TableChanges tableChanges = new
TableChanges();
- tableChanges.create(
-
dialect.queryTableSchema(jdbcConnection, tableId)
- .getTable());
- return connectTableChangeSerializer
- .serialize(tableChanges)
- .get(0);
- }));
- } catch (Exception e) {
- throw new SeaTunnelException(e);
- }
- }
}
diff --git
a/seatunnel-connectors-v2/connector-cdc/connector-cdc-postgres/src/test/java/org/apache/seatunnel/connectors/seatunnel/cdc/postgres/config/PostgresSourceConfigFactoryTest.java
b/seatunnel-connectors-v2/connector-cdc/connector-cdc-postgres/src/test/java/org/apache/seatunnel/connectors/seatunnel/cdc/postgres/config/PostgresSourceConfigFactoryTest.java
index b531c60f12..b18094753a 100644
---
a/seatunnel-connectors-v2/connector-cdc/connector-cdc-postgres/src/test/java/org/apache/seatunnel/connectors/seatunnel/cdc/postgres/config/PostgresSourceConfigFactoryTest.java
+++
b/seatunnel-connectors-v2/connector-cdc/connector-cdc-postgres/src/test/java/org/apache/seatunnel/connectors/seatunnel/cdc/postgres/config/PostgresSourceConfigFactoryTest.java
@@ -18,31 +18,157 @@
package org.apache.seatunnel.connectors.seatunnel.cdc.postgres.config;
import org.apache.seatunnel.connectors.cdc.base.config.StartupConfig;
+import org.apache.seatunnel.connectors.cdc.base.config.StopConfig;
import org.apache.seatunnel.connectors.cdc.base.option.StartupMode;
+import org.apache.seatunnel.connectors.cdc.base.option.StopMode;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
+import java.util.Properties;
+
+/** Tests the PG-base-backed PostgreSQL source config factory behavior that
must stay compatible. */
public class PostgresSourceConfigFactoryTest {
+ @Test
+ public void testCreateFormatsSchemaQualifiedTableIdentifiers() {
+ PostgresSourceConfigFactory factory = baseFactory();
+ factory.tableList("inventory.orders", "db1.public.customers");
+
+ PostgresSourceConfig sourceConfig = factory.create(0);
+
+ Assertions.assertEquals(
+ "inventory.orders,public.customers",
+
sourceConfig.getDbzConfiguration().getString("table.include.list"));
+ }
+
+ @Test
+ public void testCreateRejectsInvalidTableIdentifier() {
+ PostgresSourceConfigFactory factory = baseFactory();
+ factory.tableList("orders");
+
+ IllegalArgumentException exception =
+ Assertions.assertThrows(IllegalArgumentException.class, () ->
factory.create(0));
+
+ // Pin the wording: this message is user-facing and predates the
PG-base extraction.
+ Assertions.assertEquals(
+ "Invalid table name: orders ,Postgres identifier is of the
form schemaName.tableName",
+ exception.getMessage());
+ }
+
@Test
public void shouldDisableDebeziumSnapshotForCommittedOffsetStartup() {
- PostgresSourceConfigFactory configFactory =
- (PostgresSourceConfigFactory)
- new PostgresSourceConfigFactory()
- .hostname("localhost")
- .username("user")
- .password("password")
- .databaseList("database")
- .startupOptions(
- new StartupConfig(
- StartupMode.COMMITTED_OFFSET,
null, null, null));
-
- Assertions.assertEquals(
- "never",
configFactory.create(0).getDbzConfiguration().getString("snapshot.mode"));
+ PostgresSourceConfigFactory factory = baseFactory();
+ factory.startupOptions(new StartupConfig(StartupMode.COMMITTED_OFFSET,
null, null, null));
+
+ Assertions.assertEquals(
+ "never",
factory.create(0).getDbzConfiguration().getString("snapshot.mode"));
// "database.include.list" must stay unset: Debezium turns it into a
catalog predicate on
// dataCollectionFilter(), which rejects the catalog-less TableIds
used by PostgreSQL.
Assertions.assertNull(
-
configFactory.create(0).getDbzConfiguration().getString("database.include.list"));
+
factory.create(0).getDbzConfiguration().getString("database.include.list"));
+ }
+
+ @Test
+ public void shouldRunDebeziumSnapshotOnlyForSnapshotOnlyStartup() {
+ PostgresSourceConfigFactory factory = baseFactory();
+ factory.startupOptions(new StartupConfig(StartupMode.SNAPSHOT_ONLY,
null, null, null));
+
+ Assertions.assertEquals(
+ "initial_only",
factory.create(0).getDbzConfiguration().getString("snapshot.mode"));
+ }
+
+ @Test
+ public void shouldLeaveSnapshotModeUnsetForInitialStartup() {
+ PostgresSourceConfigFactory factory = baseFactory();
+
+ Assertions.assertNull(
+
factory.create(0).getDbzConfiguration().getString("snapshot.mode"),
+ "initial startup must keep the Debezium default snapshot
mode");
+ }
+
+ @Test
+ public void shouldKeepPostgresSpecificDebeziumProperties() {
+ PostgresSourceConfigFactory factory = baseFactory();
+ factory.tableList("inventory.orders");
+
+ PostgresSourceConfig sourceConfig = factory.create(0);
+
+ Assertions.assertEquals(
+ "postgres_cdc_source",
+
sourceConfig.getDbzConfiguration().getString("database.server.name"));
+ Assertions.assertEquals(
+
PostgresIncrementalSourceOptions.DECODING_PLUGIN_NAME.defaultValue(),
+ sourceConfig.getDbzConfiguration().getString("plugin.name"));
+ Assertions.assertEquals(
+ PostgresIncrementalSourceOptions.SLOT_NAME.defaultValue(),
+ sourceConfig.getDbzConfiguration().getString("slot.name"));
+ Assertions.assertEquals("org.postgresql.Driver",
sourceConfig.getDriverClassName());
+ }
+
+ @Test
+ public void shouldLetUserDebeziumPropertiesOverrideDefaults() {
+ PostgresSourceConfigFactory factory = baseFactory();
+ Properties dbzProperties = new Properties();
+ dbzProperties.setProperty("slot.name", "custom_slot");
+ factory.debeziumProperties(dbzProperties);
+
+ Assertions.assertEquals(
+ "custom_slot",
factory.create(0).getDbzConfiguration().getString("slot.name"));
+ }
+
+ /**
+ * Pins the RELATION-message schema-evolution wiring: {@code
include.schema.changes} must
+ * reflect the actual {@code schema-changes.enabled} option instead of
being hardcoded, since
+ * Debezium PostgreSQL only emits the RELATION messages {@link
PostgresIncrementalSource}'s
+ * schema-change resolver depends on when this flag is set.
+ */
+ @Test
+ public void shouldEnableSchemaChangesWhenSchemaEvolutionIsEnabled() {
+ PostgresSourceConfigFactory factory = baseFactory();
+ factory.schemaChangeEnabled(true);
+
+ Assertions.assertEquals(
+ "true",
+
factory.create(0).getDbzConfiguration().getString("include.schema.changes"));
+ }
+
+ @Test
+ public void shouldDisableSchemaChangesByDefault() {
+ PostgresSourceConfigFactory factory = baseFactory();
+
+ Assertions.assertEquals(
+ "false",
+
factory.create(0).getDbzConfiguration().getString("include.schema.changes"));
+ }
+
+ /**
+ * SeaTunnel's own {@code schema-changes.enabled} option must stay
authoritative even if a raw
+ * {@code debezium.*} passthrough property also happens to set {@code
include.schema.changes}.
+ */
+ @Test
+ public void
shouldKeepSchemaChangesEnabledAuthoritativeOverUserDebeziumProperties() {
+ PostgresSourceConfigFactory factory = baseFactory();
+ factory.schemaChangeEnabled(true);
+ Properties dbzProperties = new Properties();
+ dbzProperties.setProperty("include.schema.changes", "false");
+ factory.debeziumProperties(dbzProperties);
+
+ Assertions.assertEquals(
+ "true",
+
factory.create(0).getDbzConfiguration().getString("include.schema.changes"));
+ }
+
+ private PostgresSourceConfigFactory baseFactory() {
+ PostgresSourceConfigFactory factory = new
PostgresSourceConfigFactory();
+ factory.hostname("127.0.0.1");
+ factory.port(5432);
+ factory.username("user");
+ factory.password("pwd");
+ factory.originUrl("jdbc:postgresql://127.0.0.1:5432/test");
+ factory.databaseList("inventory");
+ factory.startupOptions(new StartupConfig(StartupMode.INITIAL, null,
null, null));
+ factory.stopOptions(new StopConfig(StopMode.NEVER, null, null, null));
+ return factory;
}
}
diff --git
a/seatunnel-connectors-v2/connector-cdc/connector-cdc-postgres/src/test/java/org/apache/seatunnel/connectors/seatunnel/cdc/postgres/source/PostgresIncrementalSourceTest.java
b/seatunnel-connectors-v2/connector-cdc/connector-cdc-postgres/src/test/java/org/apache/seatunnel/connectors/seatunnel/cdc/postgres/source/PostgresIncrementalSourceTest.java
new file mode 100644
index 0000000000..a5d5c1e2c2
--- /dev/null
+++
b/seatunnel-connectors-v2/connector-cdc/connector-cdc-postgres/src/test/java/org/apache/seatunnel/connectors/seatunnel/cdc/postgres/source/PostgresIncrementalSourceTest.java
@@ -0,0 +1,181 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.seatunnel.connectors.seatunnel.cdc.postgres.source;
+
+import org.apache.seatunnel.api.configuration.ReadonlyConfig;
+import org.apache.seatunnel.api.options.ConnectorCommonOptions;
+import org.apache.seatunnel.api.table.catalog.CatalogTable;
+import org.apache.seatunnel.api.table.catalog.CatalogTableUtil;
+import org.apache.seatunnel.api.table.type.BasicType;
+import org.apache.seatunnel.api.table.type.SeaTunnelDataType;
+import org.apache.seatunnel.api.table.type.SeaTunnelRow;
+import org.apache.seatunnel.api.table.type.SeaTunnelRowType;
+import org.apache.seatunnel.common.utils.SeaTunnelException;
+import org.apache.seatunnel.connectors.cdc.base.option.JdbcSourceOptions;
+import
org.apache.seatunnel.connectors.cdc.debezium.DebeziumDeserializationSchema;
+import
org.apache.seatunnel.connectors.seatunnel.cdc.postgres.config.PostgresIncrementalSourceOptions;
+import org.apache.seatunnel.connectors.seatunnel.jdbc.config.JdbcCommonOptions;
+
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+import io.debezium.connector.postgresql.connection.PostgresConnection;
+import io.debezium.connector.postgresql.connection.ServerInfo;
+import io.debezium.relational.TableId;
+
+import java.io.ObjectStreamClass;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+/**
+ * Pins that {@code createDataSourceDialect()} resolves {@code
require-replica-identity-full} from
+ * the job config.
+ *
+ * <p>This matters because the value used to be read into a {@code final}
field assigned after
+ * {@code super(...)} returned, while {@code createDataSourceDialect()} is
invoked from inside that
+ * constructor — so the dialect always saw the Java default {@code false} and
the check was silently
+ * a no-op. Reading it inside {@code createDataSourceDialect(config)} restores
the documented
+ * default of {@code true}; without this test, the same timing bug could be
reintroduced unnoticed.
+ */
+public class PostgresIncrementalSourceTest {
+
+ private static final TableId ORDERS = new TableId(null, "public",
"orders");
+
+ /**
+ * Locks the UID computed from the released 2.3.13 class for persisted job
DAG upgrades.
+ *
+ * <p>Changing the constant would make released logical DAG payloads fail
deserialization.
+ */
+ @Test
+ public void testSerialVersionUidMatchesReleasedPostgresSource() {
+ Assertions.assertEquals(
+ -9086519839702872016L,
+
ObjectStreamClass.lookup(PostgresIncrementalSource.class).getSerialVersionUID());
+ }
+
+ @Test
+ public void testDialectEnforcesReplicaIdentityFullByDefault() throws
Exception {
+ PostgresDialect dialect = buildDialect(new HashMap<>());
+
+ SeaTunnelException exception =
+ Assertions.assertThrows(
+ SeaTunnelException.class,
+ () ->
+ dialect.checkAllTablesEnabledCapture(
+
connectionReporting(ServerInfo.ReplicaIdentity.DEFAULT),
+ Collections.singletonList(ORDERS)));
+
+ Assertions.assertTrue(exception.getMessage().contains("full replica
identity"));
+
Assertions.assertTrue(exception.getMessage().contains("public.orders"));
+ // Both remediations must be named: on enumerator restore the config
cannot be edited.
+ Assertions.assertTrue(exception.getMessage().contains("REPLICA
IDENTITY FULL"));
+ Assertions.assertTrue(
+ exception.getMessage().contains("require-replica-identity-full
= false"));
+ }
+
+ @Test
+ public void testDialectHonorsConfiguredReplicaIdentityOptOut() throws
Exception {
+ Map<String, Object> options = new HashMap<>();
+
options.put(PostgresIncrementalSourceOptions.REQUIRE_REPLICA_IDENTITY_FULL.key(),
false);
+
+ PostgresDialect dialect = buildDialect(options);
+
+ Assertions.assertDoesNotThrow(
+ () ->
+ dialect.checkAllTablesEnabledCapture(
+
connectionReporting(ServerInfo.ReplicaIdentity.DEFAULT),
+ Collections.singletonList(ORDERS)));
+ }
+
+ @Test
+ public void testDialectAcceptsFullReplicaIdentityWhenEnforced() throws
Exception {
+ PostgresDialect dialect = buildDialect(new HashMap<>());
+
+ Assertions.assertDoesNotThrow(
+ () ->
+ dialect.checkAllTablesEnabledCapture(
+
connectionReporting(ServerInfo.ReplicaIdentity.FULL),
+ Collections.singletonList(ORDERS)));
+ }
+
+ /**
+ * Builds the dialect through the real {@code
createDataSourceDialect(config)} entry point, so
+ * the test exercises the wiring under review rather than constructing the
dialect directly.
+ */
+ private static PostgresDialect buildDialect(Map<String, Object>
extraOptions) {
+ Map<String, Object> options = new HashMap<>(baseOptions());
+ options.putAll(extraOptions);
+ ReadonlyConfig config = ReadonlyConfig.fromMap(options);
+
+ TestingPostgresIncrementalSource source = new
TestingPostgresIncrementalSource(config);
+ return (PostgresDialect) source.createDataSourceDialect(config);
+ }
+
+ private static Map<String, Object> baseOptions() {
+ Map<String, Object> options = new HashMap<>();
+ options.put(JdbcCommonOptions.URL.key(),
"jdbc:postgresql://127.0.0.1:5432/inventory");
+ options.put(JdbcSourceOptions.USERNAME.key(), "user");
+ options.put(JdbcSourceOptions.PASSWORD.key(), "pwd");
+ options.put(JdbcSourceOptions.DATABASE_NAMES.key(),
Collections.singletonList("inventory"));
+ options.put(
+ ConnectorCommonOptions.TABLE_NAMES.key(),
+ Collections.singletonList("inventory.public.orders"));
+ return options;
+ }
+
+ /** A connection whose replica-identity probe reports the given identity
for every table. */
+ private static PostgresConnection
connectionReporting(ServerInfo.ReplicaIdentity identity)
+ throws Exception {
+ PostgresConnection connection = mock(PostgresConnection.class);
+
when(connection.readReplicaIdentityInfo(any(TableId.class))).thenReturn(identity);
+ return connection;
+ }
+
+ private static List<CatalogTable> catalogTables() {
+ SeaTunnelRowType rowType =
+ new SeaTunnelRowType(
+ new String[] {"id"}, new SeaTunnelDataType[]
{BasicType.INT_TYPE});
+ return Collections.singletonList(
+ CatalogTableUtil.getCatalogTable(
+ "postgres", "inventory", "public", "orders", rowType));
+ }
+
+ /**
+ * Keeps the real config-factory and dialect wiring and only stubs out
schema loading, which
+ * would otherwise open a JDBC connection during construction.
+ */
+ private static final class TestingPostgresIncrementalSource
+ extends PostgresIncrementalSource<SeaTunnelRow> {
+
+ private TestingPostgresIncrementalSource(ReadonlyConfig options) {
+ super(options, catalogTables());
+ }
+
+ @Override
+ public DebeziumDeserializationSchema<SeaTunnelRow>
createDebeziumDeserializationSchema(
+ ReadonlyConfig config) {
+ return null;
+ }
+ }
+}
diff --git a/seatunnel-connectors-v2/connector-cdc/pom.xml
b/seatunnel-connectors-v2/connector-cdc/pom.xml
index f459bc109c..3bf905a32c 100644
--- a/seatunnel-connectors-v2/connector-cdc/pom.xml
+++ b/seatunnel-connectors-v2/connector-cdc/pom.xml
@@ -31,6 +31,7 @@
<modules>
<module>connector-cdc-base</module>
+ <module>connector-cdc-pg-base</module>
<module>connector-cdc-db2</module>
<module>connector-cdc-mysql</module>
<module>connector-cdc-oceanbase</module>
diff --git
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-postgres-e2e/src/test/java/org/apache/seatunnel/connectors/seatunnel/cdc/postgres/PostgresCDCIT.java
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-postgres-e2e/src/test/java/org/apache/seatunnel/connectors/seatunnel/cdc/postgres/PostgresCDCIT.java
index bd889c00ed..beac962e97 100644
---
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-postgres-e2e/src/test/java/org/apache/seatunnel/connectors/seatunnel/cdc/postgres/PostgresCDCIT.java
+++
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-postgres-e2e/src/test/java/org/apache/seatunnel/connectors/seatunnel/cdc/postgres/PostgresCDCIT.java
@@ -1246,7 +1246,8 @@ public class PostgresCDCIT extends TestSuiteBase
implements TestResource {
Collections.singletonList(
TableId.parse(SINK_TABLE_1))));
Assertions.assertEquals(
- "Table sink_postgres_cdc_table_1 does not have a full
replica identity, please execute: ALTER TABLE sink_postgres_cdc_table_1 REPLICA
IDENTITY FULL;",
+ "Table sink_postgres_cdc_table_1 does not have a full
replica identity, please execute: ALTER TABLE sink_postgres_cdc_table_1 REPLICA
IDENTITY FULL; "
+ + "or set require-replica-identity-full = false to
accept UPDATE/DELETE events without the previous row state.",
exception.getMessage());
}
}
diff --git a/tools/check_connector_registration.py
b/tools/check_connector_registration.py
index 81df370957..c3e3ff7bfd 100644
--- a/tools/check_connector_registration.py
+++ b/tools/check_connector_registration.py
@@ -62,6 +62,7 @@ EXCLUDED_FROM_ALL = {
"connector-file-base",
"connector-file-base-hadoop",
"connector-cdc-base",
+ "connector-cdc-pg-base",
}
# Connectors bundled into the seatunnel-dist core jar (always available,
