fix mysql async side fill data exception(IndexOutOfBoundsException)

Author: zoudaokoulife

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

@@ -109,6 +109,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

@@ -155,6 +155,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{