Merge branch 'feat_1.10_addShipflie' into 'v1.10.0_dev'

[feat] add addShipfile

See merge request dt-insight-engine/flinkStreamSQL!77

Author: FlechazoW

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

@@ -71,6 +71,7 @@ public static JobParamsInfo parseArgs(String[] args) throws Exception {
         String udfJar = launcherOptions.getAddjar();
         String queue = launcherOptions.getQueue();
         String pluginLoadMode = launcherOptions.getPluginLoadMode();
+        String addShipfile = launcherOptions.getAddShipfile();
 
         String yarnSessionConf = URLDecoder.decode(launcherOptions.getYarnSessionConf(), Charsets.UTF_8.toString());
         Properties yarnSessionConfProperties = PluginUtil.jsonStrToObject(yarnSessionConf, Properties.class);
@@ -91,6 +92,7 @@ public static JobParamsInfo parseArgs(String[] args) throws Exception {
                 .setFlinkJarPath(flinkJarPath)
                 .setPluginLoadMode(pluginLoadMode)
                 .setQueue(queue)
+                .setAddShipfile(addShipfile)
                 .build();
     }
 

launcher/src/main/java/com/dtstack/flink/sql/launcher/entity/JobParamsInfo.java

@@ -22,29 +22,31 @@
 import java.util.Properties;
 
 /**
- *   parse the original mission parameters
+ * parse the original mission parameters
  * Date: 2020/3/4
  * Company: www.dtstack.com
+ *
  * @author maqi
  */
 public class JobParamsInfo {
 
-    private String mode;
-    private String name;
-    private String queue;
-    private String localPluginRoot;
-    private String flinkConfDir;
-    private String flinkJarPath;
-    private String yarnConfDir;
-    private String pluginLoadMode;
-    private String udfJar;
-    private String[] execArgs;
-    private Properties confProperties;
-    private Properties yarnSessionConfProperties;
+    private final String mode;
+    private final String name;
+    private final String queue;
+    private final String localPluginRoot;
+    private final String flinkConfDir;
+    private final String flinkJarPath;
+    private final String yarnConfDir;
+    private final String pluginLoadMode;
+    private final String udfJar;
+    private final String[] execArgs;
+    private final Properties confProperties;
+    private final Properties yarnSessionConfProperties;
+    private final String addShipFile;
 
     private JobParamsInfo(String mode, String name, String queue, String localPluginRoot, String flinkConfDir, String yarnConfDir,
                           String pluginLoadMode, String[] execArgs, Properties confProperties, Properties yarnSessionConfProperties,
-                          String udfJar, String flinkJarPath) {
+                          String udfJar, String flinkJarPath, String addShipFile) {
         this.mode = mode;
         this.name = name;
         this.queue = queue;
@@ -57,6 +59,7 @@ private JobParamsInfo(String mode, String name, String queue, String localPlugin
         this.yarnSessionConfProperties = yarnSessionConfProperties;
         this.udfJar = udfJar;
         this.flinkJarPath = flinkJarPath;
+        this.addShipFile = addShipFile;
     }
 
     public String getMode() {
@@ -107,6 +110,10 @@ public String getFlinkJarPath() {
         return flinkJarPath;
     }
 
+    public String getAddShipFile() {
+        return addShipFile;
+    }
+
     public static JobParamsInfo.Builder builder() {
         return new JobParamsInfo.Builder();
     }
@@ -125,6 +132,7 @@ public static class Builder {
         private String udfJar;
         private Properties confProperties;
         private Properties yarnSessionConfProperties;
+        private String addShipfile;
 
         public JobParamsInfo.Builder setMode(String mode) {
             this.mode = mode;
@@ -186,9 +194,15 @@ public JobParamsInfo.Builder setFlinkJarPath(String flinkJarPath) {
             return this;
         }
 
+        public JobParamsInfo.Builder setAddShipfile(String addShipfile) {
+            this.addShipfile = addShipfile;
+            return this;
+        }
+
         public JobParamsInfo build() {
             return new JobParamsInfo(mode, name, queue, localPluginRoot, flinkConfDir,
-                    yarnConfDir, pluginLoadMode, execArgs, confProperties, yarnSessionConfProperties, udfJar, flinkJarPath);
+                    yarnConfDir, pluginLoadMode, execArgs, confProperties,
+                    yarnSessionConfProperties, udfJar, flinkJarPath, addShipfile);
         }
     }
 
@@ -207,6 +221,7 @@ public String toString() {
                 ", execArgs=" + Arrays.toString(execArgs) +
                 ", confProperties=" + confProperties +
                 ", yarnSessionConfProperties=" + yarnSessionConfProperties +
+                ", addShipFile='" + addShipFile + '\'' +
                 '}';
     }
 }

launcher/src/main/java/com/dtstack/flink/sql/launcher/executor/YarnJobClusterExecutor.java

@@ -39,6 +39,7 @@
 import java.io.File;
 import java.net.MalformedURLException;
 import java.util.ArrayList;
+import java.util.Arrays;
 import java.util.List;
 import java.util.Optional;
 
@@ -74,15 +75,23 @@ public void exec() throws Exception {
                 .createClusterDescriptor(jobParamsInfo.getYarnConfDir(), flinkConfiguration);
 
         List<File> shipFiles = getShipFiles(jobParamsInfo.getFlinkJarPath(), jobParamsInfo.getPluginLoadMode(), jobGraph, clusterDescriptor);
+
+        if (jobParamsInfo.getAddShipFile() != null) {
+            List<String> addShipFilesPath = parsePathFromStr(jobParamsInfo.getAddShipFile());
+            for (String path : addShipFilesPath) {
+                shipFiles.addAll(getShipFiles(path, jobParamsInfo.getPluginLoadMode(), jobGraph, clusterDescriptor));
+            }
+        }
+
         clusterDescriptor.addShipFiles(shipFiles);
 
         ClusterSpecification clusterSpecification = YarnClusterClientFactory.INSTANCE.getClusterSpecification(flinkConfiguration);
         ClusterClientProvider<ApplicationId> applicationIdClusterClientProvider = clusterDescriptor.deployJobCluster(clusterSpecification, jobGraph, true);
 
         String applicationId = applicationIdClusterClientProvider.getClusterClient().getClusterId().toString();
         String flinkJobId = jobGraph.getJobID().toString();
-        String tips = String.format("deploy per_job with appId: %s, jobId: %s", applicationId, flinkJobId);
-        System.out.println(tips);
+
+        LOG.info(String.format("deploy per_job with appId: %s, jobId: %s", applicationId, flinkJobId));
     }
 
     private void appendApplicationConfig(Configuration flinkConfig, JobParamsInfo jobParamsInfo) {
@@ -115,9 +124,14 @@ protected List<File> getShipFiles(String flinkJarPath, String pluginLoadMode, Jo
 
     private void dealFlinkLibJar(String flinkJarPath, YarnClusterDescriptor clusterDescriptor, List<File> shipFiles) throws MalformedURLException {
         if (StringUtils.isEmpty(flinkJarPath) || !new File(flinkJarPath).exists()) {
-            throw new RuntimeException("The param '-flinkJarPath' ref dir is not exist");
+            throw new RuntimeException("path " + flinkJarPath + " is not exist");
         }
         File[] jars = new File(flinkJarPath).listFiles();
+
+        if (jars == null || jars.length == 0) {
+            throw new RuntimeException(flinkJarPath + " no file exist !");
+        }
+
         for (File file : jars) {
             if (file.toURI().toURL().toString().contains("flink-dist")) {
                 clusterDescriptor.setLocalJarPath(new Path(file.toURI().toURL().toString()));
@@ -161,5 +175,11 @@ private Optional<File> discoverLogConfigFile(final String configurationDirectory
         return logConfigFile;
     }
 
+    private static List<String> parsePathFromStr(String pathStr) {
+        if (pathStr.length() > 2) {
+            pathStr = pathStr.substring(1, pathStr.length() - 1).replace("\"", "");
+        }
 
+        return Arrays.asList(pathStr.split(","));
+    }
 }