Skip to content

[Feature][Connector-V2][Oracle-CDC] Add SCN specific startup #11171

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
goutamadwant wants to merge 4 commits into
base: dev
Choose a base branch
from
Open

[Feature][Connector-V2][Oracle-CDC] Add SCN specific startup #11171

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
f0cdeb6
[Feature][Connector-V2][Oracle-CDC] Add SCN specific startup
goutamadwant Jun 23, 2026
2ddc1bc
[Fix][Connector-V2][Oracle-CDC] Use structured SCN startup offset
goutamadwant Jun 29, 2026
99e5602
Merge branch 'dev' into GH-11041
goutamadwant Jul 2, 2026
14c604f
Merge remote-tracking branch 'origin/dev' into GH-11041
goutamadwant Jul 5, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
3 changes: 2 additions & 1 deletion docs/en/connectors/source/Oracle-CDC.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -232,8 +232,9 @@ exit;
| table-names | List | Conditionally required | - | Table names to monitor. Each value should use `database.schema.table`, for example: `ORCLCDB.DEBEZIUM.FULL_TYPES`. Configure either `table-names` or `table-pattern`. |
| table-pattern | String | Conditionally required | - | Regular expression for table names to capture. Configure either `table-names` or `table-pattern`. |
| table-names-config | List | No | - | Per-table config list. For example: `[{"table": "ORCLCDB.DEBEZIUM.FULL_TYPES","primaryKeys": ["ID"],"snapshotSplitColumn": "ID"}]`. Use this when the table has no primary key, needs a custom primary key, or needs an explicit snapshot split column. |
| startup.mode | Enum | No | INITIAL | Optional startup mode for Oracle CDC consumer, valid enumerations are `initial`, `latest` and `timestamp`. <br/> `initial`: Synchronize historical data at startup, and then synchronize incremental data.<br/> `latest`: Start from the latest offset and skip the initial snapshot.<br/> `timestamp`: Start from the SCN resolved from `startup.timestamp`. |
| startup.mode | Enum | No | INITIAL | Optional startup mode for Oracle CDC consumer, valid enumerations are `initial`, `latest`, `timestamp` and `specific`. <br/> `initial`: Synchronize historical data at startup, and then synchronize incremental data.<br/> `latest`: Start from the latest offset and skip the initial snapshot.<br/> `timestamp`: Start from the SCN resolved from `startup.timestamp`.<br/> `specific`: Start from user-supplied SCN. |
| startup.timestamp | Long | No | - | Start from the specified timestamp (milliseconds since Unix epoch). This timestamp is converted with `server-time-zone` when `startup.mode = timestamp`. **Note, This option is required when the `startup.mode` option used `timestamp`.** |
| startup.specific-offset.scn | Long | No | - | Start from the specified Oracle SCN. **Note, This option is required when the `startup.mode` option uses `specific`. The SCN must still be available to the selected Oracle log mining backend.** |
| stop.mode | Enum | No | NEVER | Optional stop mode for Oracle CDC consumer. The only valid value is `never`, so a streaming Oracle CDC source keeps running until the job is stopped. |
| 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. |
Expand Down
3 changes: 2 additions & 1 deletion docs/zh/connectors/source/Oracle-CDC.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -231,8 +231,9 @@ exit;
| table-names | List | 条件必填 | - | 要监控的数据库表名，建议使用 `database.schema.table` 格式，例如：`ORCLCDB.DEBEZIUM.FULL_TYPES`。`table-names` 和 `table-pattern` 二选一配置。 |
| table-pattern | String | 条件必填 | - | 要捕获的表名正则表达式。`table-names` 和 `table-pattern` 二选一配置。 |
| table-names-config | List | 否 | - | 按表单独配置。例如：`[{"table": "ORCLCDB.DEBEZIUM.FULL_TYPES","primaryKeys": ["ID"],"snapshotSplitColumn": "ID"}]`。当表没有主键、需要自定义主键，或需要指定快照拆分列时使用。 |
| startup.mode | Enum | 否 | INITIAL | Oracle CDC 使用者的可选启动模式，有效枚举值为 `initial`、`latest` 和 `timestamp`。<br/> `initial`：启动时同步历史数据，然后同步增量数据。<br/> `latest`：从最新偏移量启动，并跳过初始快照。<br/> `timestamp`：从 `startup.timestamp` 解析出的 SCN 启动。 |
| startup.mode | Enum | 否 | INITIAL | Oracle CDC 使用者的可选启动模式，有效枚举值为 `initial`、`latest`、`timestamp` 和 `specific`。<br/> `initial`：启动时同步历史数据，然后同步增量数据。<br/> `latest`：从最新偏移量启动，并跳过初始快照。<br/> `timestamp`：从 `startup.timestamp` 解析出的 SCN 启动。<br/> `specific`：从用户提供的 SCN 启动。 |
| startup.timestamp | Long | 否 | - | 从指定的时间戳（自 Unix 纪元以来的毫秒数）启动。当 `startup.mode = timestamp` 时，该时间戳会按 `server-time-zone` 转换。**注意，当 `startup.mode` 选项使用 `timestamp` 时，此选项是必需的。** |
| startup.specific-offset.scn | Long | 否 | - | 从指定的 Oracle SCN 启动。**注意，当 `startup.mode` 选项使用 `specific` 时，此选项是必需的。该 SCN 必须仍可被所选 Oracle 日志挖掘后端读取。** |
| stop.mode | Enum | 否 | NEVER | Oracle CDC 使用者的可选停止模式。当前唯一有效值是 `never`，因此流式 Oracle CDC source 会一直运行，直到任务被停止。 |
| snapshot.split.size | Integer | 否 | 8096 | 表快照的拆分大小（行数），在读取表快照时，捕获的表将被拆分为多个拆分块。 |
| snapshot.fetch.size | Integer | 否 | 1024 | 读取表快照时每次轮询的最大获取大小。 |
Expand Down
36 changes: 34 additions & 2 deletions ...cdc-base/src/main/java/org/apache/seatunnel/connectors/cdc/base/config/StartupConfig.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -21,21 +21,50 @@
import org.apache.seatunnel.connectors.cdc.base.source.offset.Offset;
import org.apache.seatunnel.connectors.cdc.base.source.offset.OffsetFactory;

