[Optimize-4127][gateway] The user-defined flink conf path overrides the flink conf path parameter

Author: aiwenmo

Diff for: 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 for: dinky-core/src/main/java/org/dinky/job/JobConfig.java

@@ -29,7 +29,6 @@
 import org.dinky.gateway.enums.SavePointStrategy;
 import org.dinky.gateway.model.FlinkClusterConfig;
 
-import org.apache.commons.lang3.StringUtils;
 import org.apache.flink.configuration.Configuration;
 import org.apache.flink.configuration.CoreOptions;
 import org.apache.flink.configuration.RestOptions;
@@ -43,14 +42,12 @@
 import lombok.AllArgsConstructor;
 import lombok.Builder;
 import lombok.Data;
-import lombok.extern.slf4j.Slf4j;
 
 /**
  * JobConfig
  *
  * @since 2021/6/27 18:45
  */
-@Slf4j
 @Data
 @Builder
 @AllArgsConstructor
@@ -261,17 +258,9 @@ public void buildGatewayConfig(FlinkClusterConfig config) {
             flinkConfig.getConfiguration().put(customConfig.getName(), customConfig.getValue());
         }
 
-        Map<String, String> configuration = flinkConfig.getConfiguration();
-
-        // In Kubernetes mode, must set jobmanager.memory.process.size.
-        if (StringUtils.isBlank(configuration.get("jobmanager.memory.process.size"))) {
-            log.warn("In Kubernetes mode, please configure 'jobmanager.memory.process.size', default 2048m");
-            configuration.put("jobmanager.memory.process.size", "2048m");
-        }
-
-        // Load job configuration content afterwards
-        configuration.putAll(getConfigJson());
-        configuration.put(CoreOptions.DEFAULT_PARALLELISM.key(), String.valueOf(parallelism));
+        // Load job configuration content afterwords
+        flinkConfig.getConfiguration().putAll(getConfigJson());
+        flinkConfig.getConfiguration().put(CoreOptions.DEFAULT_PARALLELISM.key(), String.valueOf(parallelism));
         flinkConfig.setJobName(getJobName());
 
         gatewayConfig = GatewayConfig.build(config);

Diff for: 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,10 @@ 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 +228,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();
         }