Merge branch 'v1.4.1'

# Conflicts:
#	README.md

Author: zoudaokoulife

README.md

@@ -120,7 +120,7 @@ sh submit.sh -sql D:\sideSql.txt  -name xctest -remoteSqlPluginPath /opt/dtstack
 * **allowNonRestoredState**
 	* 描述：指示保存点是否允许非还原状态的标志
 	* 必选：否
-	* 默认值：false	
+	* 默认值：false
 
 ## 2 结构
 ### 2.1 源表插件

core/src/main/java/com/dtstack/flink/sql/Main.java

@@ -105,6 +105,9 @@ public static void main(String[] args) throws Exception {
         options.addOption("confProp", true, "env properties");
         options.addOption("mode", true, "deploy mode");
 
+        options.addOption("savePointPath", true, "Savepoint restore path");
+        options.addOption("allowNonRestoredState", true, "Flag indicating whether non restored state is allowed if the savepoint");
+
         CommandLineParser parser = new DefaultParser();
         CommandLine cl = parser.parse(options, args);
         String sql = cl.getOptionValue("sql");

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

@@ -105,6 +105,7 @@ public LauncherOptionParser(String[] args) {
                 Preconditions.checkNotNull(remotePlugin);
                 properties.setRemoteSqlPluginPath(remotePlugin);
             }
+
             String name = Preconditions.checkNotNull(cl.getOptionValue(OPTION_NAME));
             properties.setName(name);
             String addJar = cl.getOptionValue(OPTION_ADDJAR);
@@ -153,6 +154,11 @@ public List<String> getProgramExeArgList() throws Exception {
                     || OPTION_YARN_CONF_DIR.equalsIgnoreCase(key)){
                 continue;
             }
+
+            if(one.getValue() == null){
+                continue;
+            }
+
             args.add("-" + key);
             args.add(one.getValue().toString());
         }

mysql/mysql-side/mysql-async-side/src/main/java/com/dtstack/flink/sql/side/mysql/MysqlAsyncReqRow.java

@@ -191,7 +191,7 @@ public Row fillData(Row input, Object line){
             row.setField(entry.getKey(), obj);
         }
 
-        for(Map.Entry<Integer, Integer> entry : sideInfo.getInFieldIndex().entrySet()){
+        for(Map.Entry<Integer, Integer> entry : sideInfo.getSideFieldIndex().entrySet()){
             if(jsonArray == null){
                 row.setField(entry.getKey(), null);
             }else{