import lombok.AllArgsConstructor;
import lombok.EqualsAndHashCode;
import lombok.Getter;

import java.io.Serializable;
import java.util.HashMap;
import java.util.Map;
import java.util.Objects;

@AllArgsConstructor
@EqualsAndHashCode
public final class StartupConfig implements Serializable {
private static final long serialVersionUID = 1L;

@Getter private final StartupMode startupMode;
private final String specificOffsetFile;
private final Long specificOffsetPos;
private final Map<String, String> specificOffset;
@Getter private final Long timestamp;

public StartupConfig(
StartupMode startupMode,
String specificOffsetFile,
Long specificOffsetPos,
Long timestamp) {
this(startupMode, specificOffsetFile, specificOffsetPos, null, timestamp);
}

public static StartupConfig specificOffset(Map<String, String> specificOffset) {
return new StartupConfig(
StartupMode.SPECIFIC, null, null, Objects.requireNonNull(specificOffset), null);
}

private StartupConfig(
StartupMode startupMode,
String specificOffsetFile,
Long specificOffsetPos,
Map<String, String> specificOffset,
Long timestamp) {
this.startupMode = startupMode;
this.specificOffsetFile = specificOffsetFile;
this.specificOffsetPos = specificOffsetPos;
this.specificOffset = specificOffset == null ? null : new HashMap<>(specificOffset);
this.timestamp = timestamp;
}

public Offset getStartupOffset(OffsetFactory offsetFactory) {
switch (startupMode) {
case EARLIEST:
Expand All @@ -45,6 +74,9 @@ public Offset getStartupOffset(OffsetFactory offsetFactory) {
case INITIAL:
return null;
case SPECIFIC:
if (specificOffset != null) {
return offsetFactory.specific(new HashMap<>(specificOffset));
}
return offsetFactory.specific(specificOffsetFile, specificOffsetPos);
case TIMESTAMP:
return offsetFactory.timestamp(timestamp);
Expand Down
54 changes: 54 additions & 0 deletions .../org/apache/seatunnel/connectors/seatunnel/cdc/oracle/source/OracleIncrementalSource.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -26,6 +26,7 @@
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.SourceOptions;
Expand All @@ -40,6 +41,7 @@
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.oracle.config.OracleSourceConfigFactory;
import org.apache.seatunnel.connectors.seatunnel.cdc.oracle.source.offset.RedoLogOffset;
import org.apache.seatunnel.connectors.seatunnel.cdc.oracle.source.offset.RedoLogOffsetFactory;
import org.apache.seatunnel.connectors.seatunnel.jdbc.config.JdbcCommonOptions;

Expand All @@ -51,6 +53,7 @@

import java.time.ZoneId;
import java.util.Arrays;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Optional;
Expand Down Expand Up @@ -81,6 +84,57 @@ public Option<StopMode> getStopModeOption() {
return OracleIncrementalSourceOptions.STOP_MODE;
}

@Override
protected StartupConfig getStartupConfig(ReadonlyConfig config) {
return getOracleStartupConfig(config);
}

static StartupConfig getOracleStartupConfig(ReadonlyConfig config) {
StartupMode startupMode = config.get(OracleIncrementalSourceOptions.STARTUP_MODE);
Optional<Long> startupSpecificOffsetScn =
config.getOptional(OracleIncrementalSourceOptions.STARTUP_SPECIFIC_OFFSET_SCN);

if (startupMode != StartupMode.SPECIFIC) {
if (startupSpecificOffsetScn.isPresent()) {
throw new IllegalArgumentException(
OracleIncrementalSourceOptions.STARTUP_SPECIFIC_OFFSET_SCN.key()
+ " is only supported when startup.mode is specific.");
}
return new StartupConfig(
startupMode,
config.get(SourceOptions.STARTUP_SPECIFIC_OFFSET_FILE),
config.get(SourceOptions.STARTUP_SPECIFIC_OFFSET_POS),
config.get(SourceOptions.STARTUP_TIMESTAMP));
}

if (config.getOptional(SourceOptions.STARTUP_SPECIFIC_OFFSET_FILE).isPresent()
|| config.getOptional(SourceOptions.STARTUP_SPECIFIC_OFFSET_POS).isPresent()) {
throw new IllegalArgumentException(
"Oracle-CDC specific startup mode uses "
+ OracleIncrementalSourceOptions.STARTUP_SPECIFIC_OFFSET_SCN.key()
+ " instead of file or position offsets.");
}

Long scn =
startupSpecificOffsetScn.orElseThrow(
() ->
new IllegalArgumentException(
OracleIncrementalSourceOptions.STARTUP_SPECIFIC_OFFSET_SCN
.key()
+ " is required when startup.mode is specific."));
if (scn <= 0) {
throw new IllegalArgumentException(
OracleIncrementalSourceOptions.STARTUP_SPECIFIC_OFFSET_SCN.key()
+ " must be greater than 0.");
}

Map<String, String> specificOffset = new HashMap<>();
specificOffset.put(RedoLogOffset.SCN_KEY, String.valueOf(scn));
specificOffset.put(RedoLogOffset.COMMIT_SCN_KEY, "0");
specificOffset.put(RedoLogOffset.LCR_POSITION_KEY, null);
return StartupConfig.specificOffset(specificOffset);
}

@Override
public SourceConfig.Factory<JdbcSourceConfig> createSourceConfigFactory(ReadonlyConfig config) {
OracleSourceConfigFactory configFactory = new OracleSourceConfigFactory();
Expand Down
Loading
Loading