update config path

apache · Jan 8, 2025 · 25f720b · 25f720b
1 parent 9d1418b
commit 25f720b
Showing 1 changed file with 4 additions and 1 deletion.
diff --git a/...c/main/java/org/apache/seatunnel/connectors/seatunnel/kafka/source/KafkaSourceConfig.java b/...c/main/java/org/apache/seatunnel/connectors/seatunnel/kafka/source/KafkaSourceConfig.java
@@ -134,7 +134,10 @@ private Map<TablePath, ConsumerMetadata> createMapConsumerMetadata(
         return consumerMetadataList.stream()
                 .collect(
                         Collectors.toMap(
-                                consumerMetadata -> TablePath.of(consumerMetadata.getTopic()),
+                                consumerMetadata -> {
+                                    String topic = consumerMetadata.getTopic().replace(".", "_");
+                                    return TablePath.of("kafka", topic);
+                                },
                                 consumerMetadata -> consumerMetadata));
     }