Merge branch 'v1.5.0_dev_addqueue' into 'v1.5.0_dev'

add queue parameter



See merge request !15

Author: zoudaokoulife

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 runing yarn queue");
 
         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;
             }
 

launcher/src/main/java/com/dtstack/flink/sql/launcher/perjob/PerJobSubmitter.java

@@ -20,15 +20,23 @@
 
 import com.dtstack.flink.sql.launcher.LauncherOptions;
 import com.dtstack.flink.sql.util.PluginUtil;
+import org.apache.commons.io.Charsets;
+import org.apache.commons.lang3.StringUtils;
 import org.apache.flink.client.deployment.ClusterSpecification;
 import org.apache.flink.client.program.ClusterClient;
+import org.apache.flink.core.fs.Path;
 import org.apache.flink.runtime.jobgraph.JobGraph;
+import org.apache.flink.shaded.guava18.com.google.common.base.Strings;
+import org.apache.flink.shaded.guava18.com.google.common.collect.Sets;
 import org.apache.flink.yarn.AbstractYarnClusterDescriptor;
 import org.apache.hadoop.yarn.api.records.ApplicationId;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
-import java.util.Properties;
+import java.net.MalformedURLException;
+import java.net.URL;
+import java.net.URLDecoder;
+import java.util.*;
 
 /**
  * per job mode submitter
@@ -43,7 +51,20 @@ public class PerJobSubmitter {
 
     public static String submit(LauncherOptions launcherOptions, JobGraph jobGraph) throws Exception {
 
-        Properties confProperties = PluginUtil.jsonStrToObject(launcherOptions.getConfProp(), Properties.class);
+		fillJobGraphClassPath(jobGraph);
+
+		String addjarPath = URLDecoder.decode(launcherOptions.getAddjar(), Charsets.UTF_8.toString());
+		if (StringUtils.isNotBlank(addjarPath) ){
+			List<String>  paths = getJarPaths(addjarPath);
+			paths.forEach( path ->{
+				jobGraph.addJar(new Path("file://" + path));
+			});
+
+		}
+
+		String confProp = launcherOptions.getConfProp();
+        confProp = URLDecoder.decode(confProp, Charsets.UTF_8.toString());
+        Properties confProperties = PluginUtil.jsonStrToObject(confProp, Properties.class);
         ClusterSpecification clusterSpecification = FLinkPerJobResourceUtil.createClusterSpecification(confProperties);
 
         PerJobClusterClientBuilder perJobClusterClientBuilder = new PerJobClusterClientBuilder();
@@ -63,4 +84,41 @@ public static String submit(LauncherOptions launcherOptions, JobGraph jobGraph)
 
         return applicationId;
     }
+
+	private static List<String> getJarPaths(String addjarPath) {
+		if (addjarPath.length() > 2) {
+			addjarPath = addjarPath.substring(1,addjarPath.length()-1).replace("\"","");
+		}
+		List<String> paths = Arrays.asList(addjarPath.split(","));
+		return paths;
+	}
+
+	private static void fillJobGraphClassPath(JobGraph jobGraph) throws MalformedURLException {
+		Map<String, String> jobCacheFileConfig = jobGraph.getJobConfiguration().toMap();
+		Set<String> classPathKeySet = Sets.newHashSet();
+
+		for(Map.Entry<String, String> tmp : jobCacheFileConfig.entrySet()){
+			if(Strings.isNullOrEmpty(tmp.getValue())){
+				continue;
+			}
+
+			if(tmp.getValue().startsWith("class_path")){
+				//DISTRIBUTED_CACHE_FILE_NAME_1
+				//DISTRIBUTED_CACHE_FILE_PATH_1
+				String key = tmp.getKey();
+				String[] array = key.split("_");
+				if(array.length < 5){
+					continue;
+				}
+
+				array[3] = "PATH";
+				classPathKeySet.add(StringUtils.join(array, "_"));
+			}
+		}
+
+		for(String key : classPathKeySet){
+			String pathStr = jobCacheFileConfig.get(key);
+			jobGraph.getClasspaths().add(new URL("file:" + pathStr));
+		}
+	}
 }