[incubator-paimon] branch master updated: [flink][refactor] Remove duplicate code in MySqlSyncTableAction (#1098)

[yuzelin]

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

yuzelin 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 024751c6a [flink][refactor] Remove duplicate code in 
MySqlSyncTableAction (#1098)
024751c6a is described below

commit 024751c6a8479874c0d2c84287df3f6d0db26975
Author: yuzelin <33053040+yuze...@users.noreply.github.com>
AuthorDate: Tue May 9 10:37:30 2023 +0800

    [flink][refactor] Remove duplicate code in MySqlSyncTableAction (#1098)
---
 .../action/cdc/mysql/MySqlSyncTableAction.java     | 36 +++++-----------------
 1 file changed, 8 insertions(+), 28 deletions(-)

diff --git 
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/cdc/mysql/MySqlSyncTableAction.java
 
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/cdc/mysql/MySqlSyncTableAction.java
index 13f62d480..855a69ee4 100644
--- 
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/cdc/mysql/MySqlSyncTableAction.java
+++ 
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/cdc/mysql/MySqlSyncTableAction.java
@@ -46,7 +46,6 @@ import java.time.ZoneId;
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Collections;
-import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.Optional;
@@ -54,6 +53,7 @@ import java.util.regex.Matcher;
 import java.util.regex.Pattern;
 import java.util.stream.Collectors;
 
+import static org.apache.paimon.flink.action.Action.getConfigMap;
 import static org.apache.paimon.utils.Preconditions.checkArgument;
 
 /**
@@ -263,43 +263,23 @@ public class MySqlSyncTableAction implements Action {
                             .collect(Collectors.toList());
         }
 
-        Map<String, String> mySqlConfig = getConfigMap(params, "mysql-conf");
-        Map<String, String> catalogConfig = getConfigMap(params, 
"catalog-conf");
-        Map<String, String> tableConfig = getConfigMap(params, "table-conf");
-        if (mySqlConfig == null) {
+        Optional<Map<String, String>> mySqlConfig = getConfigMap(params, 
"mysql-conf");
+        Optional<Map<String, String>> catalogConfig = getConfigMap(params, 
"catalog-conf");
+        Optional<Map<String, String>> tableConfig = getConfigMap(params, 
"table-conf");
+        if (!mySqlConfig.isPresent()) {
             return Optional.empty();
         }
 
         return Optional.of(
                 new MySqlSyncTableAction(
-                        mySqlConfig,
+                        mySqlConfig.get(),
                         tablePath.f0,
                         tablePath.f1,
                         tablePath.f2,
                         partitionKeys,
                         primaryKeys,
-                        catalogConfig == null ? Collections.emptyMap() : 
catalogConfig,
-                        tableConfig == null ? Collections.emptyMap() : 
tableConfig));
-    }
-
-    private static Map<String, String> getConfigMap(MultipleParameterTool 
params, String key) {
-        if (!params.has(key)) {
-            return null;
-        }
-
-        Map<String, String> map = new HashMap<>();
-        for (String param : params.getMultiParameter(key)) {
-            String[] kv = param.split("=");
-            if (kv.length == 2) {
-                map.put(kv[0], kv[1]);
-                continue;
-            }
-
-            System.err.println(
-                    "Invalid " + key + " " + param + ".\nRun mysql-sync-table 
--help for help.");
-            return null;
-        }
-        return map;
+                        catalogConfig.orElse(Collections.emptyMap()),
+                        tableConfig.orElse(Collections.emptyMap())));
     }
 
     private static void printHelp() {