add udf jar

Author: todd5167

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

@@ -24,6 +24,7 @@
 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;
@@ -35,9 +36,7 @@
 import java.net.MalformedURLException;
 import java.net.URL;
 import java.net.URLDecoder;
-import java.util.Map;
-import java.util.Properties;
-import java.util.Set;
+import java.util.*;
 
 /**
  * per job mode submitter
@@ -52,9 +51,18 @@ public class PerJobSubmitter {
 
     public static String submit(LauncherOptions launcherOptions, JobGraph jobGraph) throws Exception {
 
-        fillJobGraphClassPath(jobGraph);
+		fillJobGraphClassPath(jobGraph);
 
-        String confProp = launcherOptions.getConfProp();
+		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);
@@ -77,32 +85,40 @@ public static String submit(LauncherOptions launcherOptions, JobGraph jobGraph)
         return applicationId;
     }
 
-    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));
-        }
-    }
+	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));
+		}
+	}
 }