add queue parameter

Author: todd5167

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

@@ -60,7 +60,8 @@ public class LauncherMain {
 
     private static String getLocalCoreJarPath(String localSqlRootJar) throws Exception {
         String jarPath = PluginUtil.getCoreJarFileName(localSqlRootJar, CORE_JAR);
-        return jarPath;
+        String corePath = localSqlRootJar + SP + jarPath;
+        return corePath;
     }
 
     public static void main(String[] args) throws Exception {

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

@@ -67,6 +67,8 @@ public class LauncherOptionParser {
 
     public static final String OPTION_FLINK_JAR_PATH = "flinkJarPath";
 
+    public static final String OPTION_QUEUE = "queue";
+
     private Options options = new Options();
 
     private BasicParser parser = new BasicParser();
@@ -87,6 +89,7 @@ public LauncherOptionParser(String[] args) {
         options.addOption(OPTION_SAVE_POINT_PATH, true, "Savepoint restore path");
         options.addOption(OPTION_ALLOW_NON_RESTORED_STATE, true, "Flag indicating whether non restored state is allowed if the savepoint");
         options.addOption(OPTION_FLINK_JAR_PATH, true, "flink jar path for submit of perjob mode");
+        options.addOption(OPTION_QUEUE, true, "flink jar path for submit of perjob mode");
 
         try {
             CommandLine cl = parser.parse(options, args);
@@ -145,6 +148,10 @@ public LauncherOptionParser(String[] args) {
                 properties.setFlinkJarPath(flinkJarPath);
             }
 
+            String queue = cl.getOptionValue(OPTION_QUEUE);
+            if(StringUtils.isNotBlank(queue)){
+                properties.setQueue(queue);
+            }
         } catch (Exception e) {
             throw new RuntimeException(e);
         }
@@ -161,7 +168,8 @@ public List<String> getProgramExeArgList() throws Exception {
             String key = one.getKey();
             if(OPTION_FLINK_CONF_DIR.equalsIgnoreCase(key)
                     || OPTION_YARN_CONF_DIR.equalsIgnoreCase(key)
-                    || OPTION_FLINK_JAR_PATH.equalsIgnoreCase(key)){
+                    || OPTION_FLINK_JAR_PATH.equalsIgnoreCase(key)
+                    || OPTION_QUEUE.equalsIgnoreCase(key)){
                 continue;
             }