flink-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From kklou...@apache.org
Subject [2/3] flink git commit: [FLINK-7974][QS] Wait for QS abstract server to shutdown.
Date Wed, 06 Dec 2017 15:11:49 GMT
[FLINK-7974][QS] Wait for QS abstract server to shutdown.


Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/74d052bb
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/74d052bb
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/74d052bb

Branch: refs/heads/master
Commit: 74d052bb045031363652116ab8226d8ac00e0cd0
Parents: 5760677
Author: kkloudas <kkloudas@gmail.com>
Authored: Thu Nov 9 19:30:29 2017 +0100
Committer: kkloudas <kkloudas@gmail.com>
Committed: Wed Dec 6 14:33:22 2017 +0100

----------------------------------------------------------------------
 .../client/program/rest/RestClusterClient.java  |   3 +-
 .../org/apache/flink/util/ExecutorUtils.java    |  77 +++++++++++
 .../MesosApplicationMasterRunner.java           |   3 +-
 .../network/AbstractServerBase.java             |  95 ++++++++++---
 .../network/AbstractServerHandler.java          |  11 +-
 .../client/proxy/KvStateClientProxyHandler.java |  34 +++--
 .../client/proxy/KvStateClientProxyImpl.java    |   8 +-
 .../server/KvStateServerHandler.java            |   4 +-
 .../server/KvStateServerImpl.java               |   8 +-
 .../HAAbstractQueryableStateTestBase.java       |   5 +-
 .../NonHAAbstractQueryableStateTestBase.java    |   4 +-
 .../network/AbstractServerTest.java             | 135 +++++++++++--------
 .../flink/runtime/concurrent/Executors.java     |  49 +------
 .../runtime/taskexecutor/TaskManagerRunner.java |   6 +-
 .../flink/runtime/jobmanager/JobManager.scala   |   6 +-
 .../runtime/minicluster/FlinkMiniCluster.scala  |   6 +-
 .../slotmanager/SlotProtocolTest.java           |   3 +-
 .../handler/legacy/ExecutionGraphCacheTest.java |   4 +-
 .../flink/yarn/YarnApplicationMasterRunner.java |   3 +-
 19 files changed, 306 insertions(+), 158 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/flink/blob/74d052bb/flink-clients/src/main/java/org/apache/flink/client/program/rest/RestClusterClient.java
----------------------------------------------------------------------
diff --git a/flink-clients/src/main/java/org/apache/flink/client/program/rest/RestClusterClient.java
b/flink-clients/src/main/java/org/apache/flink/client/program/rest/RestClusterClient.java
index e21e94b..8c0462d 100644
--- a/flink-clients/src/main/java/org/apache/flink/client/program/rest/RestClusterClient.java
+++ b/flink-clients/src/main/java/org/apache/flink/client/program/rest/RestClusterClient.java
@@ -48,6 +48,7 @@ import org.apache.flink.runtime.rest.messages.job.savepoints.SavepointMessagePar
 import org.apache.flink.runtime.rest.messages.job.savepoints.SavepointTriggerHeaders;
 import org.apache.flink.runtime.rest.messages.job.savepoints.SavepointTriggerResponseBody;
 import org.apache.flink.runtime.util.ExecutorThreadFactory;
+import org.apache.flink.util.ExecutorUtils;
 
 import javax.annotation.Nullable;
 
@@ -90,7 +91,7 @@ public class RestClusterClient extends ClusterClient {
 			log.error("An error occurred during the client shutdown.", e);
 		}
 		this.restClient.shutdown(Time.seconds(5));
-		org.apache.flink.runtime.concurrent.Executors.gracefulShutdown(5, TimeUnit.SECONDS, this.executorService);
+		ExecutorUtils.gracefulShutdown(5, TimeUnit.SECONDS, this.executorService);
 	}
 
 	@Override

http://git-wip-us.apache.org/repos/asf/flink/blob/74d052bb/flink-core/src/main/java/org/apache/flink/util/ExecutorUtils.java
----------------------------------------------------------------------
diff --git a/flink-core/src/main/java/org/apache/flink/util/ExecutorUtils.java b/flink-core/src/main/java/org/apache/flink/util/ExecutorUtils.java
new file mode 100644
index 0000000..d98bdd2
--- /dev/null
+++ b/flink-core/src/main/java/org/apache/flink/util/ExecutorUtils.java
@@ -0,0 +1,77 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.util;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.TimeUnit;
+
+/**
+ * Utilities for {@link java.util.concurrent.Executor Executors}.
+ */
+public class ExecutorUtils {
+
+	private static final Logger LOG = LoggerFactory.getLogger(ExecutorUtils.class);
+
+	/**
+	 * Gracefully shutdown the given {@link ExecutorService}. The call waits the given timeout
that
+	 * all ExecutorServices terminate. If the ExecutorServices do not terminate in this time,
+	 * they will be shut down hard.
+	 *
+	 * @param timeout to wait for the termination of all ExecutorServices
+	 * @param unit of the timeout
+	 * @param executorServices to shut down
+	 */
+	public static void gracefulShutdown(long timeout, TimeUnit unit, ExecutorService... executorServices)
{
+		for (ExecutorService executorService: executorServices) {
+			executorService.shutdown();
+		}
+
+		boolean wasInterrupted = false;
+		final long endTime = unit.toMillis(timeout) + System.currentTimeMillis();
+		long timeLeft = unit.toMillis(timeout);
+		boolean hasTimeLeft = timeLeft > 0L;
+
+		for (ExecutorService executorService: executorServices) {
+			if (wasInterrupted || !hasTimeLeft) {
+				executorService.shutdownNow();
+			} else {
+				try {
+					if (!executorService.awaitTermination(timeLeft, TimeUnit.MILLISECONDS)) {
+						LOG.warn("ExecutorService did not terminate in time. Shutting it down now.");
+						executorService.shutdownNow();
+					}
+				} catch (InterruptedException e) {
+					LOG.warn("Interrupted while shutting down executor services. Shutting all " +
+							"remaining ExecutorServices down now.", e);
+					executorService.shutdownNow();
+
+					wasInterrupted = true;
+
+					Thread.currentThread().interrupt();
+				}
+
+				timeLeft = endTime - System.currentTimeMillis();
+				hasTimeLeft = timeLeft > 0L;
+			}
+		}
+	}
+}

