[cdc] paimon cdc support snapshot mode (apache#4164)

(cherry picked from commit f9164e4)
tongcheng-elong · Sep 11, 2024 · 046ebd1 · 046ebd1
1 parent 8b8f43c
commit 046ebd1
Show file tree

Hide file tree

Showing 4 changed files with 41 additions and 1 deletion.
diff --git a/...link-cdc/src/main/java/org/apache/paimon/flink/action/cdc/mongodb/MongoDBActionUtils.java b/...link-cdc/src/main/java/org/apache/paimon/flink/action/cdc/mongodb/MongoDBActionUtils.java
@@ -59,6 +59,7 @@ public class MongoDBActionUtils {
     private static final String INITIAL_MODE = "initial";
     private static final String LATEST_OFFSET_MODE = "latest-offset";
     private static final String TIMESTAMP_MODE = "timestamp";
+    private static final String SNAPSHOT_MODE = "snapshot";
 
     public static final ConfigOption<String> FIELD_NAME =
             ConfigOptions.key("field.name")
@@ -129,8 +130,14 @@ public static MongoDBSource<CdcSourceRecord> buildMongodbSource(
                         StartupOptions.timestamp(
                                 mongodbConfig.get(SourceOptions.SCAN_STARTUP_TIMESTAMP_MILLIS)));
                 break;
+            case SNAPSHOT_MODE:
+                sourceBuilder.startupOptions(StartupOptions.snapshot());
+                break;
             default:
-                throw new IllegalArgumentException("Unsupported startup mode: " + startupMode);
+                throw new IllegalArgumentException(
+                        String.format(
+                                "Unknown scan.startup.mode='%s'. Valid scan.startup.mode for MongoDB CDC are [initial, latest-offset, timestamp, snapshot]",
+                                startupMode));
         }
 
         Map<String, Object> customConverterConfigs = new HashMap<>();

diff --git a/...on-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/mysql/MySqlActionUtils.java b/...on-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/mysql/MySqlActionUtils.java
@@ -214,6 +214,13 @@ public static MySqlSource<CdcSourceRecord> buildMySqlSource(
             sourceBuilder.startupOptions(
                     StartupOptions.timestamp(
                             mySqlConfig.get(MySqlSourceOptions.SCAN_STARTUP_TIMESTAMP_MILLIS)));
+        } else if ("snapshot".equalsIgnoreCase(startupMode)) {
+            sourceBuilder.startupOptions(StartupOptions.snapshot());
+        } else {
+            throw new IllegalArgumentException(
+                    String.format(
+                            "Unknown scan.startup.mode='%s'. Valid scan.startup.mode for MySQL CDC are [initial, earliest-offset, latest-offset, specific-offset, timestamp, snapshot]",
+                            startupMode));
         }
 
         Properties jdbcProperties = new Properties();

diff --git a/...nk-cdc/src/main/java/org/apache/paimon/flink/action/cdc/postgres/PostgresActionUtils.java b/...nk-cdc/src/main/java/org/apache/paimon/flink/action/cdc/postgres/PostgresActionUtils.java
@@ -167,6 +167,13 @@ public static JdbcIncrementalSource<CdcSourceRecord> buildPostgresSource(
             sourceBuilder.startupOptions(StartupOptions.initial());
         } else if ("latest-offset".equalsIgnoreCase(startupMode)) {
             sourceBuilder.startupOptions(StartupOptions.latest());
+        } else if ("snapshot".equalsIgnoreCase(startupMode)) {
+            sourceBuilder.startupOptions(StartupOptions.snapshot());
+        } else {
+            throw new IllegalArgumentException(
+                    String.format(
+                            "Unknown scan.startup.mode='%s'. Valid scan.startup.mode for Postgres CDC are [initial, latest-offset, snapshot]",
+                            startupMode));
         }
 
         Properties debeziumProperties = new Properties();

diff --git a/...dc/src/test/java/org/apache/paimon/flink/action/cdc/mysql/MySqlSyncTableActionITCase.java b/...dc/src/test/java/org/apache/paimon/flink/action/cdc/mysql/MySqlSyncTableActionITCase.java
@@ -1478,4 +1478,23 @@ public void testWriteOnlyAndSchemaEvolution() throws Exception {
             waitForResult(expected, table, rowType, primaryKeys);
         }
     }
+
+    @Test
+    @Timeout(60)
+    public void testUnknowMysqlScanStartupMode() {
+        String scanStartupMode = "abc";
+        Map<String, String> mySqlConfig = getBasicMySqlConfig();
+        mySqlConfig.put("database-name", DATABASE_NAME);
+        mySqlConfig.put("table-name", "schema_evolution_multiple");
+        mySqlConfig.put("scan.startup.mode", scanStartupMode);
+
+        MySqlSyncTableAction action = syncTableActionBuilder(mySqlConfig).build();
+        assertThatThrownBy(action::run)
+                .satisfies(
+                        anyCauseMatches(
+                                IllegalArgumentException.class,
+                                "Unknown scan.startup.mode='"
+                                        + scanStartupMode
+                                        + "'. Valid scan.startup.mode for MySQL CDC are [initial, earliest-offset, latest-offset, specific-offset, timestamp, snapshot]"));
+    }
 }