This is an automated email from the ASF dual-hosted git repository.
lzljs3620320 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/paimon.git
The following commit(s) were added to refs/heads/master by this push:
new f9164e4f3 [cdc] paimon cdc support snapshot mode (#4164)
f9164e4f3 is described below
commit f9164e4f329c9ad1b8ccc0ad2ec3734bb96f23c2
Author: MOBIN <[email protected]>
AuthorDate: Wed Sep 11 09:52:11 2024 +0800
[cdc] paimon cdc support snapshot mode (#4164)
---
.../flink/action/cdc/mongodb/MongoDBActionUtils.java | 9 ++++++++-
.../flink/action/cdc/mysql/MySqlActionUtils.java | 7 +++++++
.../action/cdc/postgres/PostgresActionUtils.java | 7 +++++++
.../action/cdc/mysql/MySqlSyncTableActionITCase.java | 19 +++++++++++++++++++
4 files changed, 41 insertions(+), 1 deletion(-)
diff --git
a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/mongodb/MongoDBActionUtils.java
b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/mongodb/MongoDBActionUtils.java
index 848e96c69..ddc69a180 100644
---
a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/mongodb/MongoDBActionUtils.java
+++
b/paimon-flink/paimon-flink-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 class MongoDBActionUtils {
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/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/mysql/MySqlActionUtils.java
b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/mysql/MySqlActionUtils.java
index 6888f672e..f97aa28b1 100644
---
a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/mysql/MySqlActionUtils.java
+++
b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/mysql/MySqlActionUtils.java
@@ -214,6 +214,13 @@ public class MySqlActionUtils {
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/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/postgres/PostgresActionUtils.java
b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/postgres/PostgresActionUtils.java
index 027fef8fd..5107ba146 100644
---
a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/postgres/PostgresActionUtils.java
+++
b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/postgres/PostgresActionUtils.java
@@ -167,6 +167,13 @@ public class PostgresActionUtils {
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/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/action/cdc/mysql/MySqlSyncTableActionITCase.java
b/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/action/cdc/mysql/MySqlSyncTableActionITCase.java
index 7e11b541d..ce18d0b1f 100644
---
a/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/action/cdc/mysql/MySqlSyncTableActionITCase.java
+++
b/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/action/cdc/mysql/MySqlSyncTableActionITCase.java
@@ -1478,4 +1478,23 @@ public class MySqlSyncTableActionITCase extends
MySqlActionITCaseBase {
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]"));
+ }
}
