[cdc] kafka adds sasl and other parameters to create consumers (apach…

…e#2484)
tongcheng-elong · Dec 12, 2023 · 6200fec · 6200fec
1 parent 117bcb0
commit 6200fec
Showing 1 changed file with 9 additions and 0 deletions.
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
@@ -241,6 +241,15 @@ public static DataFormat getDataFormat(Configuration kafkaConfig) {
     public static MessageQueueSchemaUtils.ConsumerWrapper getKafkaEarliestConsumer(
             Configuration kafkaConfig) {
         Properties props = new Properties();
+
+        for (Map.Entry<String, String> entry : kafkaConfig.toMap().entrySet()) {
+            String key = entry.getKey();
+            String value = entry.getValue();
+            if (key.startsWith(PROPERTIES_PREFIX)) {
+                props.put(key.substring(PROPERTIES_PREFIX.length()), value);
+            }
+        }
+
         props.put(
                 ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,
                 kafkaConfig.get(KafkaConnectorOptions.PROPS_BOOTSTRAP_SERVERS));