[hotfix][flip6] Only create new terminationFuture if MiniCluster is running
Project: http://git-wip-us.apache.org/repos/asf/flink/repo Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/0caff357 Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/0caff357 Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/0caff357 Branch: refs/heads/master Commit: 0caff357848929b736fca0f78e358f77423ce355 Parents: 9538675 Author: gyao <[email protected]> Authored: Fri Mar 16 06:57:24 2018 +0100 Committer: Till Rohrmann <[email protected]> Committed: Sun Mar 18 15:58:13 2018 +0100 ---------------------------------------------------------------------- .../java/org/apache/flink/runtime/minicluster/MiniCluster.java | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/flink/blob/0caff357/flink-runtime/src/main/java/org/apache/flink/runtime/minicluster/MiniCluster.java ---------------------------------------------------------------------- diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/minicluster/MiniCluster.java b/flink-runtime/src/main/java/org/apache/flink/runtime/minicluster/MiniCluster.java index d660c67..74aa388 100644 --- a/flink-runtime/src/main/java/org/apache/flink/runtime/minicluster/MiniCluster.java +++ b/flink-runtime/src/main/java/org/apache/flink/runtime/minicluster/MiniCluster.java @@ -217,9 +217,6 @@ public class MiniCluster implements JobExecutorService, AutoCloseableAsync { LOG.info("Starting Flink Mini Cluster"); LOG.debug("Using configuration {}", miniClusterConfiguration); - // create a new termination future - terminationFuture = new CompletableFuture<>(); - final Configuration configuration = miniClusterConfiguration.getConfiguration(); final Time rpcTimeout = miniClusterConfiguration.getRpcTimeout(); final int numTaskManagers = miniClusterConfiguration.getNumTaskManagers(); @@ -384,6 +381,9 @@ public class MiniCluster implements JobExecutorService, AutoCloseableAsync { throw e; } + // create a new termination future + terminationFuture = new CompletableFuture<>(); + // now officially mark this as running running = true;
