Github user tillrohrmann commented on a diff in the pull request:
https://github.com/apache/flink/pull/4993#discussion_r150231729
--- Diff:
flink-queryable-state/flink-queryable-state-client-java/src/main/java/org/apache/flink/queryablestate/network/AbstractServerBase.java
---
@@ -260,25 +262,47 @@ private boolean attemptToBind(final int port) throws
Throwable {
/**
* Shuts down the server and all related thread pools.
*/
- public void shutdown() {
- LOG.info("Shutting down server {} @ {}", serverName,
serverAddress);
-
- if (handler != null) {
- handler.shutdown();
- handler = null;
- }
-
- if (queryExecutor != null) {
- queryExecutor.shutdown();
- }
+ public CompletableFuture<?> shutdownServer(Time timeout) throws
InterruptedException {
+ log.info("Shutting down {} @ {}", serverName, serverAddress);
+
+ final CompletableFuture<Boolean> queryExecShutdownFuture =
CompletableFuture.supplyAsync(() -> {
+ try {
+ if (queryExecutor != null &&
!queryExecutor.isShutdown()) {
+ queryExecutor.shutdown();
+
queryExecutor.awaitTermination(timeout.toMilliseconds(), TimeUnit.MILLISECONDS);
--- End diff --
What if `awaitTermination` returns `false`? Then we should call
`queryExecutor.shutdownNow`
---
