[cdc] Refactor kafka cdc test (apache#2672)

paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/action/cdc/kafka/KafkaCanalSyncTableActionITCase.java

@@ -20,8 +20,6 @@
 
 import org.apache.paimon.CoreOptions;
 import org.apache.paimon.catalog.FileSystemCatalogOptions;
-import org.apache.paimon.catalog.Identifier;
-import org.apache.paimon.table.AbstractFileStoreTable;
 import org.apache.paimon.table.FileStoreTable;
 import org.apache.paimon.types.DataType;
 import org.apache.paimon.types.DataTypes;
@@ -57,7 +55,9 @@
 import static org.assertj.core.api.Assertions.assertThatThrownBy;
 
 /** IT cases for {@link KafkaSyncTableAction}. */
-public class KafkaCanalSyncTableActionITCase extends KafkaActionITCaseBase {
+public class KafkaCanalSyncTableActionITCase extends KafkaSyncTableActionITCase {
+
+    private static final String CANAL = "canal";
 
     @Test
     @Timeout(60)
@@ -1247,34 +1247,6 @@ public void testComputedColumnWithCaseInsensitive(boolean triggerSchemaRetrieval
     @Test
     @Timeout(60)
     public void testWaterMarkSyncTable() throws Exception {
-        String topic = "watermark";
-        createTestTopic(topic, 1, 1);
-        writeRecordsToKafka(topic, readLines("kafka/canal/table/watermark/canal-data-1.txt"));
-
-        Map<String, String> kafkaConfig = getBasicKafkaConfig();
-        kafkaConfig.put(VALUE_FORMAT.key(), "canal-json");
-        kafkaConfig.put(TOPIC.key(), topic);
-
-        Map<String, String> config = getBasicTableConfig();
-        config.put("tag.automatic-creation", "watermark");
-        config.put("tag.creation-period", "hourly");
-        config.put("scan.watermark.alignment.group", "alignment-group-1");
-        config.put("scan.watermark.alignment.max-drift", "20 s");
-        config.put("scan.watermark.alignment.update-interval", "1 s");
-
-        KafkaSyncTableAction action =
-                syncTableActionBuilder(kafkaConfig).withTableConfig(config).build();
-        runActionWithDefaultEnv(action);
-
-        AbstractFileStoreTable table =
-                (AbstractFileStoreTable) catalog.getTable(new Identifier(database, tableName));
-        while (true) {
-            if (table.snapshotManager().snapshotCount() > 0
-                    && table.snapshotManager().latestSnapshot().watermark()
-                            != -9223372036854775808L) {
-                return;
-            }
-            Thread.sleep(1000);
-        }
+        testWaterMarkSyncTable(CANAL);
     }
 }

paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/action/cdc/kafka/KafkaDebeziumSyncDatabaseActionITCase.java

@@ -0,0 +1,74 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.paimon.flink.action.cdc.kafka;
+
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.Timeout;
+
+/** IT cases for {@link KafkaSyncDatabaseAction}. */
+public class KafkaDebeziumSyncDatabaseActionITCase extends KafkaSyncDatabaseActionITCase {
+    private static final String DEBEZIUM = "debezium";
+
+    @Test
+    @Timeout(60)
+    public void testSchemaEvolutionMultiTopic() throws Exception {
+        testSchemaEvolutionMultiTopic(DEBEZIUM);
+    }
+
+    @Test
+    @Timeout(60)
+    public void testSchemaEvolutionOneTopic() throws Exception {
+        testSchemaEvolutionOneTopic(DEBEZIUM);
+    }
+
+    @Test
+    public void testTopicIsEmpty() {
+        testTopicIsEmpty(DEBEZIUM);
+    }
+
+    @Test
+    @Timeout(60)
+    public void testTableAffixMultiTopic() throws Exception {
+        testTableAffixMultiTopic(DEBEZIUM);
+    }
+
+    @Test
+    @Timeout(60)
+    public void testTableAffixOneTopic() throws Exception {
+        testTableAffixOneTopic(DEBEZIUM);
+    }
+
+    @Test
+    @Timeout(60)
+    public void testIncludingTables() throws Exception {
+        testIncludingTables(DEBEZIUM);
+    }
+
+    @Test
+    @Timeout(60)
+    public void testExcludingTables() throws Exception {
+        testExcludingTables(DEBEZIUM);
+    }
+
+    @Test
+    @Timeout(60)
+    public void testIncludingAndExcludingTables() throws Exception {
+        testIncludingAndExcludingTables(DEBEZIUM);
+    }
+}