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 3ec61ba15d [Feature][Connector-V2][Oracle-CDC] Add SCN specific
startup (#11171)
3ec61ba15d is described below
commit 3ec61ba15dd8cea094dbf38a9b89f7dae7649cad
Author: Goutam Adwant <[email protected]>
AuthorDate: Thu Aug 13 21:13:42 2026 -0700
[Feature][Connector-V2][Oracle-CDC] Add SCN specific startup (#11171)
Signed-off-by: goutamadwant <[email protected]>
---
docs/en/connectors/source/Oracle-CDC.md | 3 +-
docs/zh/connectors/source/Oracle-CDC.md | 3 +-
.../cdc/oracle/source/OracleIncrementalSource.java | 63 +++++++++
.../source/OracleIncrementalSourceFactory.java | 15 +-
.../source/OracleIncrementalSourceOptions.java | 9 +-
.../source/OracleIncrementalSourceFactoryTest.java | 155 +++++++++++++++++++++
6 files changed, 236 insertions(+), 12 deletions(-)
diff --git a/docs/en/connectors/source/Oracle-CDC.md
b/docs/en/connectors/source/Oracle-CDC.md
index 0200a496ed..ea55214f28 100644
--- a/docs/en/connectors/source/Oracle-CDC.md
+++ b/docs/en/connectors/source/Oracle-CDC.md
@@ -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.
[...]
diff --git a/docs/zh/connectors/source/Oracle-CDC.md
b/docs/zh/connectors/source/Oracle-CDC.md
index 4d8646a084..213e6aece8 100644
--- a/docs/zh/connectors/source/Oracle-CDC.md
+++ b/docs/zh/connectors/source/Oracle-CDC.md
@@ -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 |
读取表快照时每次轮询的最大获取大小。
[...]
diff --git
a/seatunnel-connectors-v2/connector-cdc/connector-cdc-oracle/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/oracle/source/OracleIncrementalSource.java
b/seatunnel-connectors-v2/connector-cdc/connector-cdc-oracle/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/oracle/source/OracleIncrementalSource.java
index c2c9e860fa..733e6991df 100644
---
a/seatunnel-connectors-v2/connector-cdc/connector-cdc-oracle/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/oracle/source/OracleIncrementalSource.java
+++
b/seatunnel-connectors-v2/connector-cdc/connector-cdc-oracle/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/oracle/source/OracleIncrementalSource.java
@@ -26,6 +26,7 @@ import org.apache.seatunnel.api.table.schema.SchemaChangeType;
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;
@@ -40,6 +41,7 @@ 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.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;
@@ -51,6 +53,7 @@ import io.debezium.relational.history.TableChanges;
import java.time.ZoneId;
import java.util.Arrays;
+import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Optional;
@@ -81,6 +84,66 @@ public class OracleIncrementalSource<T> extends
IncrementalSource<T, JdbcSourceC
return OracleIncrementalSourceOptions.STOP_MODE;
}
+ @Override
+ protected StartupConfig getStartupConfig(ReadonlyConfig config) {
+ return getOracleStartupConfig(config);
+ }
+
+ /**
+ * Builds the startup configuration for Oracle CDC. A specific SCN is
mapped to the standard
+ * {@link RedoLogOffset} structure. Generic file and position offsets are
rejected because they
+ * do not apply to Oracle. The initial {@code commit_scn=0} and {@code
lcr_position=null} values
+ * intentionally match the offset shape expected by the existing Oracle
offset loader.
+ *
+ * @param config connector configuration
+ * @return the Oracle startup configuration
+ */
+ 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 new StartupConfig(StartupMode.SPECIFIC, specificOffset);
+ }
+
@Override
public SourceConfig.Factory<JdbcSourceConfig>
createSourceConfigFactory(ReadonlyConfig config) {
OracleSourceConfigFactory configFactory = new
OracleSourceConfigFactory();
diff --git
a/seatunnel-connectors-v2/connector-cdc/connector-cdc-oracle/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/oracle/source/OracleIncrementalSourceFactory.java
b/seatunnel-connectors-v2/connector-cdc/connector-cdc-oracle/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/oracle/source/OracleIncrementalSourceFactory.java
index 93e2e7f5b4..2c4fffea86 100644
---
a/seatunnel-connectors-v2/connector-cdc/connector-cdc-oracle/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/oracle/source/OracleIncrementalSourceFactory.java
+++
b/seatunnel-connectors-v2/connector-cdc/connector-cdc-oracle/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/oracle/source/OracleIncrementalSourceFactory.java
@@ -30,7 +30,6 @@ import
org.apache.seatunnel.api.table.factory.TableSourceFactoryContext;
import org.apache.seatunnel.connectors.cdc.base.config.JdbcSourceTableConfig;
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.BaseChangeStreamTableSourceFactory;
import org.apache.seatunnel.connectors.cdc.base.utils.CatalogTableUtils;
import
org.apache.seatunnel.connectors.seatunnel.cdc.oracle.config.OracleSourceConfigFactory;
@@ -42,6 +41,8 @@ import java.io.Serializable;
import java.util.List;
import java.util.Optional;
+import static
org.apache.seatunnel.api.configuration.util.Conditions.greaterThan;
+
@AutoService(Factory.class)
@Slf4j
public class OracleIncrementalSourceFactory extends
BaseChangeStreamTableSourceFactory {
@@ -87,19 +88,15 @@ public class OracleIncrementalSourceFactory extends
BaseChangeStreamTableSourceF
.conditional(
OracleIncrementalSourceOptions.STARTUP_MODE,
StartupMode.SPECIFIC,
- SourceOptions.STARTUP_SPECIFIC_OFFSET_POS)
+
OracleIncrementalSourceOptions.STARTUP_SPECIFIC_OFFSET_SCN)
.conditional(
- OracleIncrementalSourceOptions.STOP_MODE,
- StopMode.SPECIFIC,
- SourceOptions.STOP_SPECIFIC_OFFSET_POS)
+ OracleIncrementalSourceOptions.STARTUP_MODE,
+ StartupMode.SPECIFIC,
+
greaterThan(OracleIncrementalSourceOptions.STARTUP_SPECIFIC_OFFSET_SCN, 0L))
.conditional(
OracleIncrementalSourceOptions.STARTUP_MODE,
StartupMode.TIMESTAMP,
SourceOptions.STARTUP_TIMESTAMP)
- .conditional(
- OracleIncrementalSourceOptions.STOP_MODE,
- StopMode.TIMESTAMP,
- SourceOptions.STOP_TIMESTAMP)
.conditional(
OracleIncrementalSourceOptions.STARTUP_MODE,
StartupMode.INITIAL,
diff --git
a/seatunnel-connectors-v2/connector-cdc/connector-cdc-oracle/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/oracle/source/OracleIncrementalSourceOptions.java
b/seatunnel-connectors-v2/connector-cdc/connector-cdc-oracle/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/oracle/source/OracleIncrementalSourceOptions.java
index ec18a4aaee..d04a81901f 100644
---
a/seatunnel-connectors-v2/connector-cdc/connector-cdc-oracle/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/oracle/source/OracleIncrementalSourceOptions.java
+++
b/seatunnel-connectors-v2/connector-cdc/connector-cdc-oracle/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/oracle/source/OracleIncrementalSourceOptions.java
@@ -37,11 +37,12 @@ public class OracleIncrementalSourceOptions extends
JdbcSourceOptions {
Arrays.asList(
StartupMode.INITIAL,
StartupMode.LATEST,
+ StartupMode.SPECIFIC,
StartupMode.TIMESTAMP))
.defaultValue(StartupMode.INITIAL)
.withDescription(
"Optional startup mode for CDC source,
valid enumerations are "
- + "\"initial\", \"latest\" or
\"timestamp\"");
+ + "\"initial\", \"latest\",
\"specific\" or \"timestamp\"");
public static final SingleChoiceOption<StopMode> STOP_MODE =
(SingleChoiceOption)
@@ -58,6 +59,12 @@ public class OracleIncrementalSourceOptions extends
JdbcSourceOptions {
.noDefaultValue()
.withDescription("Schema name of the database to
monitor.");
+ public static final Option<Long> STARTUP_SPECIFIC_OFFSET_SCN =
+ Options.key("startup.specific-offset.scn")
+ .longType()
+ .noDefaultValue()
+ .withDescription("Optional SCN used in case of
\"specific\" startup mode.");
+
public static final Option<Boolean> USE_SELECT_COUNT =
Options.key("use_select_count")
.booleanType()
diff --git
a/seatunnel-connectors-v2/connector-cdc/connector-cdc-oracle/src/test/java/org/apache/seatunnel/connectors/seatunnel/cdc/oracle/source/OracleIncrementalSourceFactoryTest.java
b/seatunnel-connectors-v2/connector-cdc/connector-cdc-oracle/src/test/java/org/apache/seatunnel/connectors/seatunnel/cdc/oracle/source/OracleIncrementalSourceFactoryTest.java
index 24552b99e0..7bed08ac82 100644
---
a/seatunnel-connectors-v2/connector-cdc/connector-cdc-oracle/src/test/java/org/apache/seatunnel/connectors/seatunnel/cdc/oracle/source/OracleIncrementalSourceFactoryTest.java
+++
b/seatunnel-connectors-v2/connector-cdc/connector-cdc-oracle/src/test/java/org/apache/seatunnel/connectors/seatunnel/cdc/oracle/source/OracleIncrementalSourceFactoryTest.java
@@ -17,12 +17,167 @@
package org.apache.seatunnel.connectors.seatunnel.cdc.oracle.source;
+import org.apache.seatunnel.api.configuration.ReadonlyConfig;
+import org.apache.seatunnel.api.configuration.SingleChoiceOption;
+import org.apache.seatunnel.api.configuration.util.ConfigValidator;
+import org.apache.seatunnel.api.configuration.util.OptionValidationException;
+import org.apache.seatunnel.api.options.ConnectorCommonOptions;
+import org.apache.seatunnel.connectors.cdc.base.config.StartupConfig;
+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.base.source.split.IncrementalSplit;
+import
org.apache.seatunnel.connectors.cdc.base.source.split.state.IncrementalSplitState;
+import
org.apache.seatunnel.connectors.seatunnel.cdc.oracle.source.offset.RedoLogOffset;
+
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
+
class OracleIncrementalSourceFactoryTest {
@Test
public void testOptionRule() {
Assertions.assertNotNull((new
OracleIncrementalSourceFactory()).optionRule());
}
+
+ @Test
+ public void testOnlyNeverStopModeIsSupported() {
+ new OracleIncrementalSourceFactory()
+ .optionRule().getOptionalOptions().stream()
+ .filter((option) ->
option.key().equals(SourceOptions.STOP_MODE_KEY))
+ .forEach(
+ (option) ->
+ Assertions.assertIterableEquals(
+
Collections.singletonList(StopMode.NEVER),
+
((SingleChoiceOption<StopMode>) option)
+ .getOptionValues()));
+ }
+
+ @Test
+ public void testSpecificStartupModeRequiresScn() {
+ Map<String, Object> config = baseConfig();
+ config.put(OracleIncrementalSourceOptions.STARTUP_MODE.key(),
StartupMode.SPECIFIC);
+
+ Assertions.assertThrows(
+ OptionValidationException.class,
+ () ->
+ ConfigValidator.of(ReadonlyConfig.fromMap(config))
+ .validate(new
OracleIncrementalSourceFactory().optionRule()));
+ }
+
+ @Test
+ public void testSpecificStartupModeRejectsInvalidScn() {
+ Map<String, Object> config = specificStartupConfig(0L);
+
+ Assertions.assertThrows(
+ OptionValidationException.class,
+ () ->
+ ConfigValidator.of(ReadonlyConfig.fromMap(config))
+ .validate(new
OracleIncrementalSourceFactory().optionRule()));
+ }
+
+ @Test
+ public void testSpecificStartupModeUsesScnOffset() {
+ StartupConfig startupConfig =
+ OracleIncrementalSource.getOracleStartupConfig(
+
ReadonlyConfig.fromMap(specificStartupConfig(123456789L)));
+
+ RedoLogOffset startupOffset =
+ (RedoLogOffset) startupConfig.getStartupOffset(new
TestOffsetFactory());
+ IncrementalSplit split =
+ new IncrementalSplit(
+ "oracle-incremental-split",
+ Collections.emptyList(),
+ startupOffset,
+ RedoLogOffset.NO_STOPPING_OFFSET,
+ Collections.emptyList());
+ RedoLogOffset restoredOffset =
+ (RedoLogOffset) new
IncrementalSplitState(split).toSourceSplit().getStartupOffset();
+
+ Assertions.assertEquals(StartupMode.SPECIFIC,
startupConfig.getStartupMode());
+ Assertions.assertEquals("123456789", startupOffset.getScn());
+ Assertions.assertEquals("0", startupOffset.getCommitScn());
+ Assertions.assertNull(startupOffset.getLcrPosition());
+ Assertions.assertEquals(startupOffset, restoredOffset);
+ }
+
+ @Test
+ public void testScnOffsetOnlySupportsSpecificStartupMode() {
+ Map<String, Object> config = baseConfig();
+ config.put(OracleIncrementalSourceOptions.STARTUP_MODE.key(),
StartupMode.LATEST);
+
config.put(OracleIncrementalSourceOptions.STARTUP_SPECIFIC_OFFSET_SCN.key(),
123456789L);
+
+ Assertions.assertThrows(
+ IllegalArgumentException.class,
+ () ->
+ OracleIncrementalSource.getOracleStartupConfig(
+ ReadonlyConfig.fromMap(config)));
+ }
+
+ @Test
+ public void testOracleSpecificStartupModeRejectsFilePositionOffset() {
+ Map<String, Object> config = specificStartupConfig(123456789L);
+ config.put(SourceOptions.STARTUP_SPECIFIC_OFFSET_FILE.key(),
"redo.log");
+ config.put(SourceOptions.STARTUP_SPECIFIC_OFFSET_POS.key(), 100L);
+
+ Assertions.assertThrows(
+ IllegalArgumentException.class,
+ () ->
+ OracleIncrementalSource.getOracleStartupConfig(
+ ReadonlyConfig.fromMap(config)));
+ }
+
+ private static Map<String, Object> specificStartupConfig(long scn) {
+ Map<String, Object> config = baseConfig();
+ config.put(OracleIncrementalSourceOptions.STARTUP_MODE.key(),
StartupMode.SPECIFIC);
+
config.put(OracleIncrementalSourceOptions.STARTUP_SPECIFIC_OFFSET_SCN.key(),
scn);
+ return config;
+ }
+
+ private static Map<String, Object> baseConfig() {
+ Map<String, Object> config = new HashMap<>();
+ config.put(OracleIncrementalSourceOptions.USERNAME.key(), "user");
+ config.put(OracleIncrementalSourceOptions.PASSWORD.key(), "password");
+ config.put(ConnectorCommonOptions.TABLE_NAMES.key(),
Arrays.asList("ORCL.TEST"));
+ return config;
+ }
+
+ private static class TestOffsetFactory extends OffsetFactory {
+
+ @Override
+ public Offset earliest() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public Offset neverStop() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public Offset latest() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public Offset specific(Map<String, String> offset) {
+ return new RedoLogOffset(offset);
+ }
+
+ @Override
+ public Offset specific(String filename, Long position) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public Offset timestamp(long timestamp) {
+ throw new UnsupportedOperationException();
+ }
+ }
}
