[Fix-2890] Fix 'set xxx = xxx' in flinksql task has no effect (DataLi…

…nkDC#2932) Co-authored-by: wenmo <[email protected]>
aiwenmo · Jan 6, 2024 · 1029281 · 1029281
1 parent 944b88d
commit 1029281
Show file tree

Hide file tree

Showing 4 changed files with 12 additions and 5 deletions.
diff --git a/dinky-admin/src/main/resources/dinky-loader/FlinkConfClass b/dinky-admin/src/main/resources/dinky-loader/FlinkConfClass
@@ -25,6 +25,7 @@ org.apache.flink.configuration.HistoryServerOptions
 org.apache.flink.configuration.MetricOptions
 org.apache.flink.configuration.NettyShuffleEnvironmentOptions
 org.apache.flink.configuration.RestartStrategyOptions
+org.apache.flink.streaming.api.environment.ExecutionCheckpointingOptions
 org.apache.flink.yarn.configuration.YarnConfigOptions
 org.apache.flink.kubernetes.configuration.KubernetesConfigOptions
-org.dinky.constant.CustomerConfigureOptions
+org.dinky.constant.CustomerConfigureOptions
diff --git a/dinky-core/src/main/java/org/dinky/job/JobConfig.java b/dinky-core/src/main/java/org/dinky/job/JobConfig.java
@@ -28,6 +28,7 @@
 import org.dinky.gateway.enums.SavePointStrategy;
 import org.dinky.gateway.model.FlinkClusterConfig;
 
+import org.apache.flink.configuration.Configuration;
 import org.apache.flink.configuration.CoreOptions;
 import org.apache.flink.configuration.RestOptions;
 
@@ -237,6 +238,13 @@ public void addGatewayConfig(Map<String, String> config) {
         }
     }
 
+    public void addGatewayConfig(Configuration config) {
+        if (Asserts.isNull(gatewayConfig)) {
+            gatewayConfig = new GatewayConfig();
+        }
+        gatewayConfig.getFlinkConfig().getConfiguration().putAll(config.toMap());
+    }
+
     public boolean isUseRemote() {
         return useRemote || !GatewayType.LOCAL.equalsValue(type);
     }

diff --git a/dinky-core/src/main/java/org/dinky/job/builder/JobTransBuilder.java b/dinky-core/src/main/java/org/dinky/job/builder/JobTransBuilder.java
@@ -183,8 +183,8 @@ private GatewayResult submitByGateway(List<String> inserts) {
 
         GatewayResult gatewayResult = null;
 
-        // Use gateway need to build gateway config, include flink configeration.
-        config.addGatewayConfig(executor.getSetConfig());
+        // Use gateway need to build gateway config, include flink configuration.
+        config.addGatewayConfig(executor.getCustomTableEnvironment().getConfig().getConfiguration());
 
         if (runMode.isApplicationMode()) {
             // Application mode need to submit dinky-app.jar that in the hdfs or image.

diff --git a/dinky-gateway/src/main/java/org/dinky/gateway/config/FlinkConfig.java b/dinky-gateway/src/main/java/org/dinky/gateway/config/FlinkConfig.java
@@ -107,8 +107,6 @@ public class FlinkConfig {
 
     private static final ObjectMapper mapper = new ObjectMapper();
 
-    public static final String DEFAULT_SAVEPOINT_PREFIX = "hdfs:///flink/savepoints/";
-
     public FlinkConfig() {}
 
     public FlinkConfig(Map<String, String> configuration) {