修复perjob的savepoint不生效问题

zl03 · Jan 15, 2022 · a318721 · a318721
1 parent 237e05c
commit a318721
Show file tree

Hide file tree

Showing 2 changed files with 7 additions and 1 deletion.
diff --git a/dlink-admin/src/main/java/com/dlink/service/impl/StudioServiceImpl.java b/dlink-admin/src/main/java/com/dlink/service/impl/StudioServiceImpl.java
@@ -27,7 +27,6 @@
 import com.dlink.session.SessionInfo;
 import com.dlink.session.SessionPool;
 import com.dlink.utils.RunTimeUtil;
-import com.dlink.utils.UDFUtil;
 import com.fasterxml.jackson.core.JsonProcessingException;
 import com.fasterxml.jackson.databind.JsonNode;
 import com.fasterxml.jackson.databind.ObjectMapper;

diff --git a/dlink-core/src/main/java/com/dlink/job/JobManager.java b/dlink-core/src/main/java/com/dlink/job/JobManager.java
@@ -35,6 +35,7 @@
 import org.apache.flink.configuration.PipelineOptions;
 import org.apache.flink.runtime.jobgraph.JobGraph;
 import org.apache.flink.runtime.jobgraph.SavepointConfigOptions;
+import org.apache.flink.runtime.jobgraph.SavepointRestoreSettings;
 import org.apache.flink.streaming.api.environment.ExecutionCheckpointingOptions;
 import org.apache.flink.table.api.TableResult;
 import org.apache.flink.yarn.configuration.YarnConfigOptions;
@@ -253,6 +254,9 @@ public JobResult executeSql(String statement) {
                         gatewayResult = Gateway.build(config.getGatewayConfig()).submitJar();
                     } else {
                         config.addGatewayConfig(executor.getSetConfig());
+                        if(Asserts.isNotNullString(config.getSavePointPath())) {
+                            jobGraph.setSavepointRestoreSettings(SavepointRestoreSettings.forPath(config.getSavePointPath()));
+                        }
                         gatewayResult = Gateway.build(config.getGatewayConfig()).submitJobGraph(jobGraph);
                     }
                     job.setResult(InsertResult.success(gatewayResult.getAppId()));
@@ -290,6 +294,9 @@ public JobResult executeSql(String statement) {
                         gatewayResult = Gateway.build(config.getGatewayConfig()).submitJar();
                     } else {
                         config.addGatewayConfig(executor.getSetConfig());
+                        if(Asserts.isNotNullString(config.getSavePointPath())) {
+                            jobGraph.setSavepointRestoreSettings(SavepointRestoreSettings.forPath(config.getSavePointPath()));
+                        }
                         gatewayResult = Gateway.build(config.getGatewayConfig()).submitJobGraph(jobGraph);
                     }
                     job.setResult(InsertResult.success(gatewayResult.getAppId()));