http://git-wip-us.apache.org/repos/asf/flink/blob/74d052bb/flink-mesos/src/main/java/org/apache/flink/mesos/runtime/clusterframework/MesosApplicationMasterRunner.java
----------------------------------------------------------------------
diff --git a/flink-mesos/src/main/java/org/apache/flink/mesos/runtime/clusterframework/MesosApplicationMasterRunner.java
b/flink-mesos/src/main/java/org/apache/flink/mesos/runtime/clusterframework/MesosApplicationMasterRunner.java
index 93eb3c6..544150b 100755
--- a/flink-mesos/src/main/java/org/apache/flink/mesos/runtime/clusterframework/MesosApplicationMasterRunner.java
+++ b/flink-mesos/src/main/java/org/apache/flink/mesos/runtime/clusterframework/MesosApplicationMasterRunner.java
@@ -52,6 +52,7 @@ import org.apache.flink.runtime.util.SignalHandler;
 import org.apache.flink.runtime.webmonitor.WebMonitor;
 import org.apache.flink.runtime.webmonitor.retriever.impl.AkkaJobManagerRetriever;
 import org.apache.flink.runtime.webmonitor.retriever.impl.AkkaQueryServiceRetriever;
+import org.apache.flink.util.ExecutorUtils;
 
 import akka.actor.ActorRef;
 import akka.actor.ActorSystem;
@@ -439,7 +440,7 @@ public class MesosApplicationMasterRunner {
 			}
 		}
 
