This is an automated email from the ASF dual-hosted git repository.
davidzollo pushed a commit to branch dev
in repository https://gitbox.apache.org/repos/asf/seatunnel.git
The following commit(s) were added to refs/heads/dev by this push:
new ddcc764d20 [Feature][Connector-V2][PostgreSQL-CDC] Add
committed-offset and snapshot-only startup modes (#11380)
ddcc764d20 is described below
commit ddcc764d209979b9968b5ae9b062041ed341bd43
Author: cloverdew <[email protected]>
AuthorDate: Sat Jul 25 20:26:12 2026 +0800
[Feature][Connector-V2][PostgreSQL-CDC] Add committed-offset and
snapshot-only startup modes (#11380)
---
docs/en/connectors/source/PostgreSQL-CDC.md | 6 +-
docs/zh/connectors/source/PostgreSQL-CDC.md | 11 +-
.../connectors/cdc/base/config/StartupConfig.java | 3 +
.../connectors/cdc/base/option/StartupMode.java | 11 +-
.../cdc/base/source/IncrementalSource.java | 56 +++-
.../source/enumerator/HybridSplitAssigner.java | 6 +-
.../enumerator/IncrementalSplitAssigner.java | 47 +++-
.../enumerator/SnapshotOnlySplitAssigner.java | 96 +++++++
.../enumerator/state/IncrementalPhaseState.java | 18 +-
.../cdc/base/source/offset/OffsetFactory.java | 5 +
.../enumerator/IncrementalSplitAssignerTest.java | 80 ++++++
.../config/PostgresSourceConfigFactory.java | 7 +
.../exception/PostgresConnectorErrorCode.java | 4 +-
.../postgres/source/PostgresIncrementalSource.java | 25 ++
.../source/PostgresIncrementalSourceFactory.java | 4 +
.../cdc/postgres/source/PostgresSourceOptions.java | 4 +-
.../postgres/source/offset/LsnOffsetFactory.java | 78 ++++++
.../reader/PostgresSourceFetchTaskContext.java | 30 +-
.../config/PostgresSourceConfigFactoryTest.java | 44 +++
.../source/offset/LsnOffsetFactoryTest.java | 109 ++++++++
.../seatunnel/cdc/postgres/PostgresCDCIT.java | 309 +++++++++++++++++++++
.../postgrescdc_to_postgres_committed_offset.conf | 53 ++++
.../postgrescdc_to_postgres_snapshot_only.conf | 51 ++++
23 files changed, 1013 insertions(+), 44 deletions(-)
diff --git a/docs/en/connectors/source/PostgreSQL-CDC.md
b/docs/en/connectors/source/PostgreSQL-CDC.md
index a7601b6285..7c1b5f54e4 100644
--- a/docs/en/connectors/source/PostgreSQL-CDC.md
+++ b/docs/en/connectors/source/PostgreSQL-CDC.md
@@ -97,7 +97,7 @@ ALTER TABLE your_table_name REPLICA IDENTITY FULL;
| table-names | List | Yes, if
`table-pattern` is not used | - | Tables to monitor. Use the fully
qualified `database.schema.table` format, for example:
`postgres_cdc.inventory.orders`.
[...]
| 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:
`postgres_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 PostgreSQL CDC consumer, valid enumerations are
`initial`, `earliest` and `latest`. <br/> `initial`: Synchronize historical
data at startup, and then synchronize incremental data.<br/> `earliest`:
Startup from the earliest offset possible.<br/> `latest`: Startup from the
latest offset.
[...]
+| startup.mode | Enum | No | INITIAL |
Optional startup mode for PostgreSQL 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 s [...]
| 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.
[...]
| slot.name | String | No | seatunnel
| The PostgreSQL logical decoding slot name. Use a different slot name for each
CDC job that reads from the same PostgreSQL instance.
[...]
@@ -112,7 +112,7 @@ ALTER TABLE your_table_name REPLICA IDENTITY FULL;
| 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. |
| enable_concurrent_read | Boolean | No | true |
Whether to enable concurrent read with split during the snapshot phase. When
set to false, the source skips split analysis and reads the table as a single
split, which is useful for tables without indexes. 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 PostgreSQL CDC, valid enumerations are `DEFAULT`,
`COMPATIBLE_DEBEZIUM_JSON`.
[...]
| 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.
[...]
| 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
PostgreSQL server.
[...]
@@ -251,6 +251,8 @@ By default, PostgreSQL CDC requires primary keys. You can
specify a custom prima
### How are replication slots managed?
SeaTunnel creates or reuses the replication slot identified by `slot.name`
when the job starts.
+When `startup.mode` is `committed-offset`, the replication slot must already
exist because SeaTunnel
+uses its `confirmed_flush_lsn` as the startup offset.
Unused replication slots hold WAL segments on disk, which can cause unbounded
WAL growth. When a
CDC job is permanently decommissioned, drop the unused replication slot
manually on PostgreSQL.
diff --git a/docs/zh/connectors/source/PostgreSQL-CDC.md
b/docs/zh/connectors/source/PostgreSQL-CDC.md
index 521635ad25..de1eb73efb 100644
--- a/docs/zh/connectors/source/PostgreSQL-CDC.md
+++ b/docs/zh/connectors/source/PostgreSQL-CDC.md
@@ -95,7 +95,7 @@ ALTER TABLE your_table_name REPLICA IDENTITY FULL;
| table-names | List | 二选一 | - |
需要监控的表。请使用完整的 `database.schema.table` 格式，例如：`postgres_cdc.inventory.orders`。
[...]
| table-pattern | String | 二选一 | - |
需要监控的表名正则表达式。正则需要匹配完整表名，例如：`postgres_cdc\\.inventory\\..*`。`table-names` 和
`table-pattern` 互斥。
[...]
| table-names-config | List | 否 | - |
表级配置列表。例如：`[{"table": "db1.schema1.table1","primaryKeys":
["key1"],"snapshotSplitColumn": "key2"}]`。无物理主键表可通过 `primaryKeys`
指定唯一键。`snapshotSplitColumn` 必须是唯一键，否则 SeaTunnel 会忽略该配置并自动选择拆分列。
[...]
-| startup.mode | List | 否 | INITIAL |
PostgreSQL CDC 消费者的可选启动模式，有效枚举为 `initial`、`earliest` 和 `latest`。<br/>
`initial`: 启动时同步历史数据，然后同步增量数据。<br/> `earliest`: 从可能的最早偏移量启动。<br/> `latest`:
从最新偏移量启动。
[...]
+| startup.mode | Enum | 否 | INITIAL |
PostgreSQL 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`: 从最新偏移量启动。 |
| snapshot.split.size | Integer | 否 | 8096 |
表快照的拆分大小（行数），捕获的表在读取表快照时被拆分成多个拆分。
[...]
| snapshot.fetch.size | Integer | 否 | 1024 |
读取表快照时每次轮询的最大获取大小。
[...]
| slot.name | String | 否 | seatunnel |
PostgreSQL 逻辑解码槽名称。同一个 PostgreSQL 实例上如果有多个 CDC 任务，请为每个任务配置不同的 `slot.name`。
|
@@ -110,7 +110,7 @@ ALTER TABLE your_table_name REPLICA IDENTITY FULL;
| inverse-sampling.rate | Integer | 否 | 1000 |
在采样分片策略中使用的采样率的倒数。例如，如果此值设置为 1000，则意味着在采样过程中应用 1/1000
的采样率。此选项提供了控制采样粒度的灵活性，从而影响最终的分片数量。在处理非常大数据集时，较低的采样率尤为有用。默认值为 1000。
|
| split.allow-sampling | Boolean | 否 | true |
是否允许基于采样的分片策略。当设置为 false 时，无论预估分片数是否超过阈值，系统都将回退到非均匀分片方式（迭代查询方式）。默认值为 true。 |
| enable_concurrent_read | Boolean | 否 | true |
是否在快照阶段启用基于分片的并发读取。当设置为 false 时，source 会跳过分片分析，并以单个 split 读取整张表，适合没有索引的表。默认值为
true。 |
-| exactly_once | Boolean | 否 | false |
在初始快照阶段启用精确一次语义。仅当 `startup.mode` 为 `initial` 时可用。
[...]
+| exactly_once | Boolean | 否 | false |
在快照阶段启用精确一次语义。仅当 `startup.mode` 为 `initial` 或 `snapshot-only` 时可用。
[...]
| format | Enum | 否 | DEFAULT |
PostgreSQL CDC 的可选输出格式，有效枚举为 `DEFAULT`、`COMPATIBLE_DEBEZIUM_JSON`。
[...]
| require-replica-identity-full | Boolean | 否 | true |
要求表具有 REPLICA IDENTITY FULL。设置为 false 时，允许表使用其他副本标识设置，但 UPDATE/DELETE
事件可能不包含之前的状态。此选项仅应用于仅追加的表（例如 outbox 模式）。默认为 true 以保持向后兼容性。
[...]
| debezium | Config | 否 | - | 将
[Debezium
的属性](https://github.com/debezium/debezium/blob/v1.9.8.Final/documentation/modules/ROOT/pages/connectors/postgresql.adoc#connector-configuration-properties)
传递给用于捕获 PostgreSQL 服务器数据更改的 Debezium 嵌入式引擎。
[...]
@@ -248,9 +248,10 @@ SeaTunnel PostgreSQL CDC 支持 `pgoutput`（PostgreSQL 10
起内置）、`wal2j
### 复制槽如何管理？
-SeaTunnel 在任务启动时会创建或复用 `slot.name` 指定的复制槽。未使用的复制槽会持续占用
-磁盘上的 WAL 段，导致 WAL 持续增长。当 CDC 任务永久下线时，应在 PostgreSQL 侧手动删除
-不再使用的复制槽。
+SeaTunnel 在任务启动时会创建或复用 `slot.name` 指定的复制槽。当 `startup.mode` 为
+`committed-offset` 时，复制槽必须已存在，因为 SeaTunnel 会使用其 `confirmed_flush_lsn`
+作为启动偏移量。未使用的复制槽会持续占用磁盘上的 WAL 段，导致 WAL 持续增长。当 CDC
+任务永久下线时，应在 PostgreSQL 侧手动删除不再使用的复制槽。
### PostgreSQL CDC 为什么会滞后？
diff --git
a/seatunnel-connectors-v2/connector-cdc/connector-cdc-base/src/main/java/org/apache/seatunnel/connectors/cdc/base/config/StartupConfig.java
b/seatunnel-connectors-v2/connector-cdc/connector-cdc-base/src/main/java/org/apache/seatunnel/connectors/cdc/base/config/StartupConfig.java
index 0f07bc4670..9dae773741 100644
---
a/seatunnel-connectors-v2/connector-cdc/connector-cdc-base/src/main/java/org/apache/seatunnel/connectors/cdc/base/config/StartupConfig.java
+++
b/seatunnel-connectors-v2/connector-cdc/connector-cdc-base/src/main/java/org/apache/seatunnel/connectors/cdc/base/config/StartupConfig.java
@@ -69,7 +69,10 @@ public final class StartupConfig implements Serializable {
case LATEST:
return offsetFactory.latest();
case INITIAL:
+ case SNAPSHOT_ONLY:
return null;
+ case COMMITTED_OFFSET:
+ return offsetFactory.committedOffset();
case SPECIFIC:
if (specificOffset != null) {
return offsetFactory.specific(specificOffset);
diff --git
a/seatunnel-connectors-v2/connector-cdc/connector-cdc-base/src/main/java/org/apache/seatunnel/connectors/cdc/base/option/StartupMode.java
b/seatunnel-connectors-v2/connector-cdc/connector-cdc-base/src/main/java/org/apache/seatunnel/connectors/cdc/base/option/StartupMode.java
index 9b3b607d9c..de74b0c47b 100644
---
a/seatunnel-connectors-v2/connector-cdc/connector-cdc-base/src/main/java/org/apache/seatunnel/connectors/cdc/base/option/StartupMode.java
+++
b/seatunnel-connectors-v2/connector-cdc/connector-cdc-base/src/main/java/org/apache/seatunnel/connectors/cdc/base/option/StartupMode.java
@@ -25,8 +25,17 @@ public enum StartupMode {
LATEST,
/** Synchronize historical data at startup, and then synchronize
incremental data. */
INITIAL,
+ /** Read historical data only and finish without entering incremental
reading. */
+ SNAPSHOT_ONLY,
+ /** Startup from the committed offset of a database-managed replication
position. */
+ COMMITTED_OFFSET,
/** Start from user-supplied timestamp. */
TIMESTAMP,
/** Startup from user-supplied specific offsets. */
- SPECIFIC
+ SPECIFIC;
+
+ @Override
+ public String toString() {
+ return name().toLowerCase().replace('_', '-');
+ }
}
diff --git
a/seatunnel-connectors-v2/connector-cdc/connector-cdc-base/src/main/java/org/apache/seatunnel/connectors/cdc/base/source/IncrementalSource.java
b/seatunnel-connectors-v2/connector-cdc/connector-cdc-base/src/main/java/org/apache/seatunnel/connectors/cdc/base/source/IncrementalSource.java
index 1c4e84a522..1e263508e7 100644
---
a/seatunnel-connectors-v2/connector-cdc/connector-cdc-base/src/main/java/org/apache/seatunnel/connectors/cdc/base/source/IncrementalSource.java
+++
b/seatunnel-connectors-v2/connector-cdc/connector-cdc-base/src/main/java/org/apache/seatunnel/connectors/cdc/base/source/IncrementalSource.java
@@ -44,6 +44,7 @@ import
org.apache.seatunnel.connectors.cdc.base.schema.SchemaChangeResolver;
import
org.apache.seatunnel.connectors.cdc.base.source.enumerator.HybridSplitAssigner;
import
org.apache.seatunnel.connectors.cdc.base.source.enumerator.IncrementalSourceEnumerator;
import
org.apache.seatunnel.connectors.cdc.base.source.enumerator.IncrementalSplitAssigner;
+import
org.apache.seatunnel.connectors.cdc.base.source.enumerator.SnapshotOnlySplitAssigner;
import
org.apache.seatunnel.connectors.cdc.base.source.enumerator.SplitAssigner;
import
org.apache.seatunnel.connectors.cdc.base.source.enumerator.state.HybridPendingSplitsState;
import
org.apache.seatunnel.connectors.cdc.base.source.enumerator.state.IncrementalPhaseState;
@@ -251,7 +252,10 @@ public abstract class IncrementalSource<T, C extends
SourceConfig>
@Override
public Boundedness getBoundedness() {
- return stopMode == StopMode.NEVER ? Boundedness.UNBOUNDED :
Boundedness.BOUNDED;
+ return stopMode == StopMode.NEVER
+ && startupConfig.getStartupMode() !=
StartupMode.SNAPSHOT_ONLY
+ ? Boundedness.UNBOUNDED
+ : Boundedness.BOUNDED;
}
@SuppressWarnings("MagicNumber")
@@ -316,20 +320,31 @@ public abstract class IncrementalSource<T, C extends
SourceConfig>
new HashSet<>(remainingTables),
new HashMap<>(),
new HashMap<>());
- if (sourceConfig.getStartupConfig().getStartupMode() ==
StartupMode.INITIAL) {
+ if (sourceConfig.getStartupConfig().getStartupMode() ==
StartupMode.INITIAL
+ || sourceConfig.getStartupConfig().getStartupMode() ==
StartupMode.SNAPSHOT_ONLY) {
try {
boolean isTableIdCaseSensitive =
dataSourceDialect.isDataCollectionIdCaseSensitive(sourceConfig);
- splitAssigner =
- new HybridSplitAssigner<>(
- assignerContext,
- enumeratorContext.currentParallelism(),
- incrementalParallelism,
- remainingTables,
- isTableIdCaseSensitive,
- dataSourceDialect,
- offsetFactory);
+ if (sourceConfig.getStartupConfig().getStartupMode() ==
StartupMode.SNAPSHOT_ONLY) {
+ splitAssigner =
+ new SnapshotOnlySplitAssigner<>(
+ assignerContext,
+ enumeratorContext.currentParallelism(),
+ remainingTables,
+ isTableIdCaseSensitive,
+ dataSourceDialect);
+ } else {
+ splitAssigner =
+ new HybridSplitAssigner<>(
+ assignerContext,
+ enumeratorContext.currentParallelism(),
+ incrementalParallelism,
+ remainingTables,
+ isTableIdCaseSensitive,
+ dataSourceDialect,
+ offsetFactory);
+ }
} catch (Exception e) {
throw new RuntimeException("Failed to discover captured tables
for enumerator", e);
}
@@ -378,13 +393,30 @@ public abstract class IncrementalSource<T, C extends
SourceConfig>
(HybridPendingSplitsState) checkpointState,
dataSourceDialect,
offsetFactory);
+ } else if (checkpointState instanceof SnapshotPhaseState) {
+ SnapshotPhaseState checkpointSnapshotState = (SnapshotPhaseState)
checkpointState;
+ SplitAssigner.Context<C> assignerContext =
+ new SplitAssigner.Context<>(
+ sourceConfig,
+ capturedTables,
+ checkpointSnapshotState.getAssignedSplits(),
+
checkpointSnapshotState.getSplitCompletedOffsets());
+ splitAssigner =
+ new SnapshotOnlySplitAssigner<>(
+ assignerContext,
+ enumeratorContext.currentParallelism(),
+ checkpointSnapshotState,
+ dataSourceDialect);
} else if (checkpointState instanceof IncrementalPhaseState) {
SplitAssigner.Context<C> assignerContext =
new SplitAssigner.Context<>(
sourceConfig, capturedTables, new HashMap<>(), new
HashMap<>());
splitAssigner =
new IncrementalSplitAssigner<>(
- assignerContext, incrementalParallelism,
offsetFactory);
+ assignerContext,
+ incrementalParallelism,
+ offsetFactory,
+ (IncrementalPhaseState) checkpointState);
} else {
throw new UnsupportedOperationException(
"Unsupported restored PendingSplitsState: " +
checkpointState);
diff --git
a/seatunnel-connectors-v2/connector-cdc/connector-cdc-base/src/main/java/org/apache/seatunnel/connectors/cdc/base/source/enumerator/HybridSplitAssigner.java
b/seatunnel-connectors-v2/connector-cdc/connector-cdc-base/src/main/java/org/apache/seatunnel/connectors/cdc/base/source/enumerator/HybridSplitAssigner.java
index 4acd092478..2409d53b85 100644
---
a/seatunnel-connectors-v2/connector-cdc/connector-cdc-base/src/main/java/org/apache/seatunnel/connectors/cdc/base/source/enumerator/HybridSplitAssigner.java
+++
b/seatunnel-connectors-v2/connector-cdc/connector-cdc-base/src/main/java/org/apache/seatunnel/connectors/cdc/base/source/enumerator/HybridSplitAssigner.java
@@ -76,7 +76,11 @@ public class HybridSplitAssigner<C extends SourceConfig>
implements SplitAssigne
this(
new SnapshotSplitAssigner<>(
context, currentParallelism,
checkpoint.getSnapshotPhaseState(), dialect),
- new IncrementalSplitAssigner<>(context,
incrementalParallelism, offsetFactory));
+ new IncrementalSplitAssigner<>(
+ context,
+ incrementalParallelism,
+ offsetFactory,
+ checkpoint.getIncrementalPhaseState()));
}
private HybridSplitAssigner(
diff --git
a/seatunnel-connectors-v2/connector-cdc/connector-cdc-base/src/main/java/org/apache/seatunnel/connectors/cdc/base/source/enumerator/IncrementalSplitAssigner.java
b/seatunnel-connectors-v2/connector-cdc/connector-cdc-base/src/main/java/org/apache/seatunnel/connectors/cdc/base/source/enumerator/IncrementalSplitAssigner.java
index 580f927d69..e47e08ecf8 100644
---
a/seatunnel-connectors-v2/connector-cdc/connector-cdc-base/src/main/java/org/apache/seatunnel/connectors/cdc/base/source/enumerator/IncrementalSplitAssigner.java
+++
b/seatunnel-connectors-v2/connector-cdc/connector-cdc-base/src/main/java/org/apache/seatunnel/connectors/cdc/base/source/enumerator/IncrementalSplitAssigner.java
@@ -21,6 +21,8 @@ import
org.apache.seatunnel.shade.com.google.common.annotations.VisibleForTestin
import org.apache.seatunnel.api.table.catalog.CatalogTable;
import org.apache.seatunnel.connectors.cdc.base.config.SourceConfig;
+import org.apache.seatunnel.connectors.cdc.base.config.StartupConfig;
+import org.apache.seatunnel.connectors.cdc.base.option.StartupMode;
import
org.apache.seatunnel.connectors.cdc.base.source.enumerator.state.IncrementalPhaseState;
import
org.apache.seatunnel.connectors.cdc.base.source.event.SnapshotSplitWatermark;
import org.apache.seatunnel.connectors.cdc.base.source.offset.Offset;
@@ -78,6 +80,13 @@ public class IncrementalSplitAssigner<C extends
SourceConfig> implements SplitAs
private List<CatalogTable> checkpointTables;
private Map<TableId, byte[]> historyTableChanges;
+ /**
+ * The startup offset resolved on fresh enumerator creation and reused
after checkpoint restore.
+ */
+ private Offset startupOffset;
+
+ private final boolean restoredFromCheckpoint;
+
public IncrementalSplitAssigner(
SplitAssigner.Context<C> context,
int incrementalParallelism,
@@ -85,6 +94,25 @@ public class IncrementalSplitAssigner<C extends
SourceConfig> implements SplitAs
this.context = context;
this.incrementalParallelism = incrementalParallelism;
this.offsetFactory = offsetFactory;
+ this.restoredFromCheckpoint = false;
+ StartupConfig startupConfig =
+ context.getSourceConfig() == null
+ ? null
+ : context.getSourceConfig().getStartupConfig();
+ this.startupOffset =
+ startupConfig == null ? null :
startupConfig.getStartupOffset(offsetFactory);
+ }
+
+ public IncrementalSplitAssigner(
+ SplitAssigner.Context<C> context,
+ int incrementalParallelism,
+ OffsetFactory offsetFactory,
+ IncrementalPhaseState checkpointState) {
+ this.context = context;
+ this.incrementalParallelism = incrementalParallelism;
+ this.offsetFactory = offsetFactory;
+ this.restoredFromCheckpoint = true;
+ this.startupOffset = checkpointState == null ? null :
checkpointState.getStartupOffset();
}
@Override
@@ -159,6 +187,9 @@ public class IncrementalSplitAssigner<C extends
SourceConfig> implements SplitAs
}
tableWatermarks.put(tableId, startupOffset);
}
+ if (this.startupOffset == null) {
+ this.startupOffset = startupOffset;
+ }
checkpointTables =
incrementalSplit.getCheckpointTables();
historyTableChanges =
incrementalSplit.getHistoryTableChanges();
});
@@ -169,7 +200,7 @@ public class IncrementalSplitAssigner<C extends
SourceConfig> implements SplitAs
@Override
public IncrementalPhaseState snapshotState(long checkpointId) {
- return new IncrementalPhaseState();
+ return new IncrementalPhaseState(startupOffset);
}
@Override
@@ -249,10 +280,16 @@ public class IncrementalSplitAssigner<C extends
SourceConfig> implements SplitAs
tableId);
}
}
- Offset incrementalSplitStartOffset =
- minOffset != null
- ? minOffset
- :
sourceConfig.getStartupConfig().getStartupOffset(offsetFactory);
+ if (minOffset == null && startupOffset == null) {
+ if (restoredFromCheckpoint
+ && sourceConfig.getStartupConfig().getStartupMode()
+ == StartupMode.COMMITTED_OFFSET) {
+ throw new IllegalStateException(
+ "The restored committed-offset checkpoint does not
contain its startup offset");
+ }
+ startupOffset =
sourceConfig.getStartupConfig().getStartupOffset(offsetFactory);
+ }
+ Offset incrementalSplitStartOffset = minOffset != null ? minOffset :
startupOffset;
return new IncrementalSplit(
String.format(INCREMENTAL_SPLIT_ID, index),
capturedTables,
diff --git
a/seatunnel-connectors-v2/connector-cdc/connector-cdc-base/src/main/java/org/apache/seatunnel/connectors/cdc/base/source/enumerator/SnapshotOnlySplitAssigner.java
b/seatunnel-connectors-v2/connector-cdc/connector-cdc-base/src/main/java/org/apache/seatunnel/connectors/cdc/base/source/enumerator/SnapshotOnlySplitAssigner.java
new file mode 100644
index 0000000000..713598fdf2
--- /dev/null
+++
b/seatunnel-connectors-v2/connector-cdc/connector-cdc-base/src/main/java/org/apache/seatunnel/connectors/cdc/base/source/enumerator/SnapshotOnlySplitAssigner.java
@@ -0,0 +1,96 @@
+/*
+ * 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.cdc.base.source.enumerator;
+
+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.source.enumerator.state.PendingSplitsState;
+import
org.apache.seatunnel.connectors.cdc.base.source.enumerator.state.SnapshotPhaseState;
+import
org.apache.seatunnel.connectors.cdc.base.source.event.SnapshotSplitWatermark;
+import org.apache.seatunnel.connectors.cdc.base.source.split.SourceSplitBase;
+
+import io.debezium.relational.TableId;
+
+import java.util.Collection;
+import java.util.List;
+import java.util.Optional;
+
+/** Assigner for bounded snapshot-only CDC jobs. */
+public class SnapshotOnlySplitAssigner<C extends SourceConfig> implements
SplitAssigner {
+
+ private final SnapshotSplitAssigner<C> snapshotSplitAssigner;
+
+ public SnapshotOnlySplitAssigner(
+ SplitAssigner.Context<C> context,
+ int currentParallelism,
+ List<TableId> remainingTables,
+ boolean isTableIdCaseSensitive,
+ DataSourceDialect<C> dialect) {
+ this.snapshotSplitAssigner =
+ new SnapshotSplitAssigner<>(
+ context,
+ currentParallelism,
+ remainingTables,
+ isTableIdCaseSensitive,
+ dialect);
+ }
+
+ public SnapshotOnlySplitAssigner(
+ SplitAssigner.Context<C> context,
+ int currentParallelism,
+ SnapshotPhaseState checkpoint,
+ DataSourceDialect<C> dialect) {
+ this.snapshotSplitAssigner =
+ new SnapshotSplitAssigner<>(context, currentParallelism,
checkpoint, dialect);
+ }
+
+ @Override
+ public void open() {
+ snapshotSplitAssigner.open();
+ }
+
+ @Override
+ public Optional<SourceSplitBase> getNext() {
+ return snapshotSplitAssigner.getNext();
+ }
+
+ @Override
+ public boolean waitingForCompletedSplits() {
+ return snapshotSplitAssigner.waitingForCompletedSplits();
+ }
+
+ @Override
+ public void onCompletedSplits(List<SnapshotSplitWatermark>
completedSplitWatermarks) {
+ snapshotSplitAssigner.onCompletedSplits(completedSplitWatermarks);
+ }
+
+ @Override
+ public void addSplits(Collection<SourceSplitBase> splits) {
+ snapshotSplitAssigner.addSplits(splits);
+ }
+
+ @Override
+ public PendingSplitsState snapshotState(long checkpointId) {
+ return snapshotSplitAssigner.snapshotState(checkpointId);
+ }
+
+ @Override
+ public void notifyCheckpointComplete(long checkpointId) {
+ snapshotSplitAssigner.notifyCheckpointComplete(checkpointId);
+ }
+}
diff --git
a/seatunnel-connectors-v2/connector-cdc/connector-cdc-base/src/main/java/org/apache/seatunnel/connectors/cdc/base/source/enumerator/state/IncrementalPhaseState.java
b/seatunnel-connectors-v2/connector-cdc/connector-cdc-base/src/main/java/org/apache/seatunnel/connectors/cdc/base/source/enumerator/state/IncrementalPhaseState.java
index fcf39c3471..80d6ced1fc 100644
---
a/seatunnel-connectors-v2/connector-cdc/connector-cdc-base/src/main/java/org/apache/seatunnel/connectors/cdc/base/source/enumerator/state/IncrementalPhaseState.java
+++
b/seatunnel-connectors-v2/connector-cdc/connector-cdc-base/src/main/java/org/apache/seatunnel/connectors/cdc/base/source/enumerator/state/IncrementalPhaseState.java
@@ -17,8 +17,24 @@
package org.apache.seatunnel.connectors.cdc.base.source.enumerator.state;
+import org.apache.seatunnel.connectors.cdc.base.source.offset.Offset;
+
import lombok.Data;
/** A {@link PendingSplitsState} for pending incremental splits. */
@Data
-public class IncrementalPhaseState implements PendingSplitsState {}
+public class IncrementalPhaseState implements PendingSplitsState {
+
+ // Preserve compatibility with checkpoints written when this state had no
fields.
+ private static final long serialVersionUID = -6809026812298443356L;
+
+ private final Offset startupOffset;
+
+ public IncrementalPhaseState() {
+ this(null);
+ }
+
+ public IncrementalPhaseState(Offset startupOffset) {
+ this.startupOffset = startupOffset;
+ }
+}
diff --git
a/seatunnel-connectors-v2/connector-cdc/connector-cdc-base/src/main/java/org/apache/seatunnel/connectors/cdc/base/source/offset/OffsetFactory.java
b/seatunnel-connectors-v2/connector-cdc/connector-cdc-base/src/main/java/org/apache/seatunnel/connectors/cdc/base/source/offset/OffsetFactory.java
index fc73afbf39..d8f183fcaf 100644
---
a/seatunnel-connectors-v2/connector-cdc/connector-cdc-base/src/main/java/org/apache/seatunnel/connectors/cdc/base/source/offset/OffsetFactory.java
+++
b/seatunnel-connectors-v2/connector-cdc/connector-cdc-base/src/main/java/org/apache/seatunnel/connectors/cdc/base/source/offset/OffsetFactory.java
@@ -29,6 +29,11 @@ public abstract class OffsetFactory implements Serializable {
public abstract Offset latest();
+ public Offset committedOffset() {
+ throw new UnsupportedOperationException(
+ "not supported create new Offset by committed offset.");
+ }
+
public abstract Offset specific(Map<String, String> offset);
public abstract Offset specific(String filename, Long position);
diff --git
a/seatunnel-connectors-v2/connector-cdc/connector-cdc-base/src/test/java/org/apache/seatunnel/connectors/cdc/base/source/enumerator/IncrementalSplitAssignerTest.java
b/seatunnel-connectors-v2/connector-cdc/connector-cdc-base/src/test/java/org/apache/seatunnel/connectors/cdc/base/source/enumerator/IncrementalSplitAssignerTest.java
new file mode 100644
index 0000000000..0229f3b663
--- /dev/null
+++
b/seatunnel-connectors-v2/connector-cdc/connector-cdc-base/src/test/java/org/apache/seatunnel/connectors/cdc/base/source/enumerator/IncrementalSplitAssignerTest.java
@@ -0,0 +1,80 @@
+/*
+ * 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.cdc.base.source.enumerator;
+
+import org.apache.seatunnel.connectors.cdc.base.config.SourceConfig;
+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.apache.seatunnel.connectors.cdc.base.source.enumerator.state.IncrementalPhaseState;
+import org.apache.seatunnel.connectors.cdc.base.source.offset.Offset;
+import org.apache.seatunnel.connectors.cdc.base.source.offset.OffsetFactory;
+
+import org.junit.jupiter.api.Test;
+
+import io.debezium.relational.TableId;
+
+import java.util.Collections;
+import java.util.HashMap;
+
+import static org.junit.jupiter.api.Assertions.assertSame;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+class IncrementalSplitAssignerTest {
+
+ @Test
+ void shouldReuseCheckpointedStartupOffsetAfterRestore() {
+ SourceConfig sourceConfig = mock(SourceConfig.class);
+ OffsetFactory offsetFactory = mock(OffsetFactory.class);
+ Offset committedOffset = mock(Offset.class);
+ Offset stoppingOffset = mock(Offset.class);
+ when(sourceConfig.getStartupConfig())
+ .thenReturn(new StartupConfig(StartupMode.COMMITTED_OFFSET,
null, null, null));
+ when(sourceConfig.getStopConfig())
+ .thenReturn(new StopConfig(StopMode.NEVER, null, null, null));
+ when(offsetFactory.committedOffset()).thenReturn(committedOffset);
+ when(offsetFactory.neverStop()).thenReturn(stoppingOffset);
+
+ SplitAssigner.Context<SourceConfig> context =
createContext(sourceConfig);
+ IncrementalSplitAssigner<SourceConfig> assigner =
+ new IncrementalSplitAssigner<>(context, 1, offsetFactory);
+ assertSame(
+ committedOffset,
assigner.getNext().get().asIncrementalSplit().getStartupOffset());
+
+ IncrementalPhaseState checkpoint = assigner.snapshotState(1L);
+ IncrementalSplitAssigner<SourceConfig> restoredAssigner =
+ new IncrementalSplitAssigner<>(
+ createContext(sourceConfig), 1, offsetFactory,
checkpoint);
+ assertSame(
+ committedOffset,
+
restoredAssigner.getNext().get().asIncrementalSplit().getStartupOffset());
+
+ verify(offsetFactory).committedOffset();
+ }
+
+ private SplitAssigner.Context<SourceConfig> createContext(SourceConfig
sourceConfig) {
+ return new SplitAssigner.Context<>(
+ sourceConfig,
+ Collections.singleton(TableId.parse("database.schema.table")),
+ new HashMap<>(),
+ new HashMap<>());
+ }
+}
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 dee5aef32d..83e7089128 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
@@ -19,6 +19,7 @@ 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 io.debezium.connector.postgresql.PostgresConnector;
@@ -110,6 +111,12 @@ public class PostgresSourceConfigFactory extends
JdbcSourceConfigFactory {
if (dbzProperties != null) {
props.putAll(dbzProperties);
}
+ 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(
diff --git
a/seatunnel-connectors-v2/connector-cdc/connector-cdc-postgres/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/postgres/exception/PostgresConnectorErrorCode.java
b/seatunnel-connectors-v2/connector-cdc/connector-cdc-postgres/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/postgres/exception/PostgresConnectorErrorCode.java
index 0efe50b0fe..a212c07217 100644
---
a/seatunnel-connectors-v2/connector-cdc/connector-cdc-postgres/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/postgres/exception/PostgresConnectorErrorCode.java
+++
b/seatunnel-connectors-v2/connector-cdc/connector-cdc-postgres/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/postgres/exception/PostgresConnectorErrorCode.java
@@ -21,7 +21,9 @@ import
org.apache.seatunnel.common.exception.SeaTunnelErrorCode;
public enum PostgresConnectorErrorCode implements SeaTunnelErrorCode {
NEW_SCHEMA_FAILED("POSTGRES-01", "Failed to initialize PostgresSchema"),
- CREATE_REPLICATION_CONNECTION_FAILED("POSTGRES-02", "Failed to create
replication connection");
+ CREATE_REPLICATION_CONNECTION_FAILED("POSTGRES-02", "Failed to create
replication connection"),
+ READ_COMMITTED_OFFSET_FAILED(
+ "POSTGRES-03", "Failed to read the committed replication slot
offset");
private final String code;
private final String description;
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 38074fae78..69469cf1e9 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
@@ -25,6 +25,7 @@ import org.apache.seatunnel.common.utils.JdbcUrlUtil;
import org.apache.seatunnel.common.utils.SeaTunnelException;
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.config.StartupConfig;
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.StartupMode;
@@ -83,6 +84,13 @@ public class PostgresIncrementalSource<T> extends
IncrementalSource<T, JdbcSourc
return PostgresSourceOptions.STOP_MODE;
}
+ @Override
+ protected StartupConfig getStartupConfig(ReadonlyConfig config) {
+ StartupConfig startupConfig = super.getStartupConfig(config);
+ validateStartupOptions(config, startupConfig);
+ return startupConfig;
+ }
+
@Override
public SourceConfig.Factory<JdbcSourceConfig>
createSourceConfigFactory(ReadonlyConfig config) {
PostgresSourceConfigFactory configFactory = new
PostgresSourceConfigFactory();
@@ -137,6 +145,23 @@ public class PostgresIncrementalSource<T> extends
IncrementalSource<T, JdbcSourc
return Optional.of("org.postgresql.Driver");
}
+ private void validateStartupOptions(ReadonlyConfig options, StartupConfig
startupConfig) {
+ if (startupConfig.getStartupMode() != StartupMode.COMMITTED_OFFSET) {
+ return;
+ }
+ Optional<String> slotName =
+ options.getOptional(PostgresIncrementalSourceOptions.SLOT_NAME)
+ .map(String::trim)
+ .filter(name -> !name.isEmpty());
+ if (!slotName.isPresent()) {
+ throw new SeaTunnelException(
+ String.format(
+ "PostgreSQL-CDC startup.mode '%s' requires an
explicit '%s' option.",
+ StartupMode.COMMITTED_OFFSET,
+ PostgresIncrementalSourceOptions.SLOT_NAME.key()));
+ }
+ }
+
private Map<TableId, Struct> tableChanges() {
JdbcSourceConfig jdbcSourceConfig = configFactory.create(0);
PostgresDialect dialect =
diff --git
a/seatunnel-connectors-v2/connector-cdc/connector-cdc-postgres/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/postgres/source/PostgresIncrementalSourceFactory.java
b/seatunnel-connectors-v2/connector-cdc/connector-cdc-postgres/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/postgres/source/PostgresIncrementalSourceFactory.java
index d4aefa4139..39e36c36ad 100644
---
a/seatunnel-connectors-v2/connector-cdc/connector-cdc-postgres/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/postgres/source/PostgresIncrementalSourceFactory.java
+++
b/seatunnel-connectors-v2/connector-cdc/connector-cdc-postgres/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/postgres/source/PostgresIncrementalSourceFactory.java
@@ -79,6 +79,10 @@ public class PostgresIncrementalSourceFactory implements
TableSourceFactory {
PostgresSourceOptions.STARTUP_MODE,
StartupMode.INITIAL,
JdbcSourceOptions.EXACTLY_ONCE)
+ .conditional(
+ PostgresSourceOptions.STARTUP_MODE,
+ StartupMode.SNAPSHOT_ONLY,
+ JdbcSourceOptions.EXACTLY_ONCE)
.build();
}
diff --git
a/seatunnel-connectors-v2/connector-cdc/connector-cdc-postgres/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/postgres/source/PostgresSourceOptions.java
b/seatunnel-connectors-v2/connector-cdc/connector-cdc-postgres/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/postgres/source/PostgresSourceOptions.java
index 91fda42d34..b238d16f53 100644
---
a/seatunnel-connectors-v2/connector-cdc/connector-cdc-postgres/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/postgres/source/PostgresSourceOptions.java
+++
b/seatunnel-connectors-v2/connector-cdc/connector-cdc-postgres/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/postgres/source/PostgresSourceOptions.java
@@ -33,12 +33,14 @@ public class PostgresSourceOptions {
StartupMode.class,
Arrays.asList(
StartupMode.INITIAL,
+ StartupMode.SNAPSHOT_ONLY,
+ StartupMode.COMMITTED_OFFSET,
StartupMode.EARLIEST,
StartupMode.LATEST))
.defaultValue(StartupMode.INITIAL)
.withDescription(
"Optional startup mode for CDC source,
valid enumerations are "
- + "\"initial\", \"earliest\",
\"latest\"");
+ + "\"initial\", \"snapshot-only\",
\"committed-offset\", \"earliest\", \"latest\"");
public static final SingleChoiceOption<StopMode> STOP_MODE =
(SingleChoiceOption)
diff --git
a/seatunnel-connectors-v2/connector-cdc/connector-cdc-postgres/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/postgres/source/offset/LsnOffsetFactory.java
b/seatunnel-connectors-v2/connector-cdc/connector-cdc-postgres/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/postgres/source/offset/LsnOffsetFactory.java
index fe01bc9e80..d6ece75d15 100644
---
a/seatunnel-connectors-v2/connector-cdc/connector-cdc-postgres/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/postgres/source/offset/LsnOffsetFactory.java
+++
b/seatunnel-connectors-v2/connector-cdc/connector-cdc-postgres/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/postgres/source/offset/LsnOffsetFactory.java
@@ -17,16 +17,27 @@
package org.apache.seatunnel.connectors.seatunnel.cdc.postgres.source.offset;
+import org.apache.seatunnel.common.exception.SeaTunnelRuntimeException;
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.seatunnel.cdc.postgres.config.PostgresSourceConfig;
import
org.apache.seatunnel.connectors.seatunnel.cdc.postgres.config.PostgresSourceConfigFactory;
+import
org.apache.seatunnel.connectors.seatunnel.cdc.postgres.exception.PostgresConnectorErrorCode;
import
org.apache.seatunnel.connectors.seatunnel.cdc.postgres.source.PostgresDialect;
import
org.apache.seatunnel.connectors.seatunnel.cdc.postgres.utils.PostgresUtils;
+import io.debezium.connector.postgresql.PostgresOffsetContext;
+import io.debezium.connector.postgresql.SourceInfo;
+import io.debezium.connector.postgresql.connection.Lsn;
import io.debezium.connector.postgresql.connection.PostgresConnection;
import io.debezium.jdbc.JdbcConnection;
+import io.debezium.time.Conversions;
+import java.sql.PreparedStatement;
+import java.sql.ResultSet;
+import java.sql.SQLException;
+import java.time.Instant;
+import java.util.HashMap;
import java.util.Map;
public class LsnOffsetFactory extends OffsetFactory {
@@ -59,6 +70,73 @@ public class LsnOffsetFactory extends OffsetFactory {
}
}
+ @Override
+ public Offset committedOffset() {
+ String slotName =
sourceConfig.getDbzConfiguration().getString("slot.name");
+ try (JdbcConnection jdbcConnection =
dialect.openJdbcConnection(sourceConfig)) {
+ try (PreparedStatement statement =
+ jdbcConnection
+ .connection()
+ .prepareStatement(
+ "SELECT confirmed_flush_lsn::text,
active_pid FROM pg_replication_slots WHERE slot_name = ?")) {
+ statement.setString(1, slotName);
+ try (ResultSet resultSet = statement.executeQuery()) {
+ return readCommittedOffset(resultSet, slotName);
+ }
+ }
+ } catch (SeaTunnelRuntimeException e) {
+ throw e;
+ } catch (Exception e) {
+ throw new SeaTunnelRuntimeException(
+ PostgresConnectorErrorCode.READ_COMMITTED_OFFSET_FAILED,
+ String.format(
+ "Could not read PostgreSQL replication slot '%s'.
Verify that the slot still exists and is not being dropped.",
+ slotName),
+ e);
+ }
+ }
+
+ static LsnOffset readCommittedOffset(ResultSet resultSet, String slotName)
throws SQLException {
+ if (!resultSet.next()) {
+ throw invalidReplicationSlot(
+ slotName, "does not exist; create the logical replication
slot first");
+ }
+ Object activePid = resultSet.getObject(2);
+ if (activePid != null) {
+ throw invalidReplicationSlot(
+ slotName,
+ String.format(
+ "is already active in PostgreSQL backend PID %s;
stop the other consumer first",
+ activePid));
+ }
+ String committedLsn = resultSet.getString(1);
+ if (resultSet.wasNull() || committedLsn == null ||
committedLsn.trim().isEmpty()) {
+ throw invalidReplicationSlot(
+ slotName,
+ "has no usable confirmed_flush_lsn; wait until the slot
has committed an LSN");
+ }
+ return toLsnOffset(committedLsn.trim());
+ }
+
+ private static SeaTunnelRuntimeException invalidReplicationSlot(
+ String slotName, String reason) {
+ return new SeaTunnelRuntimeException(
+ PostgresConnectorErrorCode.READ_COMMITTED_OFFSET_FAILED,
+ String.format("PostgreSQL replication slot '%s' %s.",
slotName, reason));
+ }
+
+ static LsnOffset toLsnOffset(String committedLsn) {
+ Map<String, String> offsetMap = new HashMap<>();
+ String lsn = String.valueOf(Lsn.valueOf(committedLsn).asLong());
+ offsetMap.put(SourceInfo.LSN_KEY, lsn);
+ offsetMap.put(PostgresOffsetContext.LAST_COMPLETELY_PROCESSED_LSN_KEY,
lsn);
+ offsetMap.put(PostgresOffsetContext.LAST_COMMIT_LSN_KEY, lsn);
+ offsetMap.put(
+ SourceInfo.TIMESTAMP_USEC_KEY,
+ String.valueOf(Conversions.toEpochMicros(Instant.MIN)));
+ return LsnOffset.of(offsetMap);
+ }
+
@Override
public Offset specific(Map<String, String> offset) {
return new LsnOffset(offset);
diff --git
a/seatunnel-connectors-v2/connector-cdc/connector-cdc-postgres/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/postgres/source/reader/PostgresSourceFetchTaskContext.java
b/seatunnel-connectors-v2/connector-cdc/connector-cdc-postgres/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/postgres/source/reader/PostgresSourceFetchTaskContext.java
index 058343cc72..d3328f1718 100644
---
a/seatunnel-connectors-v2/connector-cdc/connector-cdc-postgres/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/postgres/source/reader/PostgresSourceFetchTaskContext.java
+++
b/seatunnel-connectors-v2/connector-cdc/connector-cdc-postgres/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/postgres/source/reader/PostgresSourceFetchTaskContext.java
@@ -205,21 +205,21 @@ public class PostgresSourceFetchTaskContext extends
JdbcSourceFetchTaskContext {
dataConnection,
snapshotter.shouldSnapshot(),
connectorConfig);
- try {
- // create the slot if it doesn't exist, otherwise
update slot to add new
- // table(job restore and add table)
-
replicationConnection.createReplicationSlot().orElse(null);
- } catch (SQLException ex) {
- String message = "Creation of replication slot failed";
- // PostgreSQL errors all have a 5-character SQLSTATE
code, following the SQL
- // standard specification
- //
https://www.postgresql.org/docs/current/errcodes-appendix.html
- if ("42710".equals(ex.getSQLState())) {
- message +=
- "; when setting up multiple connectors for
the same database host, please make sure to use a distinct replication slot
name for each.";
- log.warn(message);
- } else {
- throw new DebeziumException(message, ex);
+ if (slotInfo == null) {
+ try {
+
replicationConnection.createReplicationSlot().orElse(null);
+ } catch (SQLException ex) {
+ String message = "Creation of replication slot
failed";
+ // PostgreSQL errors all have a 5-character
SQLSTATE code, following the
+ // SQL standard specification
+ //
https://www.postgresql.org/docs/current/errcodes-appendix.html
+ if ("42710".equals(ex.getSQLState())) {
+ message +=
+ "; when setting up multiple connectors
for the same database host, please make sure to use a distinct replication slot
name for each.";
+ log.warn(message);
+ } else {
+ throw new DebeziumException(message, ex);
+ }
}
}
}
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
new file mode 100644
index 0000000000..67324c0a2e
--- /dev/null
+++
b/seatunnel-connectors-v2/connector-cdc/connector-cdc-postgres/src/test/java/org/apache/seatunnel/connectors/seatunnel/cdc/postgres/config/PostgresSourceConfigFactoryTest.java
@@ -0,0 +1,44 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.seatunnel.connectors.seatunnel.cdc.postgres.config;
+
+import org.apache.seatunnel.connectors.cdc.base.config.StartupConfig;
+import org.apache.seatunnel.connectors.cdc.base.option.StartupMode;
+
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+public class PostgresSourceConfigFactoryTest {
+
+ @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"));
+ }
+}
diff --git
a/seatunnel-connectors-v2/connector-cdc/connector-cdc-postgres/src/test/java/org/apache/seatunnel/connectors/seatunnel/cdc/postgres/source/offset/LsnOffsetFactoryTest.java
b/seatunnel-connectors-v2/connector-cdc/connector-cdc-postgres/src/test/java/org/apache/seatunnel/connectors/seatunnel/cdc/postgres/source/offset/LsnOffsetFactoryTest.java
new file mode 100644
index 0000000000..ce301223e3
--- /dev/null
+++
b/seatunnel-connectors-v2/connector-cdc/connector-cdc-postgres/src/test/java/org/apache/seatunnel/connectors/seatunnel/cdc/postgres/source/offset/LsnOffsetFactoryTest.java
@@ -0,0 +1,109 @@
+/*
+ * 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.offset;
+
+import org.apache.seatunnel.common.exception.SeaTunnelRuntimeException;
+import org.apache.seatunnel.common.utils.SerializationUtils;
+
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+import io.debezium.connector.postgresql.PostgresOffsetContext;
+import io.debezium.connector.postgresql.SourceInfo;
+import io.debezium.connector.postgresql.connection.Lsn;
+
+import java.sql.ResultSet;
+import java.util.Map;
+
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+public class LsnOffsetFactoryTest {
+
+ @Test
+ public void shouldCreateRestartableOffsetFromCommittedLsn() {
+ LsnOffset offset = LsnOffsetFactory.toLsnOffset("0/16B6C50");
+ Map<String, String> offsetMap = offset.getOffset();
+
+ Assertions.assertEquals(
+ String.valueOf(Lsn.valueOf("0/16B6C50").asLong()),
+ offsetMap.get(SourceInfo.LSN_KEY));
+ Assertions.assertEquals(
+ offsetMap.get(SourceInfo.LSN_KEY),
+
offsetMap.get(PostgresOffsetContext.LAST_COMPLETELY_PROCESSED_LSN_KEY));
+ Assertions.assertEquals(
+ offsetMap.get(SourceInfo.LSN_KEY),
+ offsetMap.get(PostgresOffsetContext.LAST_COMMIT_LSN_KEY));
+ }
+
+ @Test
+ public void shouldRoundTripSerializedLsnOffset() {
+ LsnOffset offset = LsnOffsetFactory.toLsnOffset("0/16B6C50");
+
+ LsnOffset restored =
SerializationUtils.deserialize(SerializationUtils.serialize(offset));
+
+ Assertions.assertEquals(offset, restored);
+ Assertions.assertEquals(offset.getOffset(), restored.getOffset());
+ }
+
+ @Test
+ public void shouldRejectReplicationSlotWithNullCommittedLsn() throws
Exception {
+ ResultSet resultSet = mock(ResultSet.class);
+ when(resultSet.next()).thenReturn(true);
+ when(resultSet.getString(1)).thenReturn(null);
+ when(resultSet.wasNull()).thenReturn(true);
+
+ SeaTunnelRuntimeException exception =
+ Assertions.assertThrows(
+ SeaTunnelRuntimeException.class,
+ () -> LsnOffsetFactory.readCommittedOffset(resultSet,
"empty_slot"));
+
+ Assertions.assertTrue(exception.getMessage().contains("empty_slot"));
+
Assertions.assertTrue(exception.getMessage().contains("confirmed_flush_lsn"));
+ }
+
+ @Test
+ public void shouldRejectMissingReplicationSlot() throws Exception {
+ ResultSet resultSet = mock(ResultSet.class);
+ when(resultSet.next()).thenReturn(false);
+
+ SeaTunnelRuntimeException exception =
+ Assertions.assertThrows(
+ SeaTunnelRuntimeException.class,
+ () -> LsnOffsetFactory.readCommittedOffset(resultSet,
"missing_slot"));
+
+ Assertions.assertTrue(exception.getMessage().contains("missing_slot"));
+ Assertions.assertTrue(exception.getMessage().contains("does not
exist"));
+ }
+
+ @Test
+ public void shouldRejectActiveReplicationSlot() throws Exception {
+ ResultSet resultSet = mock(ResultSet.class);
+ when(resultSet.next()).thenReturn(true);
+ when(resultSet.getObject(2)).thenReturn(1234);
+
+ SeaTunnelRuntimeException exception =
+ Assertions.assertThrows(
+ SeaTunnelRuntimeException.class,
+ () -> LsnOffsetFactory.readCommittedOffset(resultSet,
"active_slot"));
+
+ Assertions.assertTrue(exception.getMessage().contains("active_slot"));
+ Assertions.assertTrue(exception.getMessage().contains("1234"));
+ Assertions.assertTrue(exception.getMessage().contains("other
consumer"));
+ }
+}
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 1582616e52..c0275af2cc 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
@@ -61,6 +61,7 @@ import org.testcontainers.utility.DockerImageName;
import org.testcontainers.utility.DockerLoggerFactory;
import org.testcontainers.utility.MountableFile;
+import io.debezium.connector.postgresql.connection.Lsn;
import io.debezium.jdbc.JdbcConnection;
import io.debezium.relational.TableId;
import lombok.extern.slf4j.Slf4j;
@@ -86,6 +87,7 @@ import java.util.Properties;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicReference;
import java.util.regex.Matcher;
import java.util.regex.Pattern;
@@ -428,6 +430,246 @@ public class PostgresCDCIT extends TestSuiteBase
implements TestResource {
}
}
+ @TestTemplate
+ @DisabledOnContainer(
+ value = {},
+ type = {EngineType.SPARK, EngineType.FLINK},
+ disabledReason =
+ "This case validates bounded completion and manually
cancels a streaming job, which is currently only supported by the zeta engine.")
+ public void
testPostgresCdcSnapshotOnlyAndCommittedOffsetStartupModes(TestContainer
container)
+ throws IOException, InterruptedException, ExecutionException,
TimeoutException {
+ String snapshotSlotName = createSlotName();
+ String snapshotSlotVariable = toSlotVariable(snapshotSlotName);
+ String committedSlotName = createSlotName();
+ String committedSlotVariable = toSlotVariable(committedSlotName);
+ Long seedJobId = JobIdGenerator.newJobId();
+ Long committedOffsetJobId = JobIdGenerator.newJobId();
+ CompletableFuture<Void> seedJob = null;
+ CompletableFuture<Void> committedOffsetJob = null;
+
+ try {
+ clearTable(POSTGRESQL_SCHEMA, SOURCE_TABLE_1);
+ clearTable(POSTGRESQL_SCHEMA, SINK_TABLE_1);
+ insertSourceTableRow(POSTGRESQL_SCHEMA, SOURCE_TABLE_1, 10);
+
+ container.executeJob(
+ "/postgrescdc_to_postgres_snapshot_only.conf",
+ Collections.singletonList(snapshotSlotVariable));
+ Assertions.assertIterableEquals(
+ query("select * from " + POSTGRESQL_SCHEMA + "." +
SOURCE_TABLE_1),
+ query("select * from " + POSTGRESQL_SCHEMA + "." +
SINK_TABLE_1));
+
+ List<List<Object>> snapshotOnlySinkRows =
+ query("select * from " + POSTGRESQL_SCHEMA + "." +
SINK_TABLE_1);
+ insertSourceTableRow(POSTGRESQL_SCHEMA, SOURCE_TABLE_1, 11);
+ TimeUnit.SECONDS.sleep(5);
+ Assertions.assertIterableEquals(
+ snapshotOnlySinkRows,
+ query("select * from " + POSTGRESQL_SCHEMA + "." +
SINK_TABLE_1));
+ Assertions.assertFalse(
+ replicationSlotExists(snapshotSlotName),
+ "Snapshot-only startup must not create or retain a
replication slot");
+
+ clearTable(POSTGRESQL_SCHEMA, SINK_TABLE_1);
+ createLogicalReplicationSlot(committedSlotName);
+
+ seedJob =
+ CompletableFuture.runAsync(
+ () -> {
+ try {
+ container.executeJob(
+
"/postgrescdc_to_postgres_committed_offset.conf",
+ String.valueOf(seedJobId),
+ committedSlotVariable);
+ } catch (Exception e) {
+ log.error("Seed task exception :" +
e.getMessage());
+ throw new RuntimeException(e);
+ }
+ });
+ CompletableFuture<Void> runningSeedJob = seedJob;
+ waitForReplicationSlotActive(committedSlotName);
+ insertSourceTableRow(POSTGRESQL_SCHEMA, SOURCE_TABLE_1, 12);
+ await().pollInterval(1, TimeUnit.SECONDS)
+ .atMost(60000, TimeUnit.MILLISECONDS)
+ .untilAsserted(
+ () -> {
+ assertJobHasNoAsyncFailure(runningSeedJob);
+ List<List<Object>> seedRows =
+ query(
+ "select * from "
+ + POSTGRESQL_SCHEMA
+ + "."
+ + SINK_TABLE_1
+ + " where id = 12");
+ if (seedRows.isEmpty()) {
+ updateSourceTableBigField(
+ POSTGRESQL_SCHEMA, SOURCE_TABLE_1,
12, 2147483646);
+ }
+ Assertions.assertEquals(1, seedRows.size());
+ });
+ String postSeedLsn = getCurrentWalLsn();
+ await().atMost(30000, TimeUnit.MILLISECONDS)
+ .untilAsserted(
+ () ->
+ Assertions.assertTrue(
+ isLsnGreaterThanOrEqual(
+
getReplicationSlotCommittedLsn(
+ committedSlotName),
+ postSeedLsn)));
+ container.cancelJob(String.valueOf(seedJobId));
+ seedJob.get(30, TimeUnit.SECONDS);
+ await().atMost(30000, TimeUnit.MILLISECONDS)
+ .untilAsserted(
+ () ->
+ Assertions.assertFalse(
+
isReplicationSlotActive(committedSlotName)));
+
+ clearTable(POSTGRESQL_SCHEMA, SINK_TABLE_1);
+ insertSourceTableRow(POSTGRESQL_SCHEMA, SOURCE_TABLE_1, 13);
+
+ committedOffsetJob =
+ CompletableFuture.runAsync(
+ () -> {
+ try {
+ container.executeJob(
+
"/postgrescdc_to_postgres_committed_offset.conf",
+
String.valueOf(committedOffsetJobId),
+ committedSlotVariable);
+ } catch (Exception e) {
+ log.error("Commit task exception :" +
e.getMessage());
+ throw new RuntimeException(e);
+ }
+ });
+
+ CompletableFuture<Void> runningCommittedOffsetJob =
committedOffsetJob;
+ await().atMost(60000, TimeUnit.MILLISECONDS)
+ .untilAsserted(
+ () -> {
+
assertJobHasNoAsyncFailure(runningCommittedOffsetJob);
+ Assertions.assertEquals(
+ 1,
+ query(
+ "select * from "
+ +
POSTGRESQL_SCHEMA
+ + "."
+ + SINK_TABLE_1
+ + " where id =
13")
+ .size());
+ Assertions.assertTrue(
+ query(
+ "select * from "
+ +
POSTGRESQL_SCHEMA
+ + "."
+ + SINK_TABLE_1
+ + " where id =
12")
+ .isEmpty());
+ });
+
+ insertSourceTableRow(POSTGRESQL_SCHEMA, SOURCE_TABLE_1, 14);
+ await().atMost(60000, TimeUnit.MILLISECONDS)
+ .untilAsserted(
+ () -> {
+
assertJobHasNoAsyncFailure(runningCommittedOffsetJob);
+ List<List<Object>> insertedRows =
+ query(
+ "select f_big from "
+ + POSTGRESQL_SCHEMA
+ + "."
+ + SINK_TABLE_1
+ + " where id = 14");
+ if
(!Collections.singletonList(Collections.singletonList(10000L))
+ .equals(insertedRows)) {
+ updateSourceTableBigField(
+ POSTGRESQL_SCHEMA, SOURCE_TABLE_1,
14, 10000);
+ }
+ Assertions.assertEquals(
+ Collections.singletonList(
+
Collections.singletonList(10000L)),
+ insertedRows);
+ });
+ deleteSourceTableRow(POSTGRESQL_SCHEMA, SOURCE_TABLE_1, 13);
+
+ await().atMost(60000, TimeUnit.MILLISECONDS)
+ .untilAsserted(
+ () -> {
+
assertJobHasNoAsyncFailure(runningCommittedOffsetJob);
+ Assertions.assertTrue(
+ query(
+ "select * from "
+ +
POSTGRESQL_SCHEMA
+ + "."
+ + SINK_TABLE_1
+ + " where id
in (10, 11, 12, 13)")
+ .isEmpty());
+ Assertions.assertEquals(
+ Collections.singletonList(
+
Collections.singletonList(10000L)),
+ query(
+ "select f_big from "
+ + POSTGRESQL_SCHEMA
+ + "."
+ + SINK_TABLE_1
+ + " where id = 14"));
+ });
+
+ Assertions.assertEquals(
+ 0,
container.savepointJob(String.valueOf(committedOffsetJobId)).getExitCode());
+ committedOffsetJob.get(30, TimeUnit.SECONDS);
+ insertSourceTableRow(POSTGRESQL_SCHEMA, SOURCE_TABLE_1, 15);
+
+ committedOffsetJob =
+ CompletableFuture.runAsync(
+ () -> {
+ try {
+ container.restoreJob(
+
"/postgrescdc_to_postgres_committed_offset.conf",
+
String.valueOf(committedOffsetJobId),
+ committedSlotVariable);
+ } catch (Exception e) {
+ log.error("Restore committed-offset task
exception", e);
+ throw new RuntimeException(e);
+ }
+ });
+ CompletableFuture<Void> restoredCommittedOffsetJob =
committedOffsetJob;
+ await().atMost(60000, TimeUnit.MILLISECONDS)
+ .untilAsserted(
+ () -> {
+
assertJobHasNoAsyncFailure(restoredCommittedOffsetJob);
+ Assertions.assertEquals(
+ 1,
+ query(
+ "select * from "
+ +
POSTGRESQL_SCHEMA
+ + "."
+ + SINK_TABLE_1
+ + " where id =
15")
+ .size());
+ });
+ } finally {
+ try {
+ try {
+ container.cancelJob(String.valueOf(seedJobId));
+ } catch (Exception e) {
+ log.warn("Failed to cancel committed-offset seed job", e);
+ }
+ try {
+ container.cancelJob(String.valueOf(committedOffsetJobId));
+ } catch (Exception e) {
+ log.warn("Failed to cancel committed-offset test job", e);
+ }
+ if (seedJob != null && !seedJob.isDone()) {
+ seedJob.get(30, TimeUnit.SECONDS);
+ }
+ if (committedOffsetJob != null) {
+ committedOffsetJob.get(30, TimeUnit.SECONDS);
+ }
+ } finally {
+ clearTable(POSTGRESQL_SCHEMA, SOURCE_TABLE_1);
+ clearTable(POSTGRESQL_SCHEMA, SINK_TABLE_1);
+ }
+ }
+ }
+
@TestTemplate
@DisabledOnContainer(
value = {},
@@ -1070,6 +1312,73 @@ public class PostgresCDCIT extends TestSuiteBase
implements TestResource {
}
}
+ private boolean replicationSlotExists(String slotName) {
+ try (Connection connection = getJdbcConnection();
+ Statement statement = connection.createStatement();
+ ResultSet resultSet =
+ statement.executeQuery(
+ "SELECT 1 FROM pg_replication_slots WHERE
slot_name = '"
+ + slotName
+ + "'")) {
+ return resultSet.next();
+ } catch (SQLException e) {
+ throw new RuntimeException("Failed to query replication slot: " +
slotName, e);
+ }
+ }
+
+ private void assertJobHasNoAsyncFailure(CompletableFuture<Void> future) {
+ if (future.isDone()) {
+ future.join();
+ }
+ }
+
+ private void createLogicalReplicationSlot(String slotName) {
+ try (Connection connection = getJdbcConnection();
+ Statement statement = connection.createStatement()) {
+ statement.execute(
+ "SELECT * FROM pg_create_logical_replication_slot('"
+ + slotName
+ + "', 'decoderbufs')");
+ } catch (SQLException e) {
+ throw new RuntimeException("Failed to create replication slot: " +
slotName, e);
+ }
+ }
+
+ private String getReplicationSlotCommittedLsn(String slotName) {
+ try (Connection connection = getJdbcConnection();
+ Statement statement = connection.createStatement();
+ ResultSet resultSet =
+ statement.executeQuery(
+ "SELECT confirmed_flush_lsn::text FROM
pg_replication_slots WHERE slot_name = '"
+ + slotName
+ + "'")) {
+ if (!resultSet.next()) {
+ throw new IllegalStateException("Replication slot does not
exist: " + slotName);
+ }
+ return resultSet.getString(1);
+ } catch (SQLException e) {
+ throw new RuntimeException(
+ "Failed to query committed LSN for replication slot: " +
slotName, e);
+ }
+ }
+
+ private String getCurrentWalLsn() {
+ try (Connection connection = getJdbcConnection();
+ Statement statement = connection.createStatement();
+ ResultSet resultSet = statement.executeQuery("SELECT
pg_current_wal_lsn()::text")) {
+ if (!resultSet.next()) {
+ throw new IllegalStateException("Failed to query current WAL
LSN");
+ }
+ return resultSet.getString(1);
+ } catch (SQLException e) {
+ throw new RuntimeException("Failed to query current WAL LSN", e);
+ }
+ }
+
+ private boolean isLsnGreaterThanOrEqual(String actualLsn, String
expectedLsn) {
+ return Lsn.valueOf(actualLsn).compareTo(Lsn.valueOf(expectedLsn)) >= 0;
+ }
+
private void dropReplicationSlot(String slotName) {
try (Connection connection = getJdbcConnection();
Statement statement = connection.createStatement();
diff --git
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-postgres-e2e/src/test/resources/postgrescdc_to_postgres_committed_offset.conf
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-postgres-e2e/src/test/resources/postgrescdc_to_postgres_committed_offset.conf
new file mode 100644
index 0000000000..b58e8d156b
--- /dev/null
+++
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-postgres-e2e/src/test/resources/postgrescdc_to_postgres_committed_offset.conf
@@ -0,0 +1,53 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+env {
+ execution.parallelism = 1
+ job.mode = "STREAMING"
+ checkpoint.interval = 5000
+ read_limit.bytes_per_second=7000000
+ read_limit.rows_per_second=400
+}
+
+source {
+ Postgres-CDC {
+ plugin_output = "customers_postgres_cdc"
+ username = "postgres"
+ password = "postgres"
+ database-names = ["postgres_cdc"]
+ schema-names = ["inventory"]
+ table-names = ["postgres_cdc.inventory.postgres_cdc_table_1"]
+ url =
"jdbc:postgresql://postgres_cdc_e2e:5432/postgres_cdc?loggerLevel=OFF"
+ decoding.plugin.name = "decoderbufs"
+ slot.name = ${slot_name}
+ startup.mode = "committed-offset"
+ }
+}
+
+sink {
+ jdbc {
+ plugin_input = "customers_postgres_cdc"
+ url =
"jdbc:postgresql://postgres_cdc_e2e:5432/postgres_cdc?loggerLevel=OFF"
+ driver = "org.postgresql.Driver"
+ user = "postgres"
+ password = "postgres"
+ generate_sink_sql = true
+ database = postgres_cdc
+ table = inventory.sink_postgres_cdc_table_1
+ primary_keys = ["id"]
+ }
+}
diff --git
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-postgres-e2e/src/test/resources/postgrescdc_to_postgres_snapshot_only.conf
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-postgres-e2e/src/test/resources/postgrescdc_to_postgres_snapshot_only.conf
new file mode 100644
index 0000000000..d3f840b7c9
--- /dev/null
+++
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-postgres-e2e/src/test/resources/postgrescdc_to_postgres_snapshot_only.conf
@@ -0,0 +1,51 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+env {
+ execution.parallelism = 1
+ job.mode = "BATCH"
+ checkpoint.interval = 5000
+}
+
+source {
+ Postgres-CDC {
+ plugin_output = "customers_postgres_cdc"
+ username = "postgres"
+ password = "postgres"
+ database-names = ["postgres_cdc"]
+ schema-names = ["inventory"]
+ table-names = ["postgres_cdc.inventory.postgres_cdc_table_1"]
+ url =
"jdbc:postgresql://postgres_cdc_e2e:5432/postgres_cdc?loggerLevel=OFF"
+ decoding.plugin.name = "decoderbufs"
+ slot.name = ${slot_name}
+ startup.mode = "snapshot-only"
+ }
+}
+
+sink {
+ jdbc {
+ plugin_input = "customers_postgres_cdc"
+ url =
"jdbc:postgresql://postgres_cdc_e2e:5432/postgres_cdc?loggerLevel=OFF"
+ driver = "org.postgresql.Driver"
+ user = "postgres"
+ password = "postgres"
+ generate_sink_sql = true
+ database = postgres_cdc
+ table = inventory.sink_postgres_cdc_table_1
+ primary_keys = ["id"]
+ }
+}
