add queue config

Author: yanxi0227

core/pom.xml

@@ -61,13 +61,13 @@
         <dependency>
             <groupId>org.apache.flink</groupId>
             <artifactId>flink-table-planner_2.11</artifactId>
-            <version>1.8.0</version>
+            <version>${flink.version}</version>
         </dependency>
 
         <dependency>
             <groupId>org.apache.flink</groupId>
             <artifactId>flink-table-common</artifactId>
-            <version>1.8.0</version>
+            <version>${flink.version}</version>
         </dependency>
 
         <dependency>

core/src/main/java/com/dtstack/flink/sql/options/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) throws Exception {
         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 runing yarn queue");
             CommandLine cl = parser.parse(options, args);
             String mode = cl.getOptionValue(OPTION_MODE, ClusterMode.local.name());
             //check mode
@@ -145,6 +148,10 @@ public LauncherOptionParser(String[] args) throws Exception {
             if(StringUtils.isNotBlank(flinkJarPath)){
                 properties.setFlinkJarPath(flinkJarPath);
             }
+            String queue = cl.getOptionValue(OPTION_QUEUE);
+            if(StringUtils.isNotBlank(queue)){
+                properties.setQueue(queue);
+            }
     }
 
     public LauncherOptions getLauncherOptions(){
@@ -159,7 +166,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;
             }