[flink] Fix kafka CDC Ingestion multi-topic exception (#2030)

apache · Sep 18, 2023 · 119ec42 · 119ec42
1 parent a280ec6
commit 119ec42
Showing 1 changed file with 7 additions and 1 deletion.
diff --git a/...on-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/kafka/KafkaActionUtils.java b/...on-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/kafka/KafkaActionUtils.java
@@ -89,8 +89,14 @@ static Schema buildPaimonSchema(
     static KafkaSource<String> buildKafkaSource(Configuration kafkaConfig) {
         validateKafkaConfig(kafkaConfig);
         KafkaSourceBuilder<String> kafkaSourceBuilder = KafkaSource.builder();
+
+        List<String> topics =
+                kafkaConfig.get(KafkaConnectorOptions.TOPIC).stream()
+                        .flatMap(topic -> Arrays.stream(topic.split(",")))
+                        .collect(Collectors.toList());
+
         kafkaSourceBuilder
-                .setTopics(kafkaConfig.get(KafkaConnectorOptions.TOPIC))
+                .setTopics(topics)
                 .setValueOnlyDeserializer(new SimpleStringSchema())
                 .setGroupId(kafkaPropertiesGroupId(kafkaConfig));
         Properties properties = new Properties();