From ecc9b8b517eaa2811dacffdda8a95b646eb953e2 Mon Sep 17 00:00:00 2001
From: AUB <aubdiy@126.com>
Date: Sat, 12 Nov 2022 14:26:59 +0800
Subject: [PATCH] [Feat-#1392][core] add parameters in json format (#1393)

---
 .../main/java/com/dtstack/chunjun/Main.java   |  2 +-
 .../com/dtstack/chunjun/options/Options.java  | 13 +++++++++++-
 .../com/dtstack/chunjun/util/JobUtil.java     | 20 ++++++++++++++-----
 3 files changed, 28 insertions(+), 7 deletions(-)

diff --git a/chunjun-core/src/main/java/com/dtstack/chunjun/Main.java b/chunjun-core/src/main/java/com/dtstack/chunjun/Main.java
index adf8e02b01..34963b12fc 100644
--- a/chunjun-core/src/main/java/com/dtstack/chunjun/Main.java
+++ b/chunjun-core/src/main/java/com/dtstack/chunjun/Main.java
@@ -101,7 +101,7 @@ public static void main(String[] args) throws Exception {
 
         Options options = new OptionParser(args).getOptions();
         String job = URLDecoder.decode(options.getJob(), StandardCharsets.UTF_8.name());
-        String replacedJob = JobUtil.replaceJobParameter(options.getP(), job);
+        String replacedJob = JobUtil.replaceJobParameter(options.getP(), options.getPj(), job);
         Properties confProperties = PropertiesUtil.parseConf(options.getConfProp());
         StreamExecutionEnvironment env = EnvFactory.createStreamExecutionEnvironment(options);
         StreamTableEnvironment tEnv =
diff --git a/chunjun-core/src/main/java/com/dtstack/chunjun/options/Options.java b/chunjun-core/src/main/java/com/dtstack/chunjun/options/Options.java
index 831533a430..de82f1bc86 100644
--- a/chunjun-core/src/main/java/com/dtstack/chunjun/options/Options.java
+++ b/chunjun-core/src/main/java/com/dtstack/chunjun/options/Options.java
@@ -70,9 +70,12 @@ public class Options {
     @OptionRequired(description = "env properties")
     private String confProp = "{}";
 
-    @OptionRequired(description = "json modify")
+    @OptionRequired(description = "parameters in simple format")
     private String p = "";
 
+    @OptionRequired(description = "parameters in json format")
+    private String pj = "";
+
     @OptionRequired(description = "plugin load mode, by classpath or shipfile")
     private String pluginLoadMode = "shipfile";
 
@@ -176,6 +179,14 @@ public void setP(String p) {
         this.p = p;
     }
 
+    public String getPj() {
+        return pj;
+    }
+
+    public void setPj(String pj) {
+        this.pj = pj;
+    }
+
     public String getPluginLoadMode() {
         return pluginLoadMode;
     }
diff --git a/chunjun-core/src/main/java/com/dtstack/chunjun/util/JobUtil.java b/chunjun-core/src/main/java/com/dtstack/chunjun/util/JobUtil.java
index 1d0b80326e..84cab8b7c7 100644
--- a/chunjun-core/src/main/java/com/dtstack/chunjun/util/JobUtil.java
+++ b/chunjun-core/src/main/java/com/dtstack/chunjun/util/JobUtil.java
@@ -20,6 +20,8 @@
 
 import com.dtstack.chunjun.constants.ConstantValue;
 
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.core.type.TypeReference;
+
 import org.apache.commons.lang3.StringUtils;
 
 import java.util.HashMap;
@@ -31,15 +33,19 @@ private JobUtil() throws IllegalAccessException {
         throw new IllegalAccessException(getClass().getName() + " can not be instantiated");
     }
 
-    public static String replaceJobParameter(String p, String job) {
+    public static String replaceJobParameter(String p, String pj, String job) {
         if (StringUtils.isNotBlank(p)) {
-            Map<String, String> parameters = CommandTransform(p);
-            job = JsonValueReplace(job, parameters);
+            Map<String, String> parameters = commandSimpleTransform(p);
+            job = jsonValueReplace(job, parameters);
+        }
+        if (StringUtils.isNotBlank(pj)) {
+            Map<String, String> parameters = commandJsonTransform(pj);
+            job = jsonValueReplace(job, parameters);
         }
         return job;
     }
 
-    public static String JsonValueReplace(String json, Map<String, String> parameter) {
+    public static String jsonValueReplace(String json, Map<String, String> parameter) {
         for (String item : parameter.keySet()) {
             if (json.contains("${" + item + "}")) {
                 json = json.replace("${" + item + "}", parameter.get(item));
@@ -49,7 +55,7 @@ public static String JsonValueReplace(String json, Map<String, String> parameter
     }
 
     /** 将命令行中的修改命令转化为HashMap保存 */
-    public static HashMap<String, String> CommandTransform(String command) {
+    public static HashMap<String, String> commandSimpleTransform(String command) {
         HashMap<String, String> parameter = new HashMap<>();
         String[] split = StringUtils.split(command, ConstantValue.COMMA_SYMBOL);
         for (String item : split) {
@@ -58,4 +64,8 @@ public static HashMap<String, String> CommandTransform(String command) {
         }
         return parameter;
     }
+
+    public static HashMap<String, String> commandJsonTransform(String command) {
+        return JsonUtil.toObject(command, new TypeReference<HashMap<String, String>>() {});
+    }
 }