Accept startup options in table factory

Author: PatrickRen

flink-connector-mysql-cdc/src/main/java/com/ververica/cdc/connectors/mysql/table/MySqlTableSourceFactory.java

@@ -23,7 +23,6 @@
 import org.apache.flink.table.connector.source.DynamicTableSource;
 import org.apache.flink.table.factories.DynamicTableSourceFactory;
 import org.apache.flink.table.factories.FactoryUtil;
-import org.apache.flink.util.Preconditions;
 
 import com.ververica.cdc.connectors.mysql.source.config.MySqlSourceOptions;
 import com.ververica.cdc.connectors.mysql.source.config.ServerIdRange;
@@ -109,7 +108,6 @@ public DynamicTableSource createDynamicTableSource(Context context) {
         boolean enableParallelRead = config.get(SCAN_INCREMENTAL_SNAPSHOT_ENABLED);
         if (enableParallelRead) {
             validatePrimaryKeyIfEnableParallel(physicalSchema);
-            validateStartupOptionIfEnableParallel(startupOptions);
             validateIntegerOption(SCAN_INCREMENTAL_SNAPSHOT_CHUNK_SIZE, splitSize, 1);
             validateIntegerOption(CHUNK_META_GROUP_SIZE, splitMetaGroupSize, 1);
             validateIntegerOption(SCAN_SNAPSHOT_FETCH_SIZE, fetchSize, 1);
@@ -204,15 +202,15 @@ private static StartupOptions getStartupOptions(ReadableConfig config) {
                 return StartupOptions.latest();
 
             case SCAN_STARTUP_MODE_VALUE_EARLIEST:
+                return StartupOptions.earliest();
+
             case SCAN_STARTUP_MODE_VALUE_SPECIFIC_OFFSET:
+                return StartupOptions.specificOffset(
+                        config.get(SCAN_STARTUP_SPECIFIC_OFFSET_FILE),
+                        config.get(SCAN_STARTUP_SPECIFIC_OFFSET_POS));
+
             case SCAN_STARTUP_MODE_VALUE_TIMESTAMP:
-                throw new ValidationException(
-                        String.format(
-                                "Unsupported option value '%s', the options [%s, %s, %s] are not supported correctly, please do not use them until they're correctly supported",
-                                modeString,
-                                SCAN_STARTUP_MODE_VALUE_EARLIEST,
-                                SCAN_STARTUP_MODE_VALUE_SPECIFIC_OFFSET,
-                                SCAN_STARTUP_MODE_VALUE_TIMESTAMP));
+                return StartupOptions.timestamp(config.get(SCAN_STARTUP_TIMESTAMP_MILLIS));
 
             default:
                 throw new ValidationException(
@@ -234,17 +232,6 @@ private void validatePrimaryKeyIfEnableParallel(ResolvedSchema physicalSchema) {
         }
     }
 
-    private void validateStartupOptionIfEnableParallel(StartupOptions startupOptions) {
-        // validate mode
-        Preconditions.checkState(
-                startupOptions.startupMode == StartupMode.INITIAL
-                        || startupOptions.startupMode == StartupMode.LATEST_OFFSET,
-                String.format(
-                        "MySql Parallel Source only supports startup mode 'initial' and 'latest-offset',"
-                                + " but actual is %s",
-                        startupOptions.startupMode));
-    }
-
     private String validateAndGetServerId(ReadableConfig configuration) {
         final String serverIdValue = configuration.get(MySqlSourceOptions.SERVER_ID);
         if (serverIdValue != null) {