feat:Mysql-CDC supports configuring parameter scan.incremental.snapsh…

…ot.backfill.skip,Postgres-CDC supports scan.incremental.snapshot.backfill.skip and scan.incremental.close-idle-reader.enabled parameters
zhourui999 · Aug 19, 2024 · b986dcb · b986dcb
1 parent 0a11857
commit b986dcb
Show file tree

Hide file tree

Showing 2 changed files with 9 additions and 1 deletion.
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
@@ -176,10 +176,12 @@ public static MySqlSource<CdcSourceRecord> buildMySqlSource(
         mySqlConfig
                 .getOptional(MySqlSourceOptions.HEARTBEAT_INTERVAL)
                 .ifPresent(sourceBuilder::heartbeatInterval);
-
         mySqlConfig
                 .getOptional(MySqlSourceOptions.SCAN_INCREMENTAL_CLOSE_IDLE_READER_ENABLED)
                 .ifPresent(sourceBuilder::closeIdleReaders);
+        mySqlConfig
+                .getOptional(MySqlSourceOptions.SCAN_INCREMENTAL_SNAPSHOT_BACKFILL_SKIP)
+                .ifPresent(sourceBuilder::skipSnapshotBackfill);
 
         String startupMode = mySqlConfig.get(MySqlSourceOptions.SCAN_STARTUP_MODE);
         // see

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
@@ -154,6 +154,12 @@ public static JdbcIncrementalSource<CdcSourceRecord> buildPostgresSource(
         postgresConfig
                 .getOptional(PostgresSourceOptions.HEARTBEAT_INTERVAL)
                 .ifPresent(sourceBuilder::heartbeatInterval);
+        postgresConfig
+                .getOptional(PostgresSourceOptions.SCAN_INCREMENTAL_CLOSE_IDLE_READER_ENABLED)
+                .ifPresent(sourceBuilder::closeIdleReaders);
+        postgresConfig
+                .getOptional(PostgresSourceOptions.SCAN_INCREMENTAL_SNAPSHOT_BACKFILL_SKIP)
+                .ifPresent(sourceBuilder::skipSnapshotBackfill);
 
         String startupMode = postgresConfig.get(PostgresSourceOptions.SCAN_STARTUP_MODE);