[Optimize][dinky-gateway] The user-defined flink conf path overrides …

…the flink conf path parameter.
DataLinkDC · Jan 6, 2025 · 20d08c0 · 20d08c0
1 parent cf817c0
commit 20d08c0
Show file tree

Hide file tree

Showing 2 changed files with 12 additions and 4 deletions.
diff --git a/dinky-common/src/main/java/org/dinky/data/enums/Status.java b/dinky-common/src/main/java/org/dinky/data/enums/Status.java
@@ -444,8 +444,8 @@ public enum Status {
     /**
      * gateway config
      */
-    GAETWAY_KUBERNETS_TEST_FAILED(180, "gateway.kubernetes.test.failed"),
-    GAETWAY_KUBERNETS_TEST_SUCCESS(181, "gateway.kubernetes.test.success"),
+    GATEWAY_KUBERNETES_TEST_FAILED(180, "gateway.kubernetes.test.failed"),
+    GATEWAY_KUBERNETES_TEST_SUCCESS(181, "gateway.kubernetes.test.success"),
 
     /**
      * process

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
@@ -31,9 +31,11 @@
 import org.dinky.gateway.result.TestResult;
 import org.dinky.utils.TextUtil;
 
+import org.apache.commons.lang3.StringUtils;
 import org.apache.flink.configuration.ConfigOption;
 import org.apache.flink.configuration.CoreOptions;
 import org.apache.flink.configuration.DeploymentOptions;
+import org.apache.flink.configuration.DeploymentOptionsInternal;
 import org.apache.flink.configuration.GlobalConfiguration;
 import org.apache.flink.configuration.PipelineOptions;
 import org.apache.flink.kubernetes.KubernetesClusterClientFactory;
@@ -86,6 +88,12 @@ public void init() {
 
     protected void initConfig() {
         flinkConfigPath = config.getClusterConfig().getFlinkConfigPath();
+
+        // The user-defined flink conf path overrides the flink conf path parameter.
+        if (StringUtils.isNotBlank(flinkConfigPath)) {
+            addConfigParas(DeploymentOptionsInternal.CONF_DIR, flinkConfigPath);
+        }
+
         flinkConfig = config.getFlinkConfig();
         String jobName = flinkConfig.getJobName();
         if (TextUtil.isEmpty(jobName)) {
@@ -222,9 +230,9 @@ public TestResult test() {
             }
             return TestResult.success();
         } catch (Exception e) {
-            logger.error(Status.GAETWAY_KUBERNETS_TEST_FAILED.getMessage(), e);
+            logger.error(Status.GATEWAY_KUBERNETES_TEST_FAILED.getMessage(), e);
             return TestResult.fail(
-                    StrFormatter.format("{}:{}", Status.GAETWAY_KUBERNETS_TEST_FAILED.getMessage(), e.getMessage()));
+                    StrFormatter.format("{}:{}", Status.GATEWAY_KUBERNETES_TEST_FAILED.getMessage(), e.getMessage()));
         } finally {
             close();
         }