(incubator-paimon) branch master updated: [cdc] kafka adds sasl and other parameters to create consumers (#2484)

[lzljs3620320]

This is an automated email from the ASF dual-hosted git repository.

lzljs3620320 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/incubator-paimon.git


The following commit(s) were added to refs/heads/master by this push:
     new 6200fec38 [cdc] kafka adds sasl and other parameters to create 
consumers (#2484)
6200fec38 is described below

commit 6200fec38750f029b750a43eb1fddff878193d1f
Author: Kerwin <37063904+zhuangch...@users.noreply.github.com>
AuthorDate: Tue Dec 12 09:59:11 2023 +0800

    [cdc] kafka adds sasl and other parameters to create consumers (#2484)
---
 .../apache/paimon/flink/action/cdc/kafka/KafkaActionUtils.java   | 9 +++++++++
 1 file changed, 9 insertions(+)

diff --git 
a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/kafka/KafkaActionUtils.java
 
b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/kafka/KafkaActionUtils.java
index 8b98c8c06..c9ad0e615 100644
--- 
a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/kafka/KafkaActionUtils.java
+++ 
b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/kafka/KafkaActionUtils.java
@@ -241,6 +241,15 @@ public class KafkaActionUtils {
     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));