[hotfix-744][client] fix directory is not configured NPE

Author: zhujinming

flinkx-clients/src/main/java/com/dtstack/flinkx/client/Launcher.java

@@ -119,7 +119,7 @@ public static void main(String[] args) throws Exception {
         clusterClientHelper.submit(jobDeployer);
     }
 
-    private static void findDefaultConfigDir(Options launcherOptions) {
+    private static void findDefaultConfigDir(Options launcherOptions) throws ClusterDeploymentException {
         findDefaultFlinkxDistDir(launcherOptions);
 
         if (ClusterMode.local.name().equalsIgnoreCase(launcherOptions.getMode())) {
@@ -165,7 +165,7 @@ private static void findDefaultFlinkConf(Options launcherOptions) {
         }
     }
 
-    private static void findDefaultFlinkxDistDir(Options launcherOptions) {
+    private static void findDefaultFlinkxDistDir(Options launcherOptions) throws ClusterDeploymentException {
         String distDir = launcherOptions.getFlinkxDistDir();
         if (StringUtils.isEmpty(distDir)) {
             String flinkxHome = getSystemProperty(KEY_FLINKX_HOME);
@@ -180,6 +180,9 @@ private static void findDefaultFlinkxDistDir(Options launcherOptions) {
                 launcherOptions.setFlinkxDistDir(distDir);
             }
         }
+        if (StringUtils.isEmpty(distDir)) {
+            notConfiguredException(KEY_FLINKX_HOME);
+        }
         System.setProperty(ConfigConstants.ENV_FLINK_PLUGINS_DIR, distDir);
     }
 
@@ -191,4 +194,8 @@ private static String getSystemProperty(String name) {
 
         return property;
     }
+
+    private static void notConfiguredException(String propertyKey) throws ClusterDeploymentException {
+        throw new ClusterDeploymentException(propertyKey + " is not configured.");
+    }
 }