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 index efe2711db5..f68b177cc2 100644 --- 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-core/src/main/java/org/dinky/explainer/mock/MockStatementExplainer.java b/dinky-core/src/main/java/org/dinky/explainer/mock/MockStatementExplainer.java index 4106cd7f4b..3bc4c4365c 100644 --- a/dinky-core/src/main/java/org/dinky/explainer/mock/MockStatementExplainer.java +++ b/dinky-core/src/main/java/org/dinky/explainer/mock/MockStatementExplainer.java @@ -165,14 +165,14 @@ private String getSinkMockDdlStatement(String tableName, String columns) { /** * generate table identifier with mocked prefix info - * @param tableName table name + * @param tableIdentifier table identifier * @return table identifier with mocked prefix info */ - private List generateMockedTableIdentifier(String tableName) { + private List generateMockedTableIdentifier(String tableIdentifier) { List names = new ArrayList<>(); names.add("default_catalog"); names.add("default_database"); - names.add("mock_sink_" + tableName); + names.add("mock_sink_" + tableIdentifier.replace(".", "_")); return names; } } diff --git a/dinky-core/src/main/java/org/dinky/job/JobConfig.java b/dinky-core/src/main/java/org/dinky/job/JobConfig.java index 5e812393e0..a46011458e 100644 --- a/dinky-core/src/main/java/org/dinky/job/JobConfig.java +++ b/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 @@ -254,6 +251,7 @@ public ExecutorConfig getExecutorSetting() { public void buildGatewayConfig(FlinkClusterConfig config) { FlinkConfig flinkConfig = config.getFlinkConfig(); + // Prioritize loading custom Flink configuration content in the cluster configuration for (CustomConfig customConfig : flinkConfig.getFlinkConfigList()) { Assert.notNull(customConfig.getName(), "Custom flink config has null key"); @@ -261,17 +259,9 @@ public void buildGatewayConfig(FlinkClusterConfig config) { flinkConfig.getConfiguration().put(customConfig.getName(), customConfig.getValue()); } - Map 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 --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 47cb9f075c..3b2b5ee42b 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 @@ -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(); }