[FLINK-36631][source-connector][oracle] Add oracle read data by specific offset (#3675)

Author: JNSimba

docs/content.zh/docs/connectors/flink-sources/oracle-cdc.md

@@ -351,9 +351,16 @@ Connector Options
       <td style="word-wrap: break-word;">initial</td>
       <td>String</td>
       <td>Optional startup mode for Oracle CDC consumer, valid enumerations are "initial"
-           and "latest-offset". 
+           , "latest-offset" , specific-offset. 
            Please see <a href="#startup-reading-position">Startup Reading Position</a> section for more detailed information.</td>
     </tr>
+    <tr>
+      <td>scan.startup.specific-offset.scn</td>
+      <td>optional</td>
+      <td style="word-wrap: break-word;">(none)</td>
+      <td>Long</td>
+      <td>Optional SCN used in case of "specific-offset" startup mode</td>
+    </tr>
     <tr>
           <td>scan.incremental.snapshot.enabled</td>
           <td>optional</td>
@@ -543,6 +550,7 @@ The config option `scan.startup.mode` specifies the startup mode for Oracle CDC
 - `initial` (default): Performs an initial snapshot on the monitored database tables upon first startup, and continue to read the latest redo log.
 - `latest-offset`: Never to perform a snapshot on the monitored database tables upon first startup, just read from
   the change since the connector was started.
+- `specific-offset`: Skip snapshot phase and start reading redo log from a specific offset with scn.
 
 _Note: the mechanism of `scan.startup.mode` option relying on Debezium's `snapshot.mode` configuration. So please do not use them together. If you specific both `scan.startup.mode` and `debezium.snapshot.mode` options in the table DDL, it may make `scan.startup.mode` doesn't work._
 

docs/content/docs/connectors/flink-sources/oracle-cdc.md

@@ -352,9 +352,16 @@ Connector Options
       <td style="word-wrap: break-word;">initial</td>
       <td>String</td>
       <td>Optional startup mode for Oracle CDC consumer, valid enumerations are "initial"
-           and "latest-offset". 
+           , "latest-offset" , specific-offset. 
            Please see <a href="#startup-reading-position">Startup Reading Position</a> section for more detailed information.</td>
     </tr>
+    <tr>
+      <td>scan.startup.specific-offset.scn</td>
+      <td>optional</td>
+      <td style="word-wrap: break-word;">(none)</td>
+      <td>Long</td>
+      <td>Optional SCN used in case of "specific-offset" startup mode</td>
+    </tr>
     <tr>
           <td>scan.incremental.snapshot.enabled</td>
           <td>optional</td>
@@ -544,6 +551,7 @@ The config option `scan.startup.mode` specifies the startup mode for Oracle CDC
 - `initial` (default): Performs an initial snapshot on the monitored database tables upon first startup, and continue to read the latest redo log.
 - `latest-offset`: Never to perform a snapshot on the monitored database tables upon first startup, just read from
   the change since the connector was started.
+- `specific-offset`: Skip snapshot phase and start reading redo log from a specific offset with scn.
 
 _Note: the mechanism of `scan.startup.mode` option relying on Debezium's `snapshot.mode` configuration. So please do not use them together. If you specific both `scan.startup.mode` and `debezium.snapshot.mode` options in the table DDL, it may make `scan.startup.mode` doesn't work._
 

flink-cdc-connect/flink-cdc-source-connectors/flink-cdc-base/src/main/java/org/apache/flink/cdc/connectors/base/config/JdbcSourceConfigFactory.java

@@ -209,6 +209,7 @@ public JdbcSourceConfigFactory startupOptions(StartupOptions startupOptions) {
             case INITIAL:
             case SNAPSHOT:
             case LATEST_OFFSET:
+            case SPECIFIC_OFFSETS:
             case COMMITTED_OFFSETS:
                 break;
             default:

flink-cdc-connect/flink-cdc-source-connectors/flink-cdc-base/src/main/java/org/apache/flink/cdc/connectors/base/options/StartupOptions.java

@@ -17,7 +17,10 @@
 
 package org.apache.flink.cdc.connectors.base.options;
 
+import org.apache.flink.util.CollectionUtil;
+
 import java.io.Serializable;
+import java.util.Map;
 import java.util.Objects;
 
 import static org.apache.flink.util.Preconditions.checkNotNull;
@@ -30,21 +33,22 @@ public final class StartupOptions implements Serializable {
     public final String specificOffsetFile;
     public final Integer specificOffsetPos;
     public final Long startupTimestampMillis;
+    public final Map<String, String> offset;
 
     /**
      * Performs an initial snapshot on the monitored database tables upon first startup, and
      * continue to read the latest change log.
      */
     public static StartupOptions initial() {
-        return new StartupOptions(StartupMode.INITIAL, null, null, null);
+        return new StartupOptions(StartupMode.INITIAL, null, null, null, null);
     }
 
     /**
      * Performs an initial snapshot on the monitored database tables upon first startup, and not
      * read the change log anymore .
      */
     public static StartupOptions snapshot() {
-        return new StartupOptions(StartupMode.SNAPSHOT, null, null, null);
+        return new StartupOptions(StartupMode.SNAPSHOT, null, null, null, null);
     }
 
     /**
@@ -53,15 +57,15 @@ public static StartupOptions snapshot() {
      * change log is guaranteed to contain the entire history of the database.
      */
     public static StartupOptions earliest() {
-        return new StartupOptions(StartupMode.EARLIEST_OFFSET, null, null, null);
+        return new StartupOptions(StartupMode.EARLIEST_OFFSET, null, null, null, null);
     }
 
     /**
      * Never to perform snapshot on the monitored database tables upon first startup, just read from
      * the end of the change log which means only have the changes since the connector was started.
      */
     public static StartupOptions latest() {
-        return new StartupOptions(StartupMode.LATEST_OFFSET, null, null, null);
+        return new StartupOptions(StartupMode.LATEST_OFFSET, null, null, null, null);
     }
 
     /**
@@ -70,7 +74,7 @@ public static StartupOptions latest() {
      * the connector last stopped.
      */
     public static StartupOptions committed() {
-        return new StartupOptions(StartupMode.COMMITTED_OFFSETS, null, null, null);
+        return new StartupOptions(StartupMode.COMMITTED_OFFSETS, null, null, null, null);
     }
 
     /**
@@ -79,7 +83,11 @@ public static StartupOptions committed() {
      */
     public static StartupOptions specificOffset(String specificOffsetFile, int specificOffsetPos) {
         return new StartupOptions(
-                StartupMode.SPECIFIC_OFFSETS, specificOffsetFile, specificOffsetPos, null);
+                StartupMode.SPECIFIC_OFFSETS, specificOffsetFile, specificOffsetPos, null, null);
+    }
+
+    public static StartupOptions specificOffset(Map<String, String> offset) {
+        return new StartupOptions(StartupMode.SPECIFIC_OFFSETS, null, null, null, offset);
     }
 
     /**
@@ -92,18 +100,20 @@ public static StartupOptions specificOffset(String specificOffsetFile, int speci
      * @param startupTimestampMillis timestamp for the startup offsets, as milliseconds from epoch.
      */
     public static StartupOptions timestamp(long startupTimestampMillis) {
-        return new StartupOptions(StartupMode.TIMESTAMP, null, null, startupTimestampMillis);
+        return new StartupOptions(StartupMode.TIMESTAMP, null, null, startupTimestampMillis, null);
     }
 
     private StartupOptions(
             StartupMode startupMode,
             String specificOffsetFile,
             Integer specificOffsetPos,
-            Long startupTimestampMillis) {
+            Long startupTimestampMillis,
+            Map<String, String> offset) {
         this.startupMode = startupMode;
         this.specificOffsetFile = specificOffsetFile;
         this.specificOffsetPos = specificOffsetPos;
         this.startupTimestampMillis = startupTimestampMillis;
+        this.offset = offset;
 
         switch (startupMode) {
             case INITIAL:
@@ -113,8 +123,10 @@ private StartupOptions(
             case COMMITTED_OFFSETS:
                 break;
             case SPECIFIC_OFFSETS:
-                checkNotNull(specificOffsetFile, "specificOffsetFile shouldn't be null");
-                checkNotNull(specificOffsetPos, "specificOffsetPos shouldn't be null");
+                if (CollectionUtil.isNullOrEmpty(offset)) {
+                    checkNotNull(specificOffsetFile, "specificOffsetFile shouldn't be null");
+                    checkNotNull(specificOffsetPos, "specificOffsetPos shouldn't be null");
+                }
                 break;
             case TIMESTAMP:
                 checkNotNull(startupTimestampMillis, "startupTimestampMillis shouldn't be null");
@@ -124,6 +136,10 @@ private StartupOptions(
         }
     }
 
+    public Map<String, String> getOffset() {
+        return offset;
+    }
+
     public boolean isStreamOnly() {
         return startupMode == StartupMode.EARLIEST_OFFSET
                 || startupMode == StartupMode.LATEST_OFFSET
@@ -148,12 +164,13 @@ public boolean equals(Object o) {
         return startupMode == that.startupMode
                 && Objects.equals(specificOffsetFile, that.specificOffsetFile)
                 && Objects.equals(specificOffsetPos, that.specificOffsetPos)
-                && Objects.equals(startupTimestampMillis, that.startupTimestampMillis);
+                && Objects.equals(startupTimestampMillis, that.startupTimestampMillis)
+                && Objects.equals(offset, that.offset);
     }
 
     @Override
     public int hashCode() {
         return Objects.hash(
-                startupMode, specificOffsetFile, specificOffsetPos, startupTimestampMillis);
+                startupMode, specificOffsetFile, specificOffsetPos, startupTimestampMillis, offset);
     }
 }

flink-cdc-connect/flink-cdc-source-connectors/flink-cdc-base/src/main/java/org/apache/flink/cdc/connectors/base/source/assigner/StreamSplitAssigner.java

@@ -30,6 +30,7 @@
 import org.apache.flink.cdc.connectors.base.source.meta.split.SourceSplitBase;
 import org.apache.flink.cdc.connectors.base.source.meta.split.StreamSplit;
 import org.apache.flink.cdc.connectors.base.source.metrics.SourceEnumeratorMetrics;
+import org.apache.flink.util.CollectionUtil;
 
 import java.io.IOException;
 import java.util.ArrayList;
@@ -182,10 +183,14 @@ public StreamSplit createStreamSplit() {
                         offsetFactory.createTimestampOffset(startupOptions.startupTimestampMillis);
                 break;
             case SPECIFIC_OFFSETS:
-                startingOffset =
-                        offsetFactory.newOffset(
-                                startupOptions.specificOffsetFile,
-                                startupOptions.specificOffsetPos.longValue());
+                if (CollectionUtil.isNullOrEmpty(startupOptions.getOffset())) {
+                    startingOffset =
+                            offsetFactory.newOffset(
+                                    startupOptions.specificOffsetFile,
+                                    startupOptions.specificOffsetPos.longValue());
+                } else {
+                    startingOffset = offsetFactory.newOffset(startupOptions.getOffset());
+                }
                 break;
             case COMMITTED_OFFSETS:
                 startingOffset = dialect.displayCommittedOffset(sourceConfig);

flink-cdc-connect/flink-cdc-source-connectors/flink-connector-oracle-cdc/src/main/java/org/apache/flink/cdc/connectors/oracle/table/OracleTableSourceFactory.java

@@ -30,7 +30,9 @@
 import org.apache.flink.table.factories.FactoryUtil;
 
 import java.time.Duration;
+import java.util.HashMap;
 import java.util.HashSet;
+import java.util.Map;
 import java.util.Set;
 
 import static org.apache.flink.cdc.connectors.base.options.JdbcSourceOptions.CONNECTION_POOL_SIZE;
@@ -70,7 +72,9 @@ public class OracleTableSourceFactory implements DynamicTableSourceFactory {
     public DynamicTableSource createDynamicTableSource(Context context) {
         final FactoryUtil.TableFactoryHelper helper =
                 FactoryUtil.createTableFactoryHelper(this, context);
-        helper.validateExcept(DebeziumOptions.DEBEZIUM_OPTIONS_PREFIX);
+        helper.validateExcept(
+                DebeziumOptions.DEBEZIUM_OPTIONS_PREFIX,
+                SCAN_STARTUP_MODE_VALUE_SPECIFIC_OFFSETS_PREFIX);
 
         final ReadableConfig config = helper.getOptions();
         String url = config.get(URL);
@@ -91,7 +95,9 @@ public DynamicTableSource createDynamicTableSource(Context context) {
         String tableName = config.get(TABLE_NAME);
         String schemaName = config.get(SCHEMA_NAME);
         int port = config.get(PORT);
-        StartupOptions startupOptions = getStartupOptions(config);
+
+        StartupOptions startupOptions =
+                getStartupOptions(config, context.getCatalogTable().getOptions());
         ResolvedSchema physicalSchema = context.getCatalogTable().getResolvedSchema();
 
         boolean enableParallelRead = config.get(SCAN_INCREMENTAL_SNAPSHOT_ENABLED);
@@ -195,8 +201,12 @@ public Set<ConfigOption<?>> optionalOptions() {
     private static final String SCAN_STARTUP_MODE_VALUE_INITIAL = "initial";
     private static final String SCAN_STARTUP_MODE_VALUE_SNAPSHOT = "snapshot";
     private static final String SCAN_STARTUP_MODE_VALUE_LATEST = "latest-offset";
+    private static final String SCAN_STARTUP_MODE_VALUE_SPECIFIC_OFFSETS = "specific-offset";
+    private static final String SCAN_STARTUP_MODE_VALUE_SPECIFIC_OFFSETS_PREFIX =
+            "scan.startup.specific-offset.";
 
-    private static StartupOptions getStartupOptions(ReadableConfig config) {
+    private static StartupOptions getStartupOptions(
+            ReadableConfig config, Map<String, String> options) {
         String modeString = config.get(SCAN_STARTUP_MODE);
 
         switch (modeString.toLowerCase()) {
@@ -206,7 +216,9 @@ private static StartupOptions getStartupOptions(ReadableConfig config) {
                 return StartupOptions.snapshot();
             case SCAN_STARTUP_MODE_VALUE_LATEST:
                 return StartupOptions.latest();
-
+            case SCAN_STARTUP_MODE_VALUE_SPECIFIC_OFFSETS:
+                Map<String, String> offsetMap = getSpecificOffsetMap(options);
+                return StartupOptions.specificOffset(offsetMap);
             default:
                 throw new ValidationException(
                         String.format(
@@ -219,6 +231,20 @@ private static StartupOptions getStartupOptions(ReadableConfig config) {
         }
     }
 
+    private static Map<String, String> getSpecificOffsetMap(Map<String, String> options) {
+        Map<String, String> offset = new HashMap<>();
+        for (Map.Entry<String, String> entry : options.entrySet()) {
+            if (entry.getKey().startsWith(SCAN_STARTUP_MODE_VALUE_SPECIFIC_OFFSETS_PREFIX)) {
+                String subKey =
+                        entry.getKey()
+                                .substring(
+                                        SCAN_STARTUP_MODE_VALUE_SPECIFIC_OFFSETS_PREFIX.length());
+                offset.put(subKey, entry.getValue());
+            }
+        }
+        return offset;
+    }
+
     /** Checks the value of given integer option is valid. */
     private void validateIntegerOption(
             ConfigOption<Integer> option, int optionValue, int exclusiveMin) {