-		org.apache.flink.runtime.concurrent.Executors.gracefulShutdown(
+		ExecutorUtils.gracefulShutdown(
 			AkkaUtils.getTimeout(config).toMillis(),
 			TimeUnit.MILLISECONDS,
 			futureExecutor,

http://git-wip-us.apache.org/repos/asf/flink/blob/74d052bb/flink-queryable-state/flink-queryable-state-client-java/src/main/java/org/apache/flink/queryablestate/network/AbstractServerBase.java
----------------------------------------------------------------------
diff --git a/flink-queryable-state/flink-queryable-state-client-java/src/main/java/org/apache/flink/queryablestate/network/AbstractServerBase.java
b/flink-queryable-state/flink-queryable-state-client-java/src/main/java/org/apache/flink/queryablestate/network/AbstractServerBase.java
index 82a05f2..d5afeb3 100644
--- a/flink-queryable-state/flink-queryable-state-client-java/src/main/java/org/apache/flink/queryablestate/network/AbstractServerBase.java
+++ b/flink-queryable-state/flink-queryable-state-client-java/src/main/java/org/apache/flink/queryablestate/network/AbstractServerBase.java
@@ -21,6 +21,7 @@ package org.apache.flink.queryablestate.network;
 import org.apache.flink.annotation.Internal;
 import org.apache.flink.annotation.VisibleForTesting;
 import org.apache.flink.queryablestate.network.messages.MessageBody;
+import org.apache.flink.util.ExecutorUtils;
 import org.apache.flink.util.FlinkRuntimeException;
 import org.apache.flink.util.Preconditions;
 
@@ -45,10 +46,12 @@ import java.net.InetSocketAddress;
 import java.util.HashSet;
 import java.util.Iterator;
 import java.util.Set;
+import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
 import java.util.concurrent.ThreadFactory;
 import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicReference;
 
 /**
  * The base class for every server in the queryable state module.
@@ -83,6 +86,9 @@ public abstract class AbstractServerBase<REQ extends MessageBody, RESP
extends M
 	/** The number of threads to be used for query serving. */
 	private final int numQueryThreads;
 
+	/** Atomic shut down future. */
+	private final AtomicReference<CompletableFuture<Void>> serverShutdownFuture
= new AtomicReference<>(null);
+
 	/** Netty's ServerBootstrap. */
 	private ServerBootstrap bootstrap;
 
@@ -179,8 +185,8 @@ public abstract class AbstractServerBase<REQ extends MessageBody, RESP
extends M
 	 * @throws Exception If something goes wrong during the bind operation.
 	 */
 	public void start() throws Throwable {
-		Preconditions.checkState(serverAddress == null,
-				serverName + " is already running @ " + serverAddress + '.');
+		Preconditions.checkState(serverAddress == null && serverShutdownFuture.get() ==
null,
+				serverName + " is already running @ " + serverAddress + ". ");
 
 		Iterator<Integer> portIterator = bindPortRange.iterator();
 		while (portIterator.hasNext() && !attemptToBind(portIterator.next())) {}
@@ -251,7 +257,22 @@ public abstract class AbstractServerBase<REQ extends MessageBody,
RESP extends M
 			throw future.cause();
 		} catch (BindException e) {
 			log.debug("Failed to start {} on port {}: {}.", serverName, port, e.getMessage());
-			shutdown();
+			try {
+				// we shutdown the server but we reset the future every time because in
+				// case of failure to bind, we will call attemptToBind() here, and not resetting
+				// the flag will interfere with future shutdown attempts.
+
+				shutdownServer()
+						.whenComplete((ignoredV, ignoredT) -> serverShutdownFuture.getAndSet(null))
+						.get();
+			} catch (Exception r) {
+
+				// Here we were seeing this problem:
+				// https://github.com/netty/netty/issues/4357 if we do a get().
+				// this is why we now simply wait a bit so that everything is shut down.
+
+				log.warn("Problem while shutting down {}: {}", serverName, r.getMessage());
+			}
 		}
 		// any other type of exception we let it bubble up.
 		return false;
@@ -259,26 +280,62 @@ public abstract class AbstractServerBase<REQ extends MessageBody,
RESP extends M
 
 	/**
 	 * Shuts down the server and all related thread pools.
+	 * @return A {@link CompletableFuture} that will be completed upon termination of the shutdown
process.
 	 */
-	public void shutdown() {
-		log.info("Shutting down {} @ {}", serverName, serverAddress);
-
-		if (handler != null) {
-			handler.shutdown();
-			handler = null;
-		}
-
-		if (queryExecutor != null) {
-			queryExecutor.shutdown();
-		}
+	public CompletableFuture<Void> shutdownServer() {
+		CompletableFuture<Void> shutdownFuture = new CompletableFuture<>();
+		if (serverShutdownFuture.compareAndSet(null, shutdownFuture)) {
+			log.info("Shutting down {} @ {}", serverName, serverAddress);
+
+			final CompletableFuture<Void> groupShutdownFuture = new CompletableFuture<>();
+			if (bootstrap != null) {
+				EventLoopGroup group = bootstrap.group();
+				if (group != null && !group.isShutdown()) {
+					group.shutdownGracefully(0L, 0L, TimeUnit.MILLISECONDS)
+							.addListener(finished -> {
+								if (finished.isSuccess()) {
+									groupShutdownFuture.complete(null);
+								} else {
+									groupShutdownFuture.completeExceptionally(finished.cause());
+								}
+							});
+				} else {
+					groupShutdownFuture.complete(null);
+				}
+			} else {
+				groupShutdownFuture.complete(null);
+			}
 
-		if (bootstrap != null) {
-			EventLoopGroup group = bootstrap.group();
-			if (group != null) {
-				group.shutdownGracefully(0L, 10L, TimeUnit.SECONDS);
+			final CompletableFuture<Void> handlerShutdownFuture = new CompletableFuture<>();
+			if (handler == null) {
+				handlerShutdownFuture.complete(null);
+			} else {
+				handler.shutdown().whenComplete((result, throwable) -> {
+					if (throwable != null) {
+						handlerShutdownFuture.completeExceptionally(throwable);
+					} else {
+						handlerShutdownFuture.complete(null);
+					}
+				});
 			}
+
+			final CompletableFuture<Void> queryExecShutdownFuture = CompletableFuture.runAsync(()
-> {
+				if (queryExecutor != null) {
+					ExecutorUtils.gracefulShutdown(10L, TimeUnit.MINUTES, queryExecutor);
+				}
+			});
+
+			CompletableFuture.allOf(
+					queryExecShutdownFuture, groupShutdownFuture, handlerShutdownFuture
+			).whenComplete((result, throwable) -> {
+				if (throwable != null) {
+					shutdownFuture.completeExceptionally(throwable);
+				} else {
+					shutdownFuture.complete(null);
+				}
+			});
 		}
-		serverAddress = null;
+		return serverShutdownFuture.get();
 	}
 
 	/**

http://git-wip-us.apache.org/repos/asf/flink/blob/74d052bb/flink-queryable-state/flink-queryable-state-client-java/src/main/java/org/apache/flink/queryablestate/network/AbstractServerHandler.java
----------------------------------------------------------------------
diff --git a/flink-queryable-state/flink-queryable-state-client-java/src/main/java/org/apache/flink/queryablestate/network/AbstractServerHandler.java
b/flink-queryable-state/flink-queryable-state-client-java/src/main/java/org/apache/flink/queryablestate/network/AbstractServerHandler.java
index 7e71a11..a514723 100644
--- a/flink-queryable-state/flink-queryable-state-client-java/src/main/java/org/apache/flink/queryablestate/network/AbstractServerHandler.java
+++ b/flink-queryable-state/flink-queryable-state-client-java/src/main/java/org/apache/flink/queryablestate/network/AbstractServerHandler.java
@@ -183,9 +183,16 @@ public abstract class AbstractServerHandler<REQ extends MessageBody,
RESP extend
 	public abstract CompletableFuture<RESP> handleRequest(final long requestId, final
REQ request);
 
 	/**
-	 * Shuts down any handler specific resources, e.g. thread pools etc.
+	 * Shuts down any handler-specific resources, e.g. thread pools etc and returns
+	 * a {@link CompletableFuture}.
+	 *
+	 * <p>If an exception is thrown during the shutdown process, then that exception
+	 * will be included in the returned future.
+	 *
+	 * @return A {@link CompletableFuture} that will be completed when the shutdown
+	 * process actually finishes.
 	 */
-	public abstract void shutdown();
+	public abstract CompletableFuture<Void> shutdown();
 
 	/**
 	 * Task to execute the actual query against the state instance.

http://git-wip-us.apache.org/repos/asf/flink/blob/74d052bb/flink-queryable-state/flink-queryable-state-runtime/src/main/java/org/apache/flink/queryablestate/client/proxy/KvStateClientProxyHandler.java
----------------------------------------------------------------------
diff --git a/flink-queryable-state/flink-queryable-state-runtime/src/main/java/org/apache/flink/queryablestate/client/proxy/KvStateClientProxyHandler.java
b/flink-queryable-state/flink-queryable-state-runtime/src/main/java/org/apache/flink/queryablestate/client/proxy/KvStateClientProxyHandler.java
index af33701..29ee0d7 100644
--- a/flink-queryable-state/flink-queryable-state-runtime/src/main/java/org/apache/flink/queryablestate/client/proxy/KvStateClientProxyHandler.java
+++ b/flink-queryable-state/flink-queryable-state-runtime/src/main/java/org/apache/flink/queryablestate/client/proxy/KvStateClientProxyHandler.java
@@ -33,7 +33,9 @@ import org.apache.flink.queryablestate.network.messages.MessageSerializer;
 import org.apache.flink.queryablestate.network.stats.DisabledKvStateRequestStats;
 import org.apache.flink.queryablestate.network.stats.KvStateRequestStats;
 import org.apache.flink.queryablestate.server.KvStateServerImpl;
+import org.apache.flink.runtime.concurrent.Executors;
 import org.apache.flink.runtime.concurrent.FutureUtils;
+import org.apache.flink.runtime.messages.FlinkJobNotFoundException;
 import org.apache.flink.runtime.query.KvStateClientProxy;
 import org.apache.flink.runtime.query.KvStateLocation;
 import org.apache.flink.runtime.query.KvStateMessage;
@@ -42,6 +44,7 @@ import org.apache.flink.util.Preconditions;
 
 import org.apache.flink.shaded.netty4.io.netty.channel.ChannelHandler;
 
+import akka.dispatch.OnComplete;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -141,6 +144,7 @@ public class KvStateClientProxyHandler extends AbstractServerHandler<KvStateRequ
 								// KvStateLocation. Therefore we retry this query and
 								// force look up the location.
 
+								LOG.debug("Retrying after failing to retrieve state due to: {}.", throwable.getCause().getMessage());
 								executeActionAsync(result, request, true);
 							} else {
 								result.completeExceptionally(throwable);
@@ -203,20 +207,34 @@ public class KvStateClientProxyHandler extends AbstractServerHandler<KvStateRequ
 
 		LOG.debug("Retrieving location for state={} of job={} from the job manager.", jobId, queryableStateName);
 
+		final CompletableFuture<KvStateLocation> location = new CompletableFuture<>();
+		lookupCache.put(cacheKey, location);
 		return proxy.getJobManagerFuture().thenComposeAsync(
 				jobManagerGateway -> {
 					final Object msg = new KvStateMessage.LookupKvStateLocation(jobId, queryableStateName);
-					final CompletableFuture<KvStateLocation> locationFuture = FutureUtils.toJava(
-							jobManagerGateway.ask(msg, FiniteDuration.apply(1000L, TimeUnit.MILLISECONDS))
-									.mapTo(ClassTag$.MODULE$.<KvStateLocation>apply(KvStateLocation.class)));
-
-					lookupCache.put(cacheKey, locationFuture);
-					return locationFuture;
+					jobManagerGateway.ask(msg, FiniteDuration.apply(1000L, TimeUnit.MILLISECONDS))
+							.mapTo(ClassTag$.MODULE$.<KvStateLocation>apply(KvStateLocation.class))
+							.onComplete(new OnComplete<KvStateLocation>() {
+
+								@Override
+								public void onComplete(Throwable failure, KvStateLocation loc) throws Throwable {
+									if (failure != null) {
+										if (failure instanceof FlinkJobNotFoundException) {
+											// if the jobId was wrong, remove the entry from the cache.
+											lookupCache.remove(cacheKey);
+										}
+										location.completeExceptionally(failure);
+									} else {
+										location.complete(loc);
+									}
+								}
+							}, Executors.directExecutionContext());
+					return location;
 				}, queryExecutor);
 	}
 
 	@Override
-	public void shutdown() {
-		kvStateClient.shutdown();
+	public CompletableFuture<Void> shutdown() {
+		return kvStateClient.shutdown();
 	}
 }

http://git-wip-us.apache.org/repos/asf/flink/blob/74d052bb/flink-queryable-state/flink-queryable-state-runtime/src/main/java/org/apache/flink/queryablestate/client/proxy/KvStateClientProxyImpl.java
----------------------------------------------------------------------
diff --git a/flink-queryable-state/flink-queryable-state-runtime/src/main/java/org/apache/flink/queryablestate/client/proxy/KvStateClientProxyImpl.java
b/flink-queryable-state/flink-queryable-state-runtime/src/main/java/org/apache/flink/queryablestate/client/proxy/KvStateClientProxyImpl.java
index 6fcaf40..aa5e7b6 100644
--- a/flink-queryable-state/flink-queryable-state-runtime/src/main/java/org/apache/flink/queryablestate/client/proxy/KvStateClientProxyImpl.java
+++ b/flink-queryable-state/flink-queryable-state-runtime/src/main/java/org/apache/flink/queryablestate/client/proxy/KvStateClientProxyImpl.java
@@ -35,6 +35,7 @@ import java.net.InetAddress;
 import java.net.InetSocketAddress;
 import java.util.Iterator;
 import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.TimeUnit;
 
 /**
  * The default implementation of the {@link KvStateClientProxy}.
@@ -96,7 +97,12 @@ public class KvStateClientProxyImpl extends AbstractServerBase<KvStateRequest,
K
 
 	@Override
 	public void shutdown() {
-		super.shutdown();
+		try {
+			shutdownServer().get(10L, TimeUnit.SECONDS);
+			log.info("{} was shutdown successfully.", getServerName());
+		} catch (Exception e) {
+			log.warn("{} shutdown failed: {}", getServerName(), e);
+		}
 	}
 
 	@Override

http://git-wip-us.apache.org/repos/asf/flink/blob/74d052bb/flink-queryable-state/flink-queryable-state-runtime/src/main/java/org/apache/flink/queryablestate/server/KvStateServerHandler.java
----------------------------------------------------------------------
diff --git a/flink-queryable-state/flink-queryable-state-runtime/src/main/java/org/apache/flink/queryablestate/server/KvStateServerHandler.java
b/flink-queryable-state/flink-queryable-state-runtime/src/main/java/org/apache/flink/queryablestate/server/KvStateServerHandler.java
index 476f153..18a2944 100644
--- a/flink-queryable-state/flink-queryable-state-runtime/src/main/java/org/apache/flink/queryablestate/server/KvStateServerHandler.java
+++ b/flink-queryable-state/flink-queryable-state-runtime/src/main/java/org/apache/flink/queryablestate/server/KvStateServerHandler.java
@@ -101,7 +101,7 @@ public class KvStateServerHandler extends AbstractServerHandler<KvStateInternalR
 	}
 
 	@Override
-	public void shutdown() {
-		// do nothing
+	public CompletableFuture<Void> shutdown() {
+		return CompletableFuture.completedFuture(null);
 	}
 }

http://git-wip-us.apache.org/repos/asf/flink/blob/74d052bb/flink-queryable-state/flink-queryable-state-runtime/src/main/java/org/apache/flink/queryablestate/server/KvStateServerImpl.java
----------------------------------------------------------------------
diff --git a/flink-queryable-state/flink-queryable-state-runtime/src/main/java/org/apache/flink/queryablestate/server/KvStateServerImpl.java
b/flink-queryable-state/flink-queryable-state-runtime/src/main/java/org/apache/flink/queryablestate/server/KvStateServerImpl.java
index 3a37a3a..0720268 100644
--- a/flink-queryable-state/flink-queryable-state-runtime/src/main/java/org/apache/flink/queryablestate/server/KvStateServerImpl.java
+++ b/flink-queryable-state/flink-queryable-state-runtime/src/main/java/org/apache/flink/queryablestate/server/KvStateServerImpl.java
@@ -32,6 +32,7 @@ import org.apache.flink.util.Preconditions;
 import java.net.InetAddress;
 import java.net.InetSocketAddress;
 import java.util.Iterator;
+import java.util.concurrent.TimeUnit;
 
 /**
  * The default implementation of the {@link KvStateServer}.
@@ -101,6 +102,11 @@ public class KvStateServerImpl extends AbstractServerBase<KvStateInternalRequest
 
 	@Override
 	public void shutdown() {
-		super.shutdown();
+		try {
+			shutdownServer().get(10L, TimeUnit.SECONDS);
+			log.info("{} was shutdown successfully.", getServerName());
+		} catch (Exception e) {
+			log.warn("{} shutdown failed: {}", getServerName(), e);
+		}
 	}
 }

http://git-wip-us.apache.org/repos/asf/flink/blob/74d052bb/flink-queryable-state/flink-queryable-state-runtime/src/test/java/org/apache/flink/queryablestate/itcases/HAAbstractQueryableStateTestBase.java
----------------------------------------------------------------------
diff --git a/flink-queryable-state/flink-queryable-state-runtime/src/test/java/org/apache/flink/queryablestate/itcases/HAAbstractQueryableStateTestBase.java
b/flink-queryable-state/flink-queryable-state-runtime/src/test/java/org/apache/flink/queryablestate/itcases/HAAbstractQueryableStateTestBase.java
index 79809b3..b9ce7c2 100644
--- a/flink-queryable-state/flink-queryable-state-runtime/src/test/java/org/apache/flink/queryablestate/itcases/HAAbstractQueryableStateTestBase.java
+++ b/flink-queryable-state/flink-queryable-state-runtime/src/test/java/org/apache/flink/queryablestate/itcases/HAAbstractQueryableStateTestBase.java
@@ -88,11 +88,10 @@ public abstract class HAAbstractQueryableStateTestBase extends AbstractQueryable
 		try {
 			zkServer.stop();
 			zkServer.close();
-		} catch (Exception e) {
+			client.shutdownAndWait();
+		} catch (Throwable e) {
 			e.printStackTrace();
 			fail(e.getMessage());
 		}
-
-		client.shutdown();
 	}
 }

http://git-wip-us.apache.org/repos/asf/flink/blob/74d052bb/flink-queryable-state/flink-queryable-state-runtime/src/test/java/org/apache/flink/queryablestate/itcases/NonHAAbstractQueryableStateTestBase.java
----------------------------------------------------------------------
diff --git a/flink-queryable-state/flink-queryable-state-runtime/src/test/java/org/apache/flink/queryablestate/itcases/NonHAAbstractQueryableStateTestBase.java
b/flink-queryable-state/flink-queryable-state-runtime/src/test/java/org/apache/flink/queryablestate/itcases/NonHAAbstractQueryableStateTestBase.java
index 6945cca..a5e24b2 100644
--- a/flink-queryable-state/flink-queryable-state-runtime/src/test/java/org/apache/flink/queryablestate/itcases/NonHAAbstractQueryableStateTestBase.java
+++ b/flink-queryable-state/flink-queryable-state-runtime/src/test/java/org/apache/flink/queryablestate/itcases/NonHAAbstractQueryableStateTestBase.java
@@ -69,10 +69,10 @@ public abstract class NonHAAbstractQueryableStateTestBase extends AbstractQuerya
 	public static void tearDown() {
 		try {
 			cluster.stop();
-		} catch (Exception e) {
+			client.shutdownAndWait();
+		} catch (Throwable e) {
 			e.printStackTrace();
 			fail(e.getMessage());
 		}
-		client.shutdown();
 	}
 }

http://git-wip-us.apache.org/repos/asf/flink/blob/74d052bb/flink-queryable-state/flink-queryable-state-runtime/src/test/java/org/apache/flink/queryablestate/network/AbstractServerTest.java
----------------------------------------------------------------------
diff --git a/flink-queryable-state/flink-queryable-state-runtime/src/test/java/org/apache/flink/queryablestate/network/AbstractServerTest.java
b/flink-queryable-state/flink-queryable-state-runtime/src/test/java/org/apache/flink/queryablestate/network/AbstractServerTest.java
index 3d2ed40..103c638 100644
--- a/flink-queryable-state/flink-queryable-state-runtime/src/test/java/org/apache/flink/queryablestate/network/AbstractServerTest.java
+++ b/flink-queryable-state/flink-queryable-state-runtime/src/test/java/org/apache/flink/queryablestate/network/AbstractServerTest.java
@@ -22,7 +22,9 @@ import org.apache.flink.configuration.ConfigConstants;
 import org.apache.flink.queryablestate.network.messages.MessageBody;
 import org.apache.flink.queryablestate.network.messages.MessageDeserializer;
 import org.apache.flink.queryablestate.network.messages.MessageSerializer;
+import org.apache.flink.queryablestate.network.stats.AtomicKvStateRequestStats;
 import org.apache.flink.queryablestate.network.stats.DisabledKvStateRequestStats;
+import org.apache.flink.queryablestate.network.stats.KvStateRequestStats;
 import org.apache.flink.util.FlinkRuntimeException;
 import org.apache.flink.util.Preconditions;
 
@@ -37,6 +39,7 @@ import java.net.InetAddress;
 import java.net.UnknownHostException;
 import java.nio.ByteBuffer;
 import java.util.ArrayList;
+import java.util.Collections;
 import java.util.Iterator;
 import java.util.List;
 import java.util.concurrent.CompletableFuture;
@@ -60,22 +63,15 @@ public class AbstractServerTest {
 		expectedEx.expect(FlinkRuntimeException.class);
 		expectedEx.expectMessage("Unable to start Test Server 2. All ports in provided range are
occupied.");
 
-		TestServer server1 = null;
-		TestServer server2 = null;
-		try {
+		List<Integer> portList = new ArrayList<>();
+		portList.add(7777);
 
-			server1 = startServer("Test Server 1", 7777);
-			Assert.assertEquals(7777L, server1.getServerAddress().getPort());
+		try (TestServer server1 = new TestServer("Test Server 1", new DisabledKvStateRequestStats(),
portList.iterator())) {
+			server1.start();
 
-			server2 = startServer("Test Server 2", 7777);
-		} finally {
-
-			if (server1 != null) {
-				server1.shutdown();
-			}
-
-			if (server2 != null) {
-				server2.shutdown();
+			try (TestServer server2 = new TestServer("Test Server 2", new DisabledKvStateRequestStats(),
+					Collections.singletonList(server1.getServerAddress().getPort()).iterator())) {
+				server2.start();
 			}
 		}
 	}
@@ -86,69 +82,81 @@ public class AbstractServerTest {
 	 */
 	@Test
 	public void testPortRangeSuccess() throws Throwable {
-		TestServer server1 = null;
-		TestServer server2 = null;
-		Client<TestMessage, TestMessage> client = null;
 
-		try {
-			server1 = startServer("Test Server 1", 7777, 7778, 7779);
-			Assert.assertEquals(7777L, server1.getServerAddress().getPort());
-
-			server2 = startServer("Test Server 2", 7777, 7778, 7779);
-			Assert.assertEquals(7778L, server2.getServerAddress().getPort());
-
-			client = new Client<>(
-					"Test Client",
-					1,
-					new MessageSerializer<>(new TestMessage.TestMessageDeserializer(), new TestMessage.TestMessageDeserializer()),
-					new DisabledKvStateRequestStats());
+		// this is shared between the two servers.
+		AtomicKvStateRequestStats serverStats = new AtomicKvStateRequestStats();
+		AtomicKvStateRequestStats clientStats = new AtomicKvStateRequestStats();
+
+		List<Integer> portList = new ArrayList<>();
+		portList.add(7777);
+		portList.add(7778);
+		portList.add(7779);
+
+		try (
+				TestServer server1 = new TestServer("Test Server 1", serverStats, portList.iterator());
+				TestServer server2 = new TestServer("Test Server 2", serverStats, portList.iterator());
+				TestClient client = new TestClient(
+						"Test Client",
+						1,
+						new MessageSerializer<>(new TestMessage.TestMessageDeserializer(), new TestMessage.TestMessageDeserializer()),
+						clientStats
+				)
+		) {
+			server1.start();
+			Assert.assertTrue(server1.getServerAddress().getPort() >= 7777 && server1.getServerAddress().getPort()
<= 7779);
+
+			server2.start();
+			Assert.assertTrue(server2.getServerAddress().getPort() >= 7777 && server2.getServerAddress().getPort()
<= 7779);
 
 			TestMessage response1 = client.sendRequest(server1.getServerAddress(), new TestMessage("ping")).join();
 			Assert.assertEquals(server1.getServerName() + "-ping", response1.getMessage());
 
 			TestMessage response2 = client.sendRequest(server2.getServerAddress(), new TestMessage("pong")).join();
 			Assert.assertEquals(server2.getServerName() + "-pong", response2.getMessage());
-		} finally {
 
-			if (server1 != null) {
-				server1.shutdown();
-			}
-
-			if (server2 != null) {
-				server2.shutdown();
-			}
+			// the client connects to both servers and the stats object is shared.
+			Assert.assertEquals(2L, serverStats.getNumConnections());
 
-			if (client != null) {
-				client.shutdown();
-			}
+			Assert.assertEquals(2L, clientStats.getNumConnections());
+			Assert.assertEquals(0L, clientStats.getNumFailed());
+			Assert.assertEquals(2L, clientStats.getNumSuccessful());
+			Assert.assertEquals(2L, clientStats.getNumRequests());
 		}
+
+		Assert.assertEquals(0L, clientStats.getNumConnections());
+		Assert.assertEquals(0L, clientStats.getNumFailed());
+		Assert.assertEquals(2L, clientStats.getNumSuccessful());
+		Assert.assertEquals(2L, clientStats.getNumRequests());
 	}
 
-	/**
-	 * Initializes a {@link TestServer} with the given port range.
-	 * @param serverName the name of the server.
-	 * @param ports a range of ports.
-	 * @return A test server with the given name.
-	 */
-	private TestServer startServer(String serverName, int... ports) throws Throwable {
-		List<Integer> portList = new ArrayList<>(ports.length);
-		for (int p : ports) {
-			portList.add(p);
+	private static class TestClient extends Client<TestMessage, TestMessage> implements
AutoCloseable {
+
+		TestClient(
+				String clientName,
+				int numEventLoopThreads,
+				MessageSerializer<TestMessage, TestMessage> serializer,
+				KvStateRequestStats stats) {
+			super(clientName, numEventLoopThreads, serializer, stats);
 		}
 
-		final TestServer server = new TestServer(serverName, portList.iterator());
-		server.start();
-		return server;
+		@Override
+		public void close() throws Exception {
+			shutdown().join();
+			Assert.assertTrue(isEventGroupShutdown());
+		}
 	}
 
 	/**
 	 * A server that receives a {@link TestMessage test message} and returns another test
 	 * message containing the same string as the request with the name of the server prepended.
 	 */
-	private class TestServer extends AbstractServerBase<TestMessage, TestMessage> {
+	private static class TestServer extends AbstractServerBase<TestMessage, TestMessage>
implements AutoCloseable {
 
-		protected TestServer(String name, Iterator<Integer> bindPort) throws UnknownHostException
{
+		private final KvStateRequestStats requestStats;
+
+		TestServer(String name, KvStateRequestStats stats, Iterator<Integer> bindPort) throws
UnknownHostException {
 			super(name, InetAddress.getLocalHost(), bindPort, 1, 1);
+			this.requestStats = stats;
 		}
 
 		@Override
@@ -156,7 +164,7 @@ public class AbstractServerTest {
 			return new AbstractServerHandler<TestMessage, TestMessage>(
 					this,
 					new MessageSerializer<>(new TestMessage.TestMessageDeserializer(), new TestMessage.TestMessageDeserializer()),
-					new DisabledKvStateRequestStats()) {
+					requestStats) {
 
 				@Override
 				public CompletableFuture<TestMessage> handleRequest(long requestId, TestMessage
request) {
@@ -165,11 +173,22 @@ public class AbstractServerTest {
 				}
 
 				@Override
-				public void shutdown() {
-					// do nothing
+				public CompletableFuture<Void> shutdown() {
+					return CompletableFuture.completedFuture(null);
 				}
 			};
 		}
+
+		@Override
+		public void close() throws Exception {
+			shutdownServer().get();
+			if (requestStats instanceof AtomicKvStateRequestStats) {
+				AtomicKvStateRequestStats stats = (AtomicKvStateRequestStats) requestStats;
+				Assert.assertEquals(0L, stats.getNumConnections());
+			}
+			Assert.assertTrue(getQueryExecutor().isTerminated());
+			Assert.assertTrue(isEventGroupShutdown());
+		}
 	}
 
 	/**

http://git-wip-us.apache.org/repos/asf/flink/blob/74d052bb/flink-runtime/src/main/java/org/apache/flink/runtime/concurrent/Executors.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/concurrent/Executors.java
b/flink-runtime/src/main/java/org/apache/flink/runtime/concurrent/Executors.java
index 04cdce7..703ac4e 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/concurrent/Executors.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/concurrent/Executors.java
@@ -22,14 +22,13 @@ import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 import javax.annotation.Nonnull;
+
 import java.util.concurrent.Executor;
-import java.util.concurrent.ExecutorService;
-import java.util.concurrent.TimeUnit;
 
 import scala.concurrent.ExecutionContext;
 
 /**
- * Collection of {@link Executor} implementations
+ * Collection of {@link Executor} implementations.
  */
 public class Executors {
 
@@ -94,48 +93,4 @@ public class Executors {
 			return this;
 		}
 	}
-
-	/**
-	 * Gracefully shutdown the given {@link ExecutorService}. The call waits the given timeout
that
-	 * all ExecutorServices terminate. If the ExecutorServices do not terminate in this time,
-	 * they will be shut down hard.
-	 *
-	 * @param timeout to wait for the termination of all ExecutorServices
-	 * @param unit of the timeout
-	 * @param executorServices to shut down
-	 */
-	public static void gracefulShutdown(long timeout, TimeUnit unit, ExecutorService... executorServices)
{
-		for (ExecutorService executorService: executorServices) {
-			executorService.shutdown();
-		}
-
-		boolean wasInterrupted = false;
-		final long endTime = unit.toMillis(timeout) + System.currentTimeMillis();
-		long timeLeft = unit.toMillis(timeout);
-		boolean hasTimeLeft = timeLeft > 0L;
-
-		for (ExecutorService executorService: executorServices) {
-			if (wasInterrupted || !hasTimeLeft) {
-				executorService.shutdownNow();
-			} else {
-				try {
-					if (!executorService.awaitTermination(timeLeft, TimeUnit.MILLISECONDS)) {
-						LOG.warn("ExecutorService did not terminate in time. Shutting it down now.");
-						executorService.shutdownNow();
-					}
-				} catch (InterruptedException e) {
-					LOG.warn("Interrupted while shutting down executor services. Shutting all " +
-						"remaining ExecutorServices down now.", e);
-					executorService.shutdownNow();
-
-					wasInterrupted = true;
-
-					Thread.currentThread().interrupt();
-				}
-
-				timeLeft = endTime - System.currentTimeMillis();
-				hasTimeLeft = timeLeft > 0L;
-			}
-		}
-	}
 }

http://git-wip-us.apache.org/repos/asf/flink/blob/74d052bb/flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskManagerRunner.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskManagerRunner.java
b/flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskManagerRunner.java
index a24daf0..7258e52 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskManagerRunner.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskManagerRunner.java
@@ -25,7 +25,6 @@ import org.apache.flink.configuration.Configuration;
 import org.apache.flink.configuration.GlobalConfiguration;
 import org.apache.flink.runtime.akka.AkkaUtils;
 import org.apache.flink.runtime.clusterframework.types.ResourceID;
-import org.apache.flink.runtime.concurrent.Executors;
 import org.apache.flink.runtime.heartbeat.HeartbeatServices;
 import org.apache.flink.runtime.highavailability.HighAvailabilityServices;
 import org.apache.flink.runtime.highavailability.HighAvailabilityServicesUtils;
@@ -47,6 +46,7 @@ import org.apache.flink.runtime.util.JvmShutdownSafeguard;
 import org.apache.flink.runtime.util.LeaderRetrievalUtils;
 import org.apache.flink.runtime.util.SignalHandler;
 import org.apache.flink.util.ExceptionUtils;
+import org.apache.flink.util.ExecutorUtils;
 import org.apache.flink.util.Preconditions;
 
 import akka.actor.ActorSystem;
@@ -88,7 +88,7 @@ public class TaskManagerRunner implements FatalErrorHandler {
 
 	private final MetricRegistryImpl metricRegistry;
 
-	/** Executor used to run future callbacks */
+	/** Executor used to run future callbacks. */
 	private final ExecutorService executor;
 
 	private final TaskExecutor taskManager;
@@ -165,7 +165,7 @@ public class TaskManagerRunner implements FatalErrorHandler {
 				exception = ExceptionUtils.firstOrSuppressed(e, exception);
 			}
 
-			Executors.gracefulShutdown(timeout.toMilliseconds(), TimeUnit.MILLISECONDS, executor);
+			ExecutorUtils.gracefulShutdown(timeout.toMilliseconds(), TimeUnit.MILLISECONDS, executor);
 
 			if (exception != null) {
 				throw exception;

http://git-wip-us.apache.org/repos/asf/flink/blob/74d052bb/flink-runtime/src/main/scala/org/apache/flink/runtime/jobmanager/JobManager.scala
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/scala/org/apache/flink/runtime/jobmanager/JobManager.scala
b/flink-runtime/src/main/scala/org/apache/flink/runtime/jobmanager/JobManager.scala
index 7799a78..5f82159 100644
--- a/flink-runtime/src/main/scala/org/apache/flink/runtime/jobmanager/JobManager.scala
+++ b/flink-runtime/src/main/scala/org/apache/flink/runtime/jobmanager/JobManager.scala
@@ -45,7 +45,7 @@ import org.apache.flink.runtime.clusterframework.{BootstrapTools, FlinkResourceM
 import org.apache.flink.runtime.clusterframework.messages._
 import org.apache.flink.runtime.clusterframework.standalone.StandaloneResourceManager
 import org.apache.flink.runtime.clusterframework.types.ResourceID
-import org.apache.flink.runtime.concurrent.{FutureUtils, ScheduledExecutorServiceAdapter,
Executors => FlinkExecutors}
+import org.apache.flink.runtime.concurrent.{FutureUtils, ScheduledExecutorServiceAdapter}
 import org.apache.flink.runtime.execution.SuppressRestartsException
 import org.apache.flink.runtime.execution.librarycache.FlinkUserCodeClassLoaders.ResolveOrder
 import org.apache.flink.runtime.execution.librarycache.{BlobLibraryCacheManager, LibraryCacheManager}
@@ -85,7 +85,7 @@ import org.apache.flink.runtime.util._
 import org.apache.flink.runtime.webmonitor.retriever.impl.{AkkaJobManagerRetriever, AkkaQueryServiceRetriever}
 import org.apache.flink.runtime.webmonitor.{WebMonitor, WebMonitorUtils}
 import org.apache.flink.runtime.{FlinkActor, LeaderSessionMessageFilter, LogMessages}
-import org.apache.flink.util.{FlinkException, InstantiationUtil, NetUtils, SerializedThrowable}
+import org.apache.flink.util._
 
 import scala.collection.JavaConverters._
 import scala.collection.mutable
@@ -2060,7 +2060,7 @@ object JobManager {
         LOG.warn("Could not properly shut down the metric registry.", t)
     }
 
-    FlinkExecutors.gracefulShutdown(
+    ExecutorUtils.gracefulShutdown(
       timeout.toMillis,
       TimeUnit.MILLISECONDS,
       futureExecutor,

http://git-wip-us.apache.org/repos/asf/flink/blob/74d052bb/flink-runtime/src/main/scala/org/apache/flink/runtime/minicluster/FlinkMiniCluster.scala
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/scala/org/apache/flink/runtime/minicluster/FlinkMiniCluster.scala
b/flink-runtime/src/main/scala/org/apache/flink/runtime/minicluster/FlinkMiniCluster.scala
index 227b854..5554061 100644
--- a/flink-runtime/src/main/scala/org/apache/flink/runtime/minicluster/FlinkMiniCluster.scala
+++ b/flink-runtime/src/main/scala/org/apache/flink/runtime/minicluster/FlinkMiniCluster.scala
@@ -32,7 +32,7 @@ import org.apache.flink.configuration._
 import org.apache.flink.core.fs.Path
 import org.apache.flink.runtime.akka.{AkkaJobManagerGateway, AkkaUtils}
 import org.apache.flink.runtime.client.{JobClient, JobExecutionException}
-import org.apache.flink.runtime.concurrent.{FutureUtils, ScheduledExecutorServiceAdapter,
Executors => FlinkExecutors}
+import org.apache.flink.runtime.concurrent.{FutureUtils, ScheduledExecutorServiceAdapter}
 import org.apache.flink.runtime.execution.librarycache.FlinkUserCodeClassLoaders
 import org.apache.flink.runtime.highavailability.{HighAvailabilityServices, HighAvailabilityServicesUtils}
 import org.apache.flink.runtime.instance.{ActorGateway, AkkaActorGateway}
@@ -44,7 +44,7 @@ import org.apache.flink.runtime.metrics.{MetricRegistryConfiguration, MetricRegi
 import org.apache.flink.runtime.util.{ExecutorThreadFactory, Hardware}
 import org.apache.flink.runtime.webmonitor.retriever.impl.{AkkaJobManagerRetriever, AkkaQueryServiceRetriever}
 import org.apache.flink.runtime.webmonitor.{WebMonitor, WebMonitorUtils}
-import org.apache.flink.util.NetUtils
+import org.apache.flink.util.{ExecutorUtils, NetUtils}
 import org.slf4j.LoggerFactory
 
 import scala.concurrent.duration.{Duration, FiniteDuration}
@@ -440,7 +440,7 @@ abstract class FlinkMiniCluster(
 
     isRunning = false
 
-    FlinkExecutors.gracefulShutdown(
+    ExecutorUtils.gracefulShutdown(
       timeout.toMillis,
       TimeUnit.MILLISECONDS,
       futureExecutor,

http://git-wip-us.apache.org/repos/asf/flink/blob/74d052bb/flink-runtime/src/test/java/org/apache/flink/runtime/resourcemanager/slotmanager/SlotProtocolTest.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/resourcemanager/slotmanager/SlotProtocolTest.java
b/flink-runtime/src/test/java/org/apache/flink/runtime/resourcemanager/slotmanager/SlotProtocolTest.java
index 97942ea..79e38df 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/resourcemanager/slotmanager/SlotProtocolTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/resourcemanager/slotmanager/SlotProtocolTest.java
@@ -33,6 +33,7 @@ import org.apache.flink.runtime.taskexecutor.SlotReport;
 import org.apache.flink.runtime.taskexecutor.SlotStatus;
 import org.apache.flink.runtime.taskexecutor.TaskExecutorGateway;
 import org.apache.flink.runtime.testingUtils.TestingUtils;
+import org.apache.flink.util.ExecutorUtils;
 import org.apache.flink.util.TestLogger;
 
 import org.junit.AfterClass;
@@ -64,7 +65,7 @@ public class SlotProtocolTest extends TestLogger {
 
 	@AfterClass
 	public static void afterClass() {
-		Executors.gracefulShutdown(timeout, TimeUnit.MILLISECONDS, scheduledExecutorService);
+		ExecutorUtils.gracefulShutdown(timeout, TimeUnit.MILLISECONDS, scheduledExecutorService);
 	}
 
 	/**

http://git-wip-us.apache.org/repos/asf/flink/blob/74d052bb/flink-runtime/src/test/java/org/apache/flink/runtime/rest/handler/legacy/ExecutionGraphCacheTest.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/rest/handler/legacy/ExecutionGraphCacheTest.java
b/flink-runtime/src/test/java/org/apache/flink/runtime/rest/handler/legacy/ExecutionGraphCacheTest.java
index 1a8ea84..ecadaa5 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/rest/handler/legacy/ExecutionGraphCacheTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/rest/handler/legacy/ExecutionGraphCacheTest.java
@@ -23,7 +23,6 @@ import org.apache.flink.api.common.ExecutionConfig;
 import org.apache.flink.api.common.JobID;
 import org.apache.flink.api.common.time.Time;
 import org.apache.flink.runtime.accumulators.StringifiedAccumulatorResult;
-import org.apache.flink.runtime.concurrent.Executors;
 import org.apache.flink.runtime.concurrent.FutureUtils;
 import org.apache.flink.runtime.executiongraph.AccessExecutionGraph;
 import org.apache.flink.runtime.executiongraph.ArchivedExecutionGraph;
@@ -33,6 +32,7 @@ import org.apache.flink.runtime.jobgraph.JobStatus;
 import org.apache.flink.runtime.jobmanager.JobManager;
 import org.apache.flink.runtime.jobmaster.JobManagerGateway;
 import org.apache.flink.runtime.messages.FlinkJobNotFoundException;
+import org.apache.flink.util.ExecutorUtils;
 import org.apache.flink.util.FlinkException;
 import org.apache.flink.util.TestLogger;
 
@@ -233,7 +233,7 @@ public class ExecutionGraphCacheTest extends TestLogger {
 
 			verify(jobManagerGateway, times(1)).requestJob(eq(jobId), any(Time.class));
 		} finally {
-			Executors.gracefulShutdown(5000L, TimeUnit.MILLISECONDS, executor);
+			ExecutorUtils.gracefulShutdown(5000L, TimeUnit.MILLISECONDS, executor);
 		}
 	}
 

http://git-wip-us.apache.org/repos/asf/flink/blob/74d052bb/flink-yarn/src/main/java/org/apache/flink/yarn/YarnApplicationMasterRunner.java
----------------------------------------------------------------------
diff --git a/flink-yarn/src/main/java/org/apache/flink/yarn/YarnApplicationMasterRunner.java
b/flink-yarn/src/main/java/org/apache/flink/yarn/YarnApplicationMasterRunner.java
index 3bdc2ac..279981a 100644
--- a/flink-yarn/src/main/java/org/apache/flink/yarn/YarnApplicationMasterRunner.java
+++ b/flink-yarn/src/main/java/org/apache/flink/yarn/YarnApplicationMasterRunner.java
@@ -51,6 +51,7 @@ import org.apache.flink.runtime.util.SignalHandler;
 import org.apache.flink.runtime.webmonitor.WebMonitor;
 import org.apache.flink.runtime.webmonitor.retriever.impl.AkkaJobManagerRetriever;
 import org.apache.flink.runtime.webmonitor.retriever.impl.AkkaQueryServiceRetriever;
+import org.apache.flink.util.ExecutorUtils;
 import org.apache.flink.yarn.cli.FlinkYarnSessionCli;
 import org.apache.flink.yarn.configuration.YarnConfigOptions;
 
@@ -472,7 +473,7 @@ public class YarnApplicationMasterRunner {
 			}
 		}
 
-		org.apache.flink.runtime.concurrent.Executors.gracefulShutdown(
+		ExecutorUtils.gracefulShutdown(
 			AkkaUtils.getTimeout(config).toMillis(),
 			TimeUnit.MILLISECONDS,
 			futureExecutor,


Mime
View raw message