(incubator-paimon) branch master updated: [cdc] Extract method in KafkaActionUtils

[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 49a51d32e [cdc] Extract method in KafkaActionUtils
49a51d32e is described below

commit 49a51d32e8ac6a014e25ec904ed0b1f47ff2216e
Author: Jingsong <jingsongl...@gmail.com>
AuthorDate: Tue Dec 12 10:03:01 2023 +0800

    [cdc] Extract method in KafkaActionUtils
---
 .../flink/action/cdc/kafka/KafkaActionUtils.java   | 31 ++++++++++------------
 1 file changed, 14 insertions(+), 17 deletions(-)

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 c9ad0e615..996289078 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
@@ -76,14 +76,7 @@ public class KafkaActionUtils {
                 .setTopics(topics)
                 .setValueOnlyDeserializer(new SimpleStringSchema())
                 .setGroupId(kafkaPropertiesGroupId(kafkaConfig));
-        Properties properties = new Properties();
-        for (Map.Entry<String, String> entry : kafkaConfig.toMap().entrySet()) 
{
-            String key = entry.getKey();
-            String value = entry.getValue();
-            if (key.startsWith(PROPERTIES_PREFIX)) {
-                properties.put(key.substring(PROPERTIES_PREFIX.length()), 
value);
-            }
-        }
+        Properties properties = createKafkaProperties(kafkaConfig);
 
         StartupMode startupMode =
                 
fromOption(kafkaConfig.get(KafkaConnectorOptions.SCAN_STARTUP_MODE));
@@ -240,15 +233,7 @@ 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);
-            }
-        }
+        Properties props = createKafkaProperties(kafkaConfig);
 
         props.put(
                 ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,
@@ -282,6 +267,18 @@ public class KafkaActionUtils {
         return new KafkaConsumerWrapper(consumer, topic);
     }
 
+    private static Properties createKafkaProperties(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);
+            }
+        }
+        return props;
+    }
+
     private static class KafkaConsumerWrapper implements 
MessageQueueSchemaUtils.ConsumerWrapper {
 
         private final KafkaConsumer<String, String> consumer;