[FLINK-19585][minicluster] Use SavepointRestoreSettings of StreamGrap…

…h while creating JobGraph.

There is currently no way to start from savepoints while using MiniCluster. The cluster config cannot be used (and would be a mismatch) because it's overridden by MiniClusterPipelineExecutorServiceLoader#createConfiguration.
However, the jobgraph that is being generated in MiniClusterPipelineExecutorServiceLoader#MiniClusterExecutor only uses this configuration to translate the Pipeline/StreamGraph. In production code, savepoint restore settings are usually applied last and taken from the StreamGraph.

flink-test-utils-parent/flink-test-utils/src/main/java/org/apache/flink/test/util/MiniClusterPipelineExecutorServiceLoader.java

@@ -31,8 +31,10 @@
 import org.apache.flink.core.fs.FileSystem;
 import org.apache.flink.core.fs.Path;
 import org.apache.flink.runtime.jobgraph.JobGraph;
+import org.apache.flink.runtime.jobgraph.SavepointRestoreSettings;
 import org.apache.flink.runtime.minicluster.MiniCluster;
 import org.apache.flink.runtime.minicluster.MiniClusterJobClient;
+import org.apache.flink.streaming.api.graph.StreamGraph;
 
 import java.io.IOException;
 import java.net.MalformedURLException;
@@ -138,6 +140,9 @@ public CompletableFuture<JobClient> execute(
  Configuration configuration,
  ClassLoader userCodeClassLoader) throws Exception {
  final JobGraph jobGraph = PipelineExecutorUtils.getJobGraph(pipeline, configuration);
+ if (jobGraph.getSavepointRestoreSettings() == SavepointRestoreSettings.none() && pipeline instanceof StreamGraph) {
+ jobGraph.setSavepointRestoreSettings(((StreamGraph) pipeline).getSavepointRestoreSettings());
+ }
  return miniCluster.submitJob(jobGraph)
  .thenApply(result -> new MiniClusterJobClient(
  result.getJobID(),