addShipfile bug fix

Author: simenliuxing

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

@@ -79,7 +79,7 @@ public void exec() throws Exception {
         if (jobParamsInfo.getAddShipFile() != null) {
             List<String> addShipFilesPath = parsePathFromStr(jobParamsInfo.getAddShipFile());
             for (String path : addShipFilesPath) {
-                shipFiles.addAll(getShipFiles(path, jobParamsInfo.getPluginLoadMode(), jobGraph, clusterDescriptor));
+                shipFiles.add(getAddShipFile(path));
             }
         }
 
@@ -122,6 +122,18 @@ protected List<File> getShipFiles(String flinkJarPath, String pluginLoadMode, Jo
         return shipFiles;
     }
 
+    private File getAddShipFile(String addShipFile) {
+        if (StringUtils.isEmpty(addShipFile) || !new File(addShipFile).exists()) {
+            throw new RuntimeException("path " + addShipFile + " is not exist");
+        }
+        File shipFile = new File(addShipFile);
+        if (shipFile.isFile()) {
+            return shipFile;
+        } else {
+            throw new RuntimeException("addShipfile only supports file path,E.g : {\\\"a.keytab\\\",\\\"b.txt\\\"}");
+        }
+    }
+
     private void dealFlinkLibJar(String flinkJarPath, YarnClusterDescriptor clusterDescriptor, List<File> shipFiles) throws MalformedURLException {
         if (StringUtils.isEmpty(flinkJarPath) || !new File(flinkJarPath).exists()) {
             throw new RuntimeException("path " + flinkJarPath + " is not exist");