Merge branch '1.8_modify_params' into 'v1.8.0_dev'

调整savepointpath参数位置



See merge request !161

Author: toutian

README.md

@@ -128,6 +128,8 @@ sh submit.sh -sql D:\sideSql.txt  -name xctest -remoteSqlPluginPath /opt/dtstack
         * taskmanager.memory.mb: per_job模式下指定taskmanager的内存大小(单位MB, 默认值:768)
         * taskmanager.num: per_job模式下指定taskmanager的实例数(默认1)
         * taskmanager.slots：per_job模式下指定每个taskmanager对应的slot数量(默认1)
+        * savePointPath：任务恢复点的路径（默认无）
+        * allowNonRestoredState：指示保存点是否允许非还原状态的标志（默认false）
         * [prometheus 相关参数](docs/prometheus.md) per_job可指定metric写入到外部监控组件,以prometheus pushgateway举例
 
 
@@ -141,16 +143,6 @@ sh submit.sh -sql D:\sideSql.txt  -name xctest -remoteSqlPluginPath /opt/dtstack
 	* 必选：否
 	* 默认值：无
 
-* **savePointPath**
-	* 描述：任务恢复点的路径
-	* 必选：否
-	* 默认值：无	
-	
-* **allowNonRestoredState**
-	* 描述：指示保存点是否允许非还原状态的标志
-	* 必选：否
-	* 默认值：false
-	
 * **flinkJarPath**
 	* 描述：per_job 模式提交需要指定本地的flink jar存放路径
 	* 必选：否

core/src/main/java/com/dtstack/flink/sql/constrant/ConfigConstrant.java

@@ -47,8 +47,9 @@ public class ConfigConstrant {
 
     public static final String SQL_MAX_ENV_PARALLELISM = "sql.max.env.parallelism";
 
-    public static final String MR_JOB_PARALLELISM = "mr.job.parallelism";
-    
+    public static final String SAVE_POINT_PATH_KEY = "savePointPath";
+    public static final String ALLOW_NON_RESTORED_STATE_KEY = "allowNonRestoredState";
+
     public static final String SQL_BUFFER_TIMEOUT_MILLIS = "sql.buffer.timeout.millis";
 
     public static final String FLINK_TIME_CHARACTERISTIC_KEY = "time.characteristic";

core/src/main/java/com/dtstack/flink/sql/option/Options.java

@@ -57,12 +57,6 @@ public class Options {
     @OptionRequired(description = "sql ref prop,eg specify event time")
     private  String confProp = "{}";
 
-    @OptionRequired(description = "Savepoint restore path")
-    private  String savePointPath;
-
-    @OptionRequired(description = "Flag indicating whether non restored state is allowed if the savepoint")
-    private  String allowNonRestoredState = "false";
-
     @OptionRequired(description = "flink jar path for submit of perjob mode")
     private String flinkJarPath;
 
@@ -147,22 +141,6 @@ public void setConfProp(String confProp) {
         this.confProp = confProp;
     }
 
-    public String getSavePointPath() {
-        return savePointPath;
-    }
-
-    public void setSavePointPath(String savePointPath) {
-        this.savePointPath = savePointPath;
-    }
-
-    public String getAllowNonRestoredState() {
-        return allowNonRestoredState;
-    }
-
-    public void setAllowNonRestoredState(String allowNonRestoredState) {
-        this.allowNonRestoredState = allowNonRestoredState;
-    }
-
     public String getFlinkJarPath() {
         return flinkJarPath;
     }

launcher/src/main/java/com/dtstack/flink/sql/launcher/LauncherMain.java

@@ -20,6 +20,7 @@
 
 package com.dtstack.flink.sql.launcher;
 
+import com.dtstack.flink.sql.constrant.ConfigConstrant;
 import com.google.common.collect.Lists;
 import com.alibaba.fastjson.JSON;
 import com.alibaba.fastjson.TypeReference;
@@ -29,6 +30,7 @@
 import com.dtstack.flink.sql.option.OptionParser;
 import com.dtstack.flink.sql.option.Options;
 import com.dtstack.flink.sql.util.PluginUtil;
+import org.apache.commons.io.Charsets;
 import org.apache.commons.lang.BooleanUtils;
 import org.apache.commons.lang.StringUtils;
 import org.apache.flink.client.program.ClusterClient;
@@ -44,9 +46,11 @@
 import java.io.FileInputStream;
 import java.io.IOException;
 import java.io.InputStreamReader;
+import java.net.URLDecoder;
 import java.util.LinkedList;
 import java.util.List;
 import java.util.Map;
+import java.util.Properties;
 
 /**
  * Date: 2017/2/20
@@ -75,6 +79,10 @@ public static void main(String[] args) throws Exception {
         String mode = launcherOptions.getMode();
         List<String> argList = optionParser.getProgramExeArgList();
 
+        String confProp = launcherOptions.getConfProp();
+        confProp = URLDecoder.decode(confProp, Charsets.UTF_8.toString());
+        Properties confProperties = PluginUtil.jsonStrToObject(confProp, Properties.class);
+
         if(mode.equals(ClusterMode.local.name())) {
             String[] localArgs = argList.toArray(new String[argList.size()]);
             Main.main(localArgs);
@@ -86,8 +94,10 @@ public static void main(String[] args) throws Exception {
         String[] remoteArgs = argList.toArray(new String[argList.size()]);
         PackagedProgram program = new PackagedProgram(jarFile, Lists.newArrayList(), remoteArgs);
 
-        if(StringUtils.isNotBlank(launcherOptions.getSavePointPath())){
-            program.setSavepointRestoreSettings(SavepointRestoreSettings.forPath(launcherOptions.getSavePointPath(), BooleanUtils.toBoolean(launcherOptions.getAllowNonRestoredState())));
+        String savePointPath = confProperties.getProperty(ConfigConstrant.SAVE_POINT_PATH_KEY);
+        if(StringUtils.isNotBlank(savePointPath)){
+            String allowNonRestoredState = confProperties.getOrDefault(ConfigConstrant.ALLOW_NON_RESTORED_STATE_KEY, "false").toString();
+            program.setSavepointRestoreSettings(SavepointRestoreSettings.forPath(savePointPath, BooleanUtils.toBoolean(allowNonRestoredState)));
         }
 
         if(mode.equals(ClusterMode.yarnPer.name())){