flinkcof empty check

Author: todd5167

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

@@ -83,9 +83,8 @@ public static ClusterClient createStandaloneClient(Options launcherOptions) thro
     }
 
     public static ClusterClient createYarnSessionClient(Options launcherOptions) {
-
-        String flinkConfDir = launcherOptions.getFlinkconf();
-        Configuration config = GlobalConfiguration.loadConfiguration(flinkConfDir);
+        String flinkConfDir = StringUtils.isEmpty(launcherOptions.getFlinkconf()) ? "" : launcherOptions.getFlinkconf();
+        Configuration config = StringUtils.isEmpty(flinkConfDir) ? new Configuration() : GlobalConfiguration.loadConfiguration(flinkConfDir);
         String yarnConfDir = launcherOptions.getYarnconf();
 
         if (StringUtils.isNotBlank(yarnConfDir)) {

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

@@ -92,7 +92,7 @@ public static void main(String[] args) throws Exception {
 
         if(mode.equals(ClusterMode.yarnPer.name())){
             String flinkConfDir = launcherOptions.getFlinkconf();
-            Configuration config = GlobalConfiguration.loadConfiguration(flinkConfDir);
+            Configuration config = StringUtils.isEmpty(flinkConfDir) ? new Configuration() : GlobalConfiguration.loadConfiguration(flinkConfDir);
             JobGraph jobGraph = PackagedProgramUtils.createJobGraph(program, config, 1);
             PerJobSubmitter.submit(launcherOptions, jobGraph, config);
         } else {

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

@@ -69,7 +69,8 @@ public AbstractYarnClusterDescriptor createPerJobClusterDescriptor(Properties co
             throws MalformedURLException {
 
         confProp.forEach((key, val) -> flinkConfig.setString(key.toString(), val.toString()));
-        AbstractYarnClusterDescriptor clusterDescriptor = getClusterDescriptor(flinkConfig, yarnConf, launcherOptions.getFlinkconf());
+        String flinkConf = StringUtils.isEmpty(launcherOptions.getFlinkconf()) ? "" : launcherOptions.getFlinkconf();
+        AbstractYarnClusterDescriptor clusterDescriptor = getClusterDescriptor(flinkConfig, yarnConf, flinkConf);
 
         if (StringUtils.isNotBlank(flinkJarPath)) {
             if (!new File(flinkJarPath).exists()) {