[Fix-556][adimin,gateway] Fix flinkLibPath in K8S configuration is null

dlink-admin/src/main/java/com/dlink/service/impl/ClusterConfigurationServiceImpl.java

@@ -50,18 +50,16 @@ public class ClusterConfigurationServiceImpl extends SuperServiceImpl<ClusterCon
         GatewayConfig gatewayConfig = new GatewayConfig();
         if (config.containsKey("hadoopConfigPath")) {
             gatewayConfig.setClusterConfig(ClusterConfig.build(config.get("flinkConfigPath").toString(),
-                    config.get("flinkLibPath").toString(),
-                    config.get("hadoopConfigPath").toString()));
+                config.get("flinkLibPath").toString(),
+                config.get("hadoopConfigPath").toString()));
         } else {
-            gatewayConfig.setClusterConfig(ClusterConfig.build(config.get("flinkConfigPath").toString(),
-                    config.get("flinkLibPath").toString(),
-                    ""));
+            gatewayConfig.setClusterConfig(ClusterConfig.build(config.get("flinkConfigPath").toString()));
         }
         if (config.containsKey("flinkConfig")) {
             gatewayConfig.setFlinkConfig(FlinkConfig.build((Map<String, String>) config.get("flinkConfig")));
         }
         if (Asserts.isEqualsIgnoreCase(clusterConfiguration.getType(), "Yarn")) {
-            gatewayConfig.setType(GatewayType.YARN_PER_JOB);
+            gatewayConfig.setType(GatewayType.YARN_APPLICATION);
         } else if (Asserts.isEqualsIgnoreCase(clusterConfiguration.getType(), "Kubernetes")) {
             gatewayConfig.setType(GatewayType.KUBERNETES_APPLICATION);
             Map kubernetesConfig = (Map) config.get("kubernetesConfig");

dlink-gateway/src/main/java/com/dlink/gateway/config/ClusterConfig.java

@@ -20,12 +20,20 @@ public class ClusterConfig {
     public ClusterConfig() {
     }
 
+    public ClusterConfig(String flinkConfigPath) {
+        this.flinkConfigPath = flinkConfigPath;
+    }
+
     public ClusterConfig(String flinkConfigPath, String flinkLibPath, String yarnConfigPath) {
         this.flinkConfigPath = flinkConfigPath;
         this.flinkLibPath = flinkLibPath;
         this.yarnConfigPath = yarnConfigPath;
     }
 
+    public static ClusterConfig build(String flinkConfigPath) {
+        return new ClusterConfig(flinkConfigPath);
+    }
+
     public static ClusterConfig build(String flinkConfigPath, String flinkLibPath, String yarnConfigPath) {
         return new ClusterConfig(flinkConfigPath, flinkLibPath, yarnConfigPath);
     }
@@ -33,10 +41,10 @@ public class ClusterConfig {
     @Override
     public String toString() {
         return "ClusterConfig{" +
-                "flinkConfigPath='" + flinkConfigPath + '\'' +
-                ", flinkLibPath='" + flinkLibPath + '\'' +
-                ", yarnConfigPath='" + yarnConfigPath + '\'' +
-                ", appId='" + appId + '\'' +
-                '}';
+            "flinkConfigPath='" + flinkConfigPath + '\'' +
+            ", flinkLibPath='" + flinkLibPath + '\'' +
+            ", yarnConfigPath='" + yarnConfigPath + '\'' +
+            ", appId='" + appId + '\'' +
+            '}';
     }
 }

dlink-gateway/src/main/java/com/dlink/gateway/kubernetes/KubernetesGateway.java

@@ -131,11 +131,6 @@ public abstract class KubernetesGateway extends AbstractGateway {
         if (Asserts.isNull(config.getFlinkConfig().getJobId())) {
             throw new GatewayException("No job id was specified. Please specify a job to which you would like to savepont.");
         }
-        if (Asserts.isNotNullString(config.getClusterConfig().getYarnConfigPath())) {
-            configuration = GlobalConfiguration.loadConfiguration(config.getClusterConfig().getYarnConfigPath());
-        } else {
-            configuration = new Configuration();
-        }
         SavePointResult result = SavePointResult.build(getType());
         configuration.set(KubernetesConfigOptions.CLUSTER_ID, config.getClusterConfig().getAppId());
         KubernetesClusterClientFactory clusterClientFactory = new KubernetesClusterClientFactory();