diff --git a/dinky-admin/src/main/resources/dinky-loader/FlinkConfClass b/dinky-admin/src/main/resources/dinky-loader/FlinkConfClass
index fac1e1b8d8..505fcc8769 100644
--- a/dinky-admin/src/main/resources/dinky-loader/FlinkConfClass
+++ b/dinky-admin/src/main/resources/dinky-loader/FlinkConfClass
@@ -32,6 +32,7 @@ org.apache.flink.configuration.YarnConfigOptions
 org.apache.flink.kubernetes.configuration.KubernetesConfigOptions
 org.apache.flink.kubernetes.configuration.KubernetesHighAvailabilityOptions
 org.apache.flink.configuration.KubernetesConfigOptions
+org.apache.flink.configuration.DeploymentOptions
 org.apache.flink.python.PythonOptions
 org.apache.flink.table.api.config.TableConfigOptions
 org.apache.flink.table.api.config.OptimizerConfigOptions
diff --git a/dinky-gateway/src/main/java/org/dinky/gateway/kubernetes/KubernetesGateway.java b/dinky-gateway/src/main/java/org/dinky/gateway/kubernetes/KubernetesGateway.java
index bd48400795..a8a878b342 100644
--- a/dinky-gateway/src/main/java/org/dinky/gateway/kubernetes/KubernetesGateway.java
+++ b/dinky-gateway/src/main/java/org/dinky/gateway/kubernetes/KubernetesGateway.java
@@ -101,9 +101,7 @@ protected void initConfig() {
     }
 
     public SavePointResult savepointCluster(String savePoint) {
-        if (Asserts.isNull(k8sClientHelper.getClient())) {
-            initConfig();
-        }
+        initConfig();
 
         KubernetesClusterClientFactory clusterClientFactory = new KubernetesClusterClientFactory();
         addConfigParas(
@@ -120,9 +118,7 @@ public SavePointResult savepointCluster(String savePoint) {
     }
 
     public SavePointResult savepointJob(String savePoint) {
-        if (Asserts.isNull(k8sClientHelper.getClient())) {
-            initConfig();
-        }
+        initConfig();
         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.");
@@ -168,11 +164,10 @@ public TestResult test() {
 
     @Override
     public void killCluster() {
-        if (Asserts.isNull(k8sClientHelper.getClient())) {
-            initConfig();
-        }
+        log.info("Start kill cluster: " + config.getFlinkConfig().getJobName());
+        initConfig();
         addConfigParas(
-                KubernetesConfigOptions.CLUSTER_ID, config.getClusterConfig().getAppId());
+                KubernetesConfigOptions.CLUSTER_ID, config.getFlinkConfig().getJobName());
         KubernetesClusterClientFactory clusterClientFactory = new KubernetesClusterClientFactory();
         String clusterId = clusterClientFactory.getClusterId(configuration);
         if (Asserts.isNull(clusterId)) {