[cdc] Complementary flink exec mode check for mongodb and postgres cd…

…c table sync. (#4408)
apache · Oct 31, 2024 · 4c1777f · 4c1777f
1 parent 29a89c9
commit 4c1777f
Show file tree

Hide file tree

Showing 6 changed files with 53 additions and 1 deletion.
diff --git a/...c/src/main/java/org/apache/paimon/flink/action/cdc/mongodb/MongoDBSyncDatabaseAction.java b/...c/src/main/java/org/apache/paimon/flink/action/cdc/mongodb/MongoDBSyncDatabaseAction.java
@@ -66,6 +66,7 @@ protected CdcTimestampExtractor createCdcTimestampExtractor() {
 
     @Override
     protected MongoDBSource<CdcSourceRecord> buildSource() {
+        validateRuntimeExecutionMode();
         return MongoDBActionUtils.buildMongodbSource(
                 cdcSourceConfig,
                 CdcActionCommonUtils.combinedModeTableList(

diff --git a/...-cdc/src/main/java/org/apache/paimon/flink/action/cdc/mongodb/MongoDBSyncTableAction.java b/...-cdc/src/main/java/org/apache/paimon/flink/action/cdc/mongodb/MongoDBSyncTableAction.java
@@ -77,6 +77,7 @@ protected CdcTimestampExtractor createCdcTimestampExtractor() {
 
     @Override
     protected MongoDBSource<CdcSourceRecord> buildSource() {
+        validateRuntimeExecutionMode();
         String tableList =
                 cdcSourceConfig.get(MongoDBSourceOptions.DATABASE)
                         + "\\."

diff --git a/...dc/src/main/java/org/apache/paimon/flink/action/cdc/postgres/PostgresSyncTableAction.java b/...dc/src/main/java/org/apache/paimon/flink/action/cdc/postgres/PostgresSyncTableAction.java
@@ -103,6 +103,7 @@ protected Schema retrieveSchema() throws Exception {
 
     @Override
     protected JdbcIncrementalSource<CdcSourceRecord> buildSource() {
+        validateRuntimeExecutionMode();
         List<JdbcSchemasInfo.JdbcSchemaInfo> pkTables = postgresSchemasInfo.pkTables();
         Set<String> schemaList = new HashSet<>();
         String[] tableList = new String[pkTables.size()];

diff --git a/...rc/test/java/org/apache/paimon/flink/action/cdc/mongodb/MongoDBSyncTableActionITCase.java b/...rc/test/java/org/apache/paimon/flink/action/cdc/mongodb/MongoDBSyncTableActionITCase.java
@@ -34,7 +34,9 @@
 import java.util.List;
 import java.util.Map;
 
+import static org.apache.paimon.testutils.assertj.PaimonAssertions.anyCauseMatches;
 import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
 
 /** IT cases for {@link MongoDBSyncTableAction}. */
 public class MongoDBSyncTableActionITCase extends MongoDBActionITCaseBase {
@@ -398,4 +400,26 @@ public void testComputedColumnWithCaseInsensitive() throws Exception {
                 rowType,
                 Collections.singletonList("_id"));
     }
+
+    @Test
+    @Timeout(60)
+    public void testRuntimeExecutionModeCheckForCdcSync() {
+        Map<String, String> mongodbConfig = getBasicMongoDBConfig();
+        mongodbConfig.put("database", database);
+        mongodbConfig.put("collection", "products");
+        mongodbConfig.put("field.name", "_id,name,description");
+        mongodbConfig.put("parser.path", "$._id,$.name,$.description");
+        mongodbConfig.put("schema.start.mode", "specified");
+
+        MongoDBSyncTableAction action =
+                syncTableActionBuilder(mongodbConfig)
+                        .withTableConfig(getBasicTableConfig())
+                        .build();
+
+        assertThatThrownBy(() -> runActionWithBatchEnv(action))
+                .satisfies(
+                        anyCauseMatches(
+                                IllegalArgumentException.class,
+                                "It's only support STREAMING mode for flink-cdc sync table action"));
+    }
 }
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
@@ -1499,7 +1499,7 @@ public void testUnknowMysqlScanStartupMode() {
     }
 
     @Test
-    @Timeout(60)
+    @Timeout(1000)
     public void testRuntimeExecutionModeCheckForCdcSync() throws Exception {
         Map<String, String> mySqlConfig = getBasicMySqlConfig();
         mySqlConfig.put("database-name", "check_cdc_sync_runtime_execution_mode");

diff --git a/.../test/java/org/apache/paimon/flink/action/cdc/postgres/PostgresSyncTableActionITCase.java b/.../test/java/org/apache/paimon/flink/action/cdc/postgres/PostgresSyncTableActionITCase.java
@@ -789,4 +789,29 @@ public void testCatalogAndTableConfig() {
     private FileStoreTable getFileStoreTable() throws Exception {
         return getFileStoreTable(tableName);
     }
+
+    @Test
+    @Timeout(60)
+    public void testRuntimeExecutionModeCheckForCdcSync() {
+        Map<String, String> postgresConfig = getBasicPostgresConfig();
+        postgresConfig.put(PostgresSourceOptions.DATABASE_NAME.key(), DATABASE_NAME);
+        postgresConfig.put(PostgresSourceOptions.SCHEMA_NAME.key(), SCHEMA_NAME);
+        postgresConfig.put(PostgresSourceOptions.TABLE_NAME.key(), "schema_evolution_\\d+");
+
+        PostgresSyncTableAction action =
+                syncTableActionBuilder(postgresConfig)
+                        .withCatalogConfig(
+                                Collections.singletonMap(
+                                        CatalogOptions.METASTORE.key(), "test-alter-table"))
+                        .withTableConfig(getBasicTableConfig())
+                        .withPartitionKeys("pt")
+                        .withPrimaryKeys("pt", "_id")
+                        .build();
+
+        assertThatThrownBy(() -> runActionWithBatchEnv(action))
+                .satisfies(
+                        anyCauseMatches(
+                                IllegalArgumentException.class,
+                                "It's only support STREAMING mode for flink-cdc sync table action"));
+    }
 }