[hotfix] fix running failure in k8s appliation mode

Author: xiuzhu9527

flinkx-core/src/main/java/com/dtstack/flinkx/Main.java

@@ -71,8 +71,6 @@
 import org.slf4j.LoggerFactory;
 
 import java.net.URL;
-import java.net.URLDecoder;
-import java.nio.charset.StandardCharsets;
 import java.util.Arrays;
 import java.util.List;
 import java.util.Optional;
@@ -95,7 +93,7 @@ public static void main(String[] args) throws Exception {
         LOG.info("-------------------------------------------");
 
         Options options = new OptionParser(args).getOptions();
-        String job = URLDecoder.decode(options.getJob(), StandardCharsets.UTF_8.name());
+        String job = options.getJob();
         Properties confProperties = PropertiesUtil.parseConf(options.getConfProp());
         StreamExecutionEnvironment env = EnvFactory.createStreamExecutionEnvironment(options);
         StreamTableEnvironment tEnv =
@@ -315,7 +313,7 @@ private static void configStreamExecutionEnvironment(
             factoryHelper.setRemotePluginPath(options.getRemoteFlinkxDistDir());
             factoryHelper.setPluginLoadMode(options.getPluginLoadMode());
             factoryHelper.setEnv(env);
-            factoryHelper.setExecutionMode(options.getMode());
+            factoryHelper.setOptions(options);
 
             DirtyConf dirtyConf = DirtyConfUtil.parse(options);
             factoryHelper.registerCachedFile(

flinkx-core/src/main/java/com/dtstack/flinkx/util/FactoryHelper.java

@@ -18,6 +18,7 @@
 package com.dtstack.flinkx.util;
 
 import com.dtstack.flinkx.constants.ConstantValue;
+import com.dtstack.flinkx.options.Options;
 
 import org.apache.flink.configuration.ConfigOption;
 import org.apache.flink.configuration.ConfigOptions;
@@ -60,8 +61,8 @@ public class FactoryHelper {
     protected List<URL> classPathSet = new ArrayList<>();
     /** shipfile需要的jar的classPath index */
     protected int classFileNameIndex = 0;
-    /** 任务执行模式 */
-    protected String executionMode;
+    /** 任务启动参数 */
+    protected Options options;
 
     public FactoryHelper() {}
 
@@ -93,7 +94,7 @@ public void registerCachedFile(
                     this.classFileNameIndex++;
                 }
             }
-            PluginUtil.setPipelineOptionsToEnvConfig(this.env, urlList, executionMode);
+            PluginUtil.setPipelineOptionsToEnvConfig(this.env, urlList, options);
         } catch (Exception e) {
             LOG.warn("can't add jar in {} to cachedFile, e = {}", urlSet, e.getMessage());
         }
@@ -115,11 +116,11 @@ public void setEnv(StreamExecutionEnvironment env) {
         this.env = env;
     }
 
-    public String getExecutionMode() {
-        return executionMode;
+    public Options getOptions() {
+        return options;
     }
 
-    public void setExecutionMode(String executionMode) {
-        this.executionMode = executionMode;
+    public void setOptions(Options options) {
+        this.options = options;
     }
 }

flinkx-core/src/main/java/com/dtstack/flinkx/util/PluginUtil.java

@@ -346,7 +346,7 @@ public static void registerPluginUrlToCachedFile(
                     e);
         }
 
-        config.setSyncJarList(setPipelineOptionsToEnvConfig(env, urlList, options.getMode()));
+        config.setSyncJarList(setPipelineOptionsToEnvConfig(env, urlList, options));
     }
 
     /**
@@ -358,7 +358,7 @@ public static void registerPluginUrlToCachedFile(
      */
     @SuppressWarnings("all")
     public static List<String> setPipelineOptionsToEnvConfig(
-            StreamExecutionEnvironment env, List<String> urlList, String executionMode) {
+            StreamExecutionEnvironment env, List<String> urlList, Options options) {
         try {
             Configuration configuration =
                     (Configuration)
@@ -370,6 +370,7 @@ public static List<String> setPipelineOptionsToEnvConfig(
             jarList.addAll(urlList);
 
             List<String> pipelineJars = new ArrayList();
+            String executionMode = options.getMode();
             LOG.info("Flinkx executionMode: " + executionMode);
             if (ClusterMode.getByName(executionMode) == ClusterMode.kubernetesApplication) {
                 for (String jarUrl : jarList) {