You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by kl0u <gi...@git.apache.org> on 2017/11/23 14:10:00 UTC
[GitHub] flink pull request #5062: [FLINK-7880][QS] Wait for proper resource cleanup ...
GitHub user kl0u opened a pull request:
https://github.com/apache/flink/pull/5062
[FLINK-7880][QS] Wait for proper resource cleanup after each ITCase.
R @aljoscha
You can merge this pull request into a Git repository by running:
$ git pull https://github.com/kl0u/flink qs-shutdown-fin-fin
Alternatively you can review and apply these changes as the patch at:
https://github.com/apache/flink/pull/5062.patch
To close this pull request, make a commit to your master/trunk branch
with (at least) the following in the commit message:
This closes #5062
----
commit fd8ea759321b83022497e25e2dfa56ac6d976e15
Author: kkloudas <kk...@gmail.com>
Date: 2017-11-09T18:21:43Z
[FLINK-7975][QS] Wait for QS client to shutdown.
commit a474071d5810562326930a147f0ecdcfe506f13d
Author: kkloudas <kk...@gmail.com>
Date: 2017-11-09T18:30:29Z
[FLINK-7974][QS] Wait for QS abstract server to shutdown.
commit 2ffe223a8c8d89efbee0e08f8b90ea7b58fbf9df
Author: kkloudas <kk...@gmail.com>
Date: 2017-11-20T13:28:24Z
[FLINK-7880][QS] Wait for proper resource cleanup after each ITCase.
commit f7f8d2a6ebbaf464200c44a902e5b829c88f9499
Author: kkloudas <kk...@gmail.com>
Date: 2017-11-23T13:20:32Z
[hotfix][QS] ITCase refactoring.
----
---
[GitHub] flink pull request #5062: [FLINK-7880][QS] Wait for proper resource cleanup ...
Posted by kl0u <gi...@git.apache.org>.
Github user kl0u commented on a diff in the pull request:
https://github.com/apache/flink/pull/5062#discussion_r154885293
--- Diff: flink-queryable-state/flink-queryable-state-client-java/src/main/java/org/apache/flink/queryablestate/network/AbstractServerBase.java ---
@@ -251,34 +263,81 @@ private boolean attemptToBind(final int port) throws Throwable {
throw future.cause();
} catch (BindException e) {
log.debug("Failed to start {} on port {}: {}.", serverName, port, e.getMessage());
- shutdown();
+ try {
+ shutdownServer(Time.seconds(10L)).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 and then we check
+
+ log.warn("Problem while shutting down {}: {}", serverName, r.getMessage());
+ }
}
// any other type of exception we let it bubble up.
return false;
}
/**
* Shuts down the server and all related thread pools.
+ * @param timeout The time to wait for the shutdown process to complete.
+ * @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(Time timeout) throws InterruptedException {
+ 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(timeout.toMilliseconds(), TimeUnit.MILLISECONDS, queryExecutor);
+ }
+ });
+
+ CompletableFuture.allOf(
+ queryExecShutdownFuture, groupShutdownFuture, handlerShutdownFuture
+ ).whenComplete((result, throwable) -> {
+ if (throwable != null) {
+ shutdownFuture.completeExceptionally(throwable);
+ } else {
+ shutdownFuture.complete(null);
+ }
+ });
}
- serverAddress = null;
+ return serverShutdownFuture.get();
--- End diff --
I do think so, because this would mean that we can restart the server that we intentionally shut down in that past, right? (This is only done in the start() when we fail to bind to a specific port).
---
[GitHub] flink pull request #5062: [FLINK-7880][QS] Wait for proper resource cleanup ...
Posted by asfgit <gi...@git.apache.org>.
Github user asfgit closed the pull request at:
https://github.com/apache/flink/pull/5062
---
[GitHub] flink pull request #5062: [FLINK-7880][QS] Wait for proper resource cleanup ...
Posted by zentol <gi...@git.apache.org>.
Github user zentol commented on a diff in the pull request:
https://github.com/apache/flink/pull/5062#discussion_r154950558
--- Diff: flink-queryable-state/flink-queryable-state-client-java/src/main/java/org/apache/flink/queryablestate/network/AbstractServerBase.java ---
@@ -251,34 +263,81 @@ private boolean attemptToBind(final int port) throws Throwable {
throw future.cause();
} catch (BindException e) {
log.debug("Failed to start {} on port {}: {}.", serverName, port, e.getMessage());
- shutdown();
+ try {
+ shutdownServer(Time.seconds(10L)).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 and then we check
+
+ log.warn("Problem while shutting down {}: {}", serverName, r.getMessage());
+ }
}
// any other type of exception we let it bubble up.
return false;
}
/**
* Shuts down the server and all related thread pools.
+ * @param timeout The time to wait for the shutdown process to complete.
+ * @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(Time timeout) throws InterruptedException {
+ 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(timeout.toMilliseconds(), TimeUnit.MILLISECONDS, queryExecutor);
+ }
+ });
+
+ CompletableFuture.allOf(
+ queryExecShutdownFuture, groupShutdownFuture, handlerShutdownFuture
+ ).whenComplete((result, throwable) -> {
+ if (throwable != null) {
+ shutdownFuture.completeExceptionally(throwable);
+ } else {
+ shutdownFuture.complete(null);
+ }
+ });
}
- serverAddress = null;
+ return serverShutdownFuture.get();
--- End diff --
So we do not intend to support restarting the server? In this case I would suggest to check the shutdownfuture in start() and provide a better error message, as it currently would return "serverX is already running".
---
[GitHub] flink issue #5062: [FLINK-7880][QS] Wait for proper resource cleanup after e...
Posted by kl0u <gi...@git.apache.org>.
Github user kl0u commented on the issue:
https://github.com/apache/flink/pull/5062
Thanks @zentol ! I will let another travis run, and then merge.
---
[GitHub] flink pull request #5062: [FLINK-7880][QS] Wait for proper resource cleanup ...
Posted by zentol <gi...@git.apache.org>.
Github user zentol commented on a diff in the pull request:
https://github.com/apache/flink/pull/5062#discussion_r153793887
--- Diff: flink-queryable-state/flink-queryable-state-runtime/src/test/java/org/apache/flink/queryablestate/itcases/HAAbstractQueryableStateTestBase.java ---
@@ -80,19 +80,18 @@ public static void setup(int proxyPortRangeStart, int serverPortRangeStart) {
@AfterClass
public static void tearDown() {
- if (cluster != null) {
+ try {
+ client.shutdownAndWait();
+
cluster.stop();
cluster.awaitTermination();
- }
- try {
zkServer.stop();
zkServer.close();
- } catch (Exception e) {
+
+ } catch (Throwable e) {
--- End diff --
try-catch block is unnecessary
---
[GitHub] flink pull request #5062: [FLINK-7880][QS] Wait for proper resource cleanup ...
Posted by zentol <gi...@git.apache.org>.
Github user zentol commented on a diff in the pull request:
https://github.com/apache/flink/pull/5062#discussion_r153798592
--- Diff: flink-queryable-state/flink-queryable-state-client-java/src/main/java/org/apache/flink/queryablestate/network/Client.java ---
@@ -133,7 +134,7 @@ public String getClientName() {
}
public CompletableFuture<RESP> sendRequest(final InetSocketAddress serverAddress, final REQ request) {
- if (shutDown.get()) {
+ if (!clientShutdownFuture.compareAndSet(null, null)) {
--- End diff --
this should use ´clientShutdownFuture.get() != null` for clarity.
---
[GitHub] flink pull request #5062: [FLINK-7880][QS] Wait for proper resource cleanup ...
Posted by zentol <gi...@git.apache.org>.
Github user zentol commented on a diff in the pull request:
https://github.com/apache/flink/pull/5062#discussion_r154950988
--- Diff: flink-queryable-state/flink-queryable-state-client-java/src/main/java/org/apache/flink/queryablestate/network/Client.java ---
@@ -191,9 +196,12 @@ public String getClientName() {
CompletableFuture.allOf(
connectionFutures.toArray(new CompletableFuture<?>[connectionFutures.size()])
).whenComplete((result, throwable) -> {
+
if (throwable != null) {
- newShutdownFuture.completeExceptionally(throwable);
- } else if (bootstrap != null) {
+ LOG.warn("Problem while shutting down the connections at the {}: {}", clientName, throwable);
--- End diff --
`LOG.warn("Problem while shutting down the connections at the {}", clientName, throwable);` instead?
---
[GitHub] flink pull request #5062: [FLINK-7880][QS] Wait for proper resource cleanup ...
Posted by zentol <gi...@git.apache.org>.
Github user zentol commented on a diff in the pull request:
https://github.com/apache/flink/pull/5062#discussion_r153794265
--- Diff: flink-queryable-state/flink-queryable-state-runtime/src/main/java/org/apache/flink/queryablestate/client/proxy/KvStateClientProxyImpl.java ---
@@ -96,7 +98,13 @@ public void start() throws Throwable {
@Override
public void shutdown() {
- super.shutdown();
+ try {
+ Time timeout = Time.seconds(10L);
+ shutdownServer(timeout).get(timeout.toMilliseconds(), TimeUnit.MILLISECONDS);
--- End diff --
why not just get(10, TimeUnit.SECONDS)?
---
[GitHub] flink pull request #5062: [FLINK-7880][QS] Wait for proper resource cleanup ...
Posted by zentol <gi...@git.apache.org>.
Github user zentol commented on a diff in the pull request:
https://github.com/apache/flink/pull/5062#discussion_r154888733
--- Diff: flink-queryable-state/flink-queryable-state-client-java/src/main/java/org/apache/flink/queryablestate/network/Client.java ---
@@ -133,7 +134,7 @@ public String getClientName() {
}
public CompletableFuture<RESP> sendRequest(final InetSocketAddress serverAddress, final REQ request) {
- if (shutDown.get()) {
+ if (!clientShutdownFuture.compareAndSet(null, null)) {
--- End diff --
But the atomicity doesn't get you anything. Which case wouldn't be covered?
This check can only handle the case when sendRequest() is called after shutdown(), regardless of whether you use compareAndSet() or get(), so it shouldn't make a difference.
---
[GitHub] flink pull request #5062: [FLINK-7880][QS] Wait for proper resource cleanup ...
Posted by zentol <gi...@git.apache.org>.
Github user zentol commented on a diff in the pull request:
https://github.com/apache/flink/pull/5062#discussion_r153790466
--- Diff: flink-queryable-state/flink-queryable-state-runtime/src/test/java/org/apache/flink/queryablestate/network/AbstractServerTest.java ---
@@ -86,77 +83,89 @@ public void testServerInitializationFailure() throws Throwable {
*/
@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);
+ // 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.assertEquals(7777L, server1.getServerAddress().getPort());
--- End diff --
these should be removed, or generalized to `portRangeStart <= server1.getServerAddress().getPort() <= portRangeEnd`.
---
[GitHub] flink pull request #5062: [FLINK-7880][QS] Wait for proper resource cleanup ...
Posted by kl0u <gi...@git.apache.org>.
Github user kl0u commented on a diff in the pull request:
https://github.com/apache/flink/pull/5062#discussion_r155212313
--- Diff: flink-queryable-state/flink-queryable-state-runtime/src/test/java/org/apache/flink/queryablestate/itcases/AbstractQueryableStateTestBase.java ---
@@ -359,12 +355,12 @@ public Integer getKey(Tuple2<Integer, Long> value) throws Exception {
} finally {
// Free cluster resources
if (jobId != null) {
- scala.concurrent.Future<CancellationSuccess> cancellation = cluster
- .getLeaderGateway(deadline.timeLeft())
+ cluster.getLeaderGateway(deadline.timeLeft())
.ask(new JobManagerMessages.CancelJob(jobId), deadline.timeLeft())
- .mapTo(ClassTag$.MODULE$.<JobManagerMessages.CancellationSuccess>apply(JobManagerMessages.CancellationSuccess.class));
+ .mapTo(ClassTag$.MODULE$.<CancellationSuccess>apply(CancellationSuccess.class));
- Await.ready(cancellation, deadline.timeLeft());
+ // we are not waiting for the cancellation to happen because the
+ // job has actually failed, as tested above.
--- End diff --
You are right. So I will just remove the `finally` block and check if the status of the job is `FAILED`.
---
[GitHub] flink pull request #5062: [FLINK-7880][QS] Wait for proper resource cleanup ...
Posted by zentol <gi...@git.apache.org>.
Github user zentol commented on a diff in the pull request:
https://github.com/apache/flink/pull/5062#discussion_r153804310
--- Diff: flink-queryable-state/flink-queryable-state-runtime/src/test/java/org/apache/flink/queryablestate/itcases/AbstractQueryableStateTestBase.java ---
@@ -359,12 +355,12 @@ public Integer getKey(Tuple2<Integer, Long> value) throws Exception {
} finally {
// Free cluster resources
if (jobId != null) {
- scala.concurrent.Future<CancellationSuccess> cancellation = cluster
- .getLeaderGateway(deadline.timeLeft())
+ cluster.getLeaderGateway(deadline.timeLeft())
.ask(new JobManagerMessages.CancelJob(jobId), deadline.timeLeft())
- .mapTo(ClassTag$.MODULE$.<JobManagerMessages.CancellationSuccess>apply(JobManagerMessages.CancellationSuccess.class));
+ .mapTo(ClassTag$.MODULE$.<CancellationSuccess>apply(CancellationSuccess.class));
- Await.ready(cancellation, deadline.timeLeft());
+ // we are not waiting for the cancellation to happen because the
+ // job has actually failed, as tested above.
--- End diff --
you can't guarantee this in the `finally` block. (for example if submitJobDetached failed but the job is actually running)
---
[GitHub] flink pull request #5062: [FLINK-7880][QS] Wait for proper resource cleanup ...
Posted by kl0u <gi...@git.apache.org>.
Github user kl0u commented on a diff in the pull request:
https://github.com/apache/flink/pull/5062#discussion_r155212865
--- Diff: flink-queryable-state/flink-queryable-state-runtime/src/test/java/org/apache/flink/queryablestate/itcases/HAAbstractQueryableStateTestBase.java ---
@@ -80,19 +80,18 @@ public static void setup(int proxyPortRangeStart, int serverPortRangeStart) {
@AfterClass
public static void tearDown() {
- if (cluster != null) {
+ try {
+ client.shutdownAndWait();
+
cluster.stop();
cluster.awaitTermination();
- }
- try {
zkServer.stop();
zkServer.close();
- } catch (Exception e) {
+
+ } catch (Throwable e) {
--- End diff --
Done.
---
[GitHub] flink pull request #5062: [FLINK-7880][QS] Wait for proper resource cleanup ...
Posted by zentol <gi...@git.apache.org>.
Github user zentol commented on a diff in the pull request:
https://github.com/apache/flink/pull/5062#discussion_r153799484
--- Diff: flink-queryable-state/flink-queryable-state-client-java/src/main/java/org/apache/flink/queryablestate/network/AbstractServerBase.java ---
@@ -251,34 +263,81 @@ private boolean attemptToBind(final int port) throws Throwable {
throw future.cause();
} catch (BindException e) {
log.debug("Failed to start {} on port {}: {}.", serverName, port, e.getMessage());
- shutdown();
+ try {
+ shutdownServer(Time.seconds(10L)).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 and then we check
+
+ log.warn("Problem while shutting down {}: {}", serverName, r.getMessage());
+ }
}
// any other type of exception we let it bubble up.
return false;
}
/**
* Shuts down the server and all related thread pools.
+ * @param timeout The time to wait for the shutdown process to complete.
+ * @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(Time timeout) throws InterruptedException {
+ CompletableFuture<Void> shutdownFuture = new CompletableFuture<>();
+ if (serverShutdownFuture.compareAndSet(null, shutdownFuture)) {
--- End diff --
calling shutdown twice in a row will result in a future being returned that never completes.
---
[GitHub] flink pull request #5062: [FLINK-7880][QS] Wait for proper resource cleanup ...
Posted by zentol <gi...@git.apache.org>.
Github user zentol commented on a diff in the pull request:
https://github.com/apache/flink/pull/5062#discussion_r155224199
--- Diff: flink-queryable-state/flink-queryable-state-runtime/src/test/java/org/apache/flink/queryablestate/network/ClientTest.java ---
@@ -224,6 +225,7 @@ public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception
assertEquals(expectedRequests, stats.getNumSuccessful());
assertEquals(expectedRequests, stats.getNumFailed());
} finally {
+ Exception exc = null;
--- End diff --
this could be moved into the if block.
---
[GitHub] flink pull request #5062: [FLINK-7880][QS] Wait for proper resource cleanup ...
Posted by kl0u <gi...@git.apache.org>.
Github user kl0u commented on a diff in the pull request:
https://github.com/apache/flink/pull/5062#discussion_r154886964
--- Diff: flink-queryable-state/flink-queryable-state-client-java/src/main/java/org/apache/flink/queryablestate/network/Client.java ---
@@ -166,28 +167,57 @@ public String getClientName() {
* Shuts down the client and closes all connections.
*
* <p>After a call to this method, all returned futures will be failed.
+ *
+ * @return A {@link CompletableFuture} that will be completed when the shutdown process is done.
*/
- public void shutdown() {
- if (shutDown.compareAndSet(false, true)) {
+ public CompletableFuture<Void> shutdown() {
+ final CompletableFuture<Void> newShutdownFuture = new CompletableFuture<>();
+ if (clientShutdownFuture.compareAndSet(null, newShutdownFuture)) {
+
+ final List<CompletableFuture<Void>> connectionFutures = new ArrayList<>();
+
for (Map.Entry<InetSocketAddress, EstablishedConnection> conn : establishedConnections.entrySet()) {
if (establishedConnections.remove(conn.getKey(), conn.getValue())) {
- conn.getValue().close();
+ connectionFutures.add(conn.getValue().close());
}
}
for (Map.Entry<InetSocketAddress, PendingConnection> conn : pendingConnections.entrySet()) {
if (pendingConnections.remove(conn.getKey()) != null) {
- conn.getValue().close();
+ connectionFutures.add(conn.getValue().close());
}
}
- if (bootstrap != null) {
- EventLoopGroup group = bootstrap.group();
- if (group != null) {
- group.shutdownGracefully(0L, 10L, TimeUnit.SECONDS);
+ CompletableFuture.allOf(
+ connectionFutures.toArray(new CompletableFuture<?>[connectionFutures.size()])
+ ).whenComplete((result, throwable) -> {
+ if (throwable != null) {
+ newShutdownFuture.completeExceptionally(throwable);
+ } else if (bootstrap != null) {
+ EventLoopGroup group = bootstrap.group();
+ if (group != null && !group.isShutdown()) {
+ group.shutdownGracefully(0L, 0L, TimeUnit.MILLISECONDS)
+ .addListener(finished -> {
+ if (finished.isSuccess()) {
+ newShutdownFuture.complete(null);
+ } else {
+ newShutdownFuture.completeExceptionally(finished.cause());
+ }
+ });
+ } else {
+ newShutdownFuture.complete(null);
+ }
+ } else {
+ newShutdownFuture.complete(null);
}
+ });
+
+ // check again if in the meantime another thread completed the future
+ if (clientShutdownFuture.compareAndSet(null, newShutdownFuture)) {
--- End diff --
You are right!
---
[GitHub] flink pull request #5062: [FLINK-7880][QS] Wait for proper resource cleanup ...
Posted by zentol <gi...@git.apache.org>.
Github user zentol commented on a diff in the pull request:
https://github.com/apache/flink/pull/5062#discussion_r153790485
--- Diff: flink-queryable-state/flink-queryable-state-runtime/src/test/java/org/apache/flink/queryablestate/network/AbstractServerTest.java ---
@@ -86,77 +83,89 @@ public void testServerInitializationFailure() throws Throwable {
*/
@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);
+ // 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.assertEquals(7777L, server1.getServerAddress().getPort());
- server2 = startServer("Test Server 2", 7777, 7778, 7779);
+ server2.start();
Assert.assertEquals(7778L, server2.getServerAddress().getPort());
--- End diff --
same as above
---
[GitHub] flink pull request #5062: [FLINK-7880][QS] Wait for proper resource cleanup ...
Posted by zentol <gi...@git.apache.org>.
Github user zentol commented on a diff in the pull request:
https://github.com/apache/flink/pull/5062#discussion_r153785916
--- Diff: flink-queryable-state/flink-queryable-state-client-java/src/main/java/org/apache/flink/queryablestate/network/Client.java ---
@@ -166,28 +167,57 @@ public String getClientName() {
* Shuts down the client and closes all connections.
*
* <p>After a call to this method, all returned futures will be failed.
+ *
+ * @return A {@link CompletableFuture} that will be completed when the shutdown process is done.
*/
- public void shutdown() {
- if (shutDown.compareAndSet(false, true)) {
+ public CompletableFuture<Void> shutdown() {
+ final CompletableFuture<Void> newShutdownFuture = new CompletableFuture<>();
+ if (clientShutdownFuture.compareAndSet(null, newShutdownFuture)) {
+
+ final List<CompletableFuture<Void>> connectionFutures = new ArrayList<>();
+
for (Map.Entry<InetSocketAddress, EstablishedConnection> conn : establishedConnections.entrySet()) {
if (establishedConnections.remove(conn.getKey(), conn.getValue())) {
- conn.getValue().close();
+ connectionFutures.add(conn.getValue().close());
}
}
for (Map.Entry<InetSocketAddress, PendingConnection> conn : pendingConnections.entrySet()) {
if (pendingConnections.remove(conn.getKey()) != null) {
- conn.getValue().close();
+ connectionFutures.add(conn.getValue().close());
}
}
- if (bootstrap != null) {
- EventLoopGroup group = bootstrap.group();
- if (group != null) {
- group.shutdownGracefully(0L, 10L, TimeUnit.SECONDS);
+ CompletableFuture.allOf(
+ connectionFutures.toArray(new CompletableFuture<?>[connectionFutures.size()])
+ ).whenComplete((result, throwable) -> {
+ if (throwable != null) {
+ newShutdownFuture.completeExceptionally(throwable);
+ } else if (bootstrap != null) {
+ EventLoopGroup group = bootstrap.group();
+ if (group != null && !group.isShutdown()) {
+ group.shutdownGracefully(0L, 0L, TimeUnit.MILLISECONDS)
+ .addListener(finished -> {
+ if (finished.isSuccess()) {
+ newShutdownFuture.complete(null);
+ } else {
+ newShutdownFuture.completeExceptionally(finished.cause());
+ }
+ });
+ } else {
+ newShutdownFuture.complete(null);
+ }
+ } else {
+ newShutdownFuture.complete(null);
}
+ });
+
+ // check again if in the meantime another thread completed the future
+ if (clientShutdownFuture.compareAndSet(null, newShutdownFuture)) {
--- End diff --
This can never be true. We can only arrive here if `clientShutdownFuture.compareAndSet(null, newShutdownFuture)` succeeds, and there is no other code path that ever sets it to null again.
---
[GitHub] flink issue #5062: [FLINK-7880][QS] Wait for proper resource cleanup after e...
Posted by zentol <gi...@git.apache.org>.
Github user zentol commented on the issue:
https://github.com/apache/flink/pull/5062
👍
---
[GitHub] flink pull request #5062: [FLINK-7880][QS] Wait for proper resource cleanup ...
Posted by zentol <gi...@git.apache.org>.
Github user zentol commented on a diff in the pull request:
https://github.com/apache/flink/pull/5062#discussion_r153802908
--- Diff: flink-queryable-state/flink-queryable-state-client-java/src/main/java/org/apache/flink/queryablestate/network/Client.java ---
@@ -486,27 +542,25 @@ private boolean close(Throwable cause) {
@Override
public void onRequestResult(long requestId, RESP response) {
TimestampedCompletableFuture pending = pendingRequests.remove(requestId);
- if (pending != null && pending.complete(response)) {
+ if (pending != null && !pending.isDone()) {
long durationMillis = (System.nanoTime() - pending.getTimestamp()) / 1_000_000L;
stats.reportSuccessfulRequest(durationMillis);
+ pending.complete(response);
}
}
@Override
public void onRequestFailure(long requestId, Throwable cause) {
TimestampedCompletableFuture pending = pendingRequests.remove(requestId);
- if (pending != null && pending.completeExceptionally(cause)) {
+ if (pending != null && !pending.isDone()) {
stats.reportFailedRequest();
+ pending.completeExceptionally(cause);
}
}
@Override
public void onFailure(Throwable cause) {
- if (close(cause)) {
- // Remove from established channels, otherwise future
- // requests will be handled by this failed channel.
- establishedConnections.remove(serverAddress, this);
- }
+ close(cause).thenAccept(cancelled -> establishedConnections.remove(serverAddress, this));
--- End diff --
shouldn't we remove the connection in any case, since if we can't close *something* is probably wrong with it anyway?
---
[GitHub] flink pull request #5062: [FLINK-7880][QS] Wait for proper resource cleanup ...
Posted by zentol <gi...@git.apache.org>.
Github user zentol commented on a diff in the pull request:
https://github.com/apache/flink/pull/5062#discussion_r153786954
--- Diff: flink-queryable-state/flink-queryable-state-client-java/src/main/java/org/apache/flink/queryablestate/network/Client.java ---
@@ -312,32 +345,41 @@ private void handInChannel(Channel channel) {
/**
* Close the connecting channel with a ClosedChannelException.
*/
- private void close() {
- close(new ClosedChannelException());
+ private CompletableFuture<Void> close() {
+ return close(new ClosedChannelException());
}
/**
* Close the connecting channel with an Exception (can be {@code null})
* or forward to the established channel.
*/
- private void close(Throwable cause) {
- synchronized (connectLock) {
- if (!closed) {
+ private CompletableFuture<Void> close(Throwable cause) {
+ CompletableFuture<Void> future = new CompletableFuture<>();
+ if (connectionShutdownFuture.compareAndSet(null, future)) {
+ synchronized (connectLock) {
if (failureCause == null) {
failureCause = cause;
}
+ closed = true;
--- End diff --
this field should be redundant, as `(connectionShutdownFuture.get() != null) == closed` should always hold.
---
[GitHub] flink pull request #5062: [FLINK-7880][QS] Wait for proper resource cleanup ...
Posted by zentol <gi...@git.apache.org>.
Github user zentol commented on a diff in the pull request:
https://github.com/apache/flink/pull/5062#discussion_r153801762
--- Diff: flink-queryable-state/flink-queryable-state-client-java/src/main/java/org/apache/flink/queryablestate/network/Client.java ---
@@ -422,20 +467,31 @@ void close() {
* @param cause The cause to close the channel with.
* @return Channel close future
*/
- private boolean close(Throwable cause) {
- if (failureCause.compareAndSet(null, cause)) {
- channel.close();
- stats.reportInactiveConnection();
+ private CompletableFuture<Void> close(Throwable cause) {
+ final CompletableFuture<Void> shutdownFuture = new CompletableFuture<>();
- for (long requestId : pendingRequests.keySet()) {
- TimestampedCompletableFuture pending = pendingRequests.remove(requestId);
- if (pending != null && pending.completeExceptionally(cause)) {
- stats.reportFailedRequest();
+ if (connectionShutdownFuture.compareAndSet(null, shutdownFuture) &&
--- End diff --
this can result in odd race conditions, where one thread has set the shutdownFuture, but another one the failureCause. I would suggest to create a container for the future and exception and update them atomically.
---
[GitHub] flink pull request #5062: [FLINK-7880][QS] Wait for proper resource cleanup ...
Posted by kl0u <gi...@git.apache.org>.
Github user kl0u commented on a diff in the pull request:
https://github.com/apache/flink/pull/5062#discussion_r155201806
--- Diff: flink-queryable-state/flink-queryable-state-client-java/src/main/java/org/apache/flink/queryablestate/network/AbstractServerBase.java ---
@@ -251,34 +263,81 @@ private boolean attemptToBind(final int port) throws Throwable {
throw future.cause();
} catch (BindException e) {
log.debug("Failed to start {} on port {}: {}.", serverName, port, e.getMessage());
- shutdown();
+ try {
+ shutdownServer(Time.seconds(10L)).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 and then we check
+
+ log.warn("Problem while shutting down {}: {}", serverName, r.getMessage());
+ }
}
// any other type of exception we let it bubble up.
return false;
}
/**
* Shuts down the server and all related thread pools.
+ * @param timeout The time to wait for the shutdown process to complete.
+ * @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(Time timeout) throws InterruptedException {
+ 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(timeout.toMilliseconds(), TimeUnit.MILLISECONDS, queryExecutor);
+ }
+ });
+
+ CompletableFuture.allOf(
+ queryExecShutdownFuture, groupShutdownFuture, handlerShutdownFuture
+ ).whenComplete((result, throwable) -> {
+ if (throwable != null) {
+ shutdownFuture.completeExceptionally(throwable);
+ } else {
+ shutdownFuture.complete(null);
+ }
+ });
}
- serverAddress = null;
+ return serverShutdownFuture.get();
--- End diff --
Ok I will add the check in the `start()`.
---
[GitHub] flink pull request #5062: [FLINK-7880][QS] Wait for proper resource cleanup ...
Posted by zentol <gi...@git.apache.org>.
Github user zentol commented on a diff in the pull request:
https://github.com/apache/flink/pull/5062#discussion_r153790032
--- Diff: flink-queryable-state/flink-queryable-state-runtime/src/test/java/org/apache/flink/queryablestate/network/AbstractServerTest.java ---
@@ -60,23 +63,17 @@ public void testServerInitializationFailure() throws Throwable {
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);
+ try (
+ TestServer server1 = new TestServer("Test Server 1", new DisabledKvStateRequestStats(), portList.iterator());
--- End diff --
a safer pattern is to give `server1` multiple ports to choose from and specifically start `server2` on the port `server1` eventually started with.
---
[GitHub] flink pull request #5062: [FLINK-7880][QS] Wait for proper resource cleanup ...
Posted by kl0u <gi...@git.apache.org>.
Github user kl0u commented on a diff in the pull request:
https://github.com/apache/flink/pull/5062#discussion_r154890905
--- Diff: flink-queryable-state/flink-queryable-state-client-java/src/main/java/org/apache/flink/queryablestate/network/Client.java ---
@@ -486,27 +542,25 @@ private boolean close(Throwable cause) {
@Override
public void onRequestResult(long requestId, RESP response) {
TimestampedCompletableFuture pending = pendingRequests.remove(requestId);
- if (pending != null && pending.complete(response)) {
+ if (pending != null && !pending.isDone()) {
long durationMillis = (System.nanoTime() - pending.getTimestamp()) / 1_000_000L;
stats.reportSuccessfulRequest(durationMillis);
+ pending.complete(response);
}
}
@Override
public void onRequestFailure(long requestId, Throwable cause) {
TimestampedCompletableFuture pending = pendingRequests.remove(requestId);
- if (pending != null && pending.completeExceptionally(cause)) {
+ if (pending != null && !pending.isDone()) {
stats.reportFailedRequest();
+ pending.completeExceptionally(cause);
}
}
@Override
public void onFailure(Throwable cause) {
- if (close(cause)) {
- // Remove from established channels, otherwise future
- // requests will be handled by this failed channel.
- establishedConnections.remove(serverAddress, this);
- }
+ close(cause).thenAccept(cancelled -> establishedConnections.remove(serverAddress, this));
--- End diff --
Yes.
---
[GitHub] flink pull request #5062: [FLINK-7880][QS] Wait for proper resource cleanup ...
Posted by kl0u <gi...@git.apache.org>.
Github user kl0u commented on a diff in the pull request:
https://github.com/apache/flink/pull/5062#discussion_r154890227
--- Diff: flink-queryable-state/flink-queryable-state-client-java/src/main/java/org/apache/flink/queryablestate/network/Client.java ---
@@ -312,32 +345,41 @@ private void handInChannel(Channel channel) {
/**
* Close the connecting channel with a ClosedChannelException.
*/
- private void close() {
- close(new ClosedChannelException());
+ private CompletableFuture<Void> close() {
+ return close(new ClosedChannelException());
}
/**
* Close the connecting channel with an Exception (can be {@code null})
* or forward to the established channel.
*/
- private void close(Throwable cause) {
- synchronized (connectLock) {
- if (!closed) {
+ private CompletableFuture<Void> close(Throwable cause) {
+ CompletableFuture<Void> future = new CompletableFuture<>();
+ if (connectionShutdownFuture.compareAndSet(null, future)) {
+ synchronized (connectLock) {
if (failureCause == null) {
failureCause = cause;
}
+ closed = true;
--- End diff --
Yes. I will remove that.
---
[GitHub] flink pull request #5062: [FLINK-7880][QS] Wait for proper resource cleanup ...
Posted by zentol <gi...@git.apache.org>.
Github user zentol commented on a diff in the pull request:
https://github.com/apache/flink/pull/5062#discussion_r153797851
--- Diff: flink-queryable-state/flink-queryable-state-client-java/src/main/java/org/apache/flink/queryablestate/network/AbstractServerBase.java ---
@@ -251,34 +263,81 @@ private boolean attemptToBind(final int port) throws Throwable {
throw future.cause();
} catch (BindException e) {
log.debug("Failed to start {} on port {}: {}.", serverName, port, e.getMessage());
- shutdown();
+ try {
+ shutdownServer(Time.seconds(10L)).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 and then we check
+
+ log.warn("Problem while shutting down {}: {}", serverName, r.getMessage());
+ }
}
// any other type of exception we let it bubble up.
return false;
}
/**
* Shuts down the server and all related thread pools.
+ * @param timeout The time to wait for the shutdown process to complete.
+ * @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(Time timeout) throws InterruptedException {
--- End diff --
i would remove this timeout argument and simply define large timeout for shutting down the `queryExecutor`.
---
[GitHub] flink issue #5062: [FLINK-7880][QS] Wait for proper resource cleanup after e...
Posted by kl0u <gi...@git.apache.org>.
Github user kl0u commented on the issue:
https://github.com/apache/flink/pull/5062
Fixed @zentol !
---
[GitHub] flink pull request #5062: [FLINK-7880][QS] Wait for proper resource cleanup ...
Posted by kl0u <gi...@git.apache.org>.
Github user kl0u commented on a diff in the pull request:
https://github.com/apache/flink/pull/5062#discussion_r154885503
--- Diff: flink-queryable-state/flink-queryable-state-client-java/src/main/java/org/apache/flink/queryablestate/network/Client.java ---
@@ -133,7 +134,7 @@ public String getClientName() {
}
public CompletableFuture<RESP> sendRequest(final InetSocketAddress serverAddress, final REQ request) {
- if (shutDown.get()) {
+ if (!clientShutdownFuture.compareAndSet(null, null)) {
--- End diff --
Well, I agree that the code looks clearer, but the `compareAndSet` guarantees atomicity. While `get()` and then compare, does not.
---
[GitHub] flink pull request #5062: [FLINK-7880][QS] Wait for proper resource cleanup ...
Posted by zentol <gi...@git.apache.org>.
Github user zentol commented on a diff in the pull request:
https://github.com/apache/flink/pull/5062#discussion_r155224312
--- Diff: flink-queryable-state/flink-queryable-state-runtime/src/test/java/org/apache/flink/queryablestate/network/ClientTest.java ---
@@ -234,9 +236,14 @@ public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception
client.shutdown().get(10L, TimeUnit.SECONDS);
} catch (Exception e) {
- e.printStackTrace();
+ exc = e;
+ LOG.error(ExceptionUtils.stringifyException(e));
--- End diff --
use `LOG.error("An exception occurred while shutting down netty.", e)` instead
---
[GitHub] flink pull request #5062: [FLINK-7880][QS] Wait for proper resource cleanup ...
Posted by zentol <gi...@git.apache.org>.
Github user zentol commented on a diff in the pull request:
https://github.com/apache/flink/pull/5062#discussion_r153786058
--- Diff: flink-queryable-state/flink-queryable-state-client-java/src/main/java/org/apache/flink/queryablestate/network/Client.java ---
@@ -300,7 +333,7 @@ private void handInChannel(Channel channel) {
// Check shut down for possible race with shut down. We
// don't want any lingering connections after shut down,
// which can happen if we don't check this here.
- if (shutDown.get()) {
+ if (!clientShutdownFuture.compareAndSet(null, null)) {
--- End diff --
for clarity we should call `get()` instead of `compareAndSet` since the setting part is a no-op anyway.
---
[GitHub] flink pull request #5062: [FLINK-7880][QS] Wait for proper resource cleanup ...
Posted by zentol <gi...@git.apache.org>.
Github user zentol commented on a diff in the pull request:
https://github.com/apache/flink/pull/5062#discussion_r153789622
--- Diff: flink-queryable-state/flink-queryable-state-runtime/src/test/java/org/apache/flink/queryablestate/network/ClientTest.java ---
@@ -218,7 +225,18 @@ public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception
assertEquals(expectedRequests, stats.getNumFailed());
} finally {
if (client != null) {
- client.shutdown();
+ try {
+
+ // todo 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 and then we check
+
+ client.shutdown().get(10L, TimeUnit.SECONDS);
+ } catch (Exception e) {
+ e.printStackTrace();
+ }
+ Assert.assertTrue(client.isEventGroupShutdown());
--- End diff --
I would prefer if we would throw `e` in this and similar cases.
---
[GitHub] flink pull request #5062: [FLINK-7880][QS] Wait for proper resource cleanup ...
Posted by kl0u <gi...@git.apache.org>.
Github user kl0u commented on a diff in the pull request:
https://github.com/apache/flink/pull/5062#discussion_r154884801
--- Diff: flink-queryable-state/flink-queryable-state-client-java/src/main/java/org/apache/flink/queryablestate/network/AbstractServerBase.java ---
@@ -251,34 +263,81 @@ private boolean attemptToBind(final int port) throws Throwable {
throw future.cause();
} catch (BindException e) {
log.debug("Failed to start {} on port {}: {}.", serverName, port, e.getMessage());
- shutdown();
+ try {
+ shutdownServer(Time.seconds(10L)).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 and then we check
+
+ log.warn("Problem while shutting down {}: {}", serverName, r.getMessage());
+ }
}
// any other type of exception we let it bubble up.
return false;
}
/**
* Shuts down the server and all related thread pools.
+ * @param timeout The time to wait for the shutdown process to complete.
+ * @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(Time timeout) throws InterruptedException {
+ CompletableFuture<Void> shutdownFuture = new CompletableFuture<>();
+ if (serverShutdownFuture.compareAndSet(null, shutdownFuture)) {
--- End diff --
Here the idea is that we atomically set the future the first time (`serverShutdownFuture.compareAndSet(null, shutdownFuture)`), and then, at each subsequent call we return that exact future (`serverShutdownFuture.get();`). So I cannot see how this can lead to returning a future that never completes.
---
[GitHub] flink pull request #5062: [FLINK-7880][QS] Wait for proper resource cleanup ...
Posted by zentol <gi...@git.apache.org>.
Github user zentol commented on a diff in the pull request:
https://github.com/apache/flink/pull/5062#discussion_r153797256
--- Diff: flink-queryable-state/flink-queryable-state-client-java/src/main/java/org/apache/flink/queryablestate/network/AbstractServerBase.java ---
@@ -251,34 +263,81 @@ private boolean attemptToBind(final int port) throws Throwable {
throw future.cause();
} catch (BindException e) {
log.debug("Failed to start {} on port {}: {}.", serverName, port, e.getMessage());
- shutdown();
+ try {
+ shutdownServer(Time.seconds(10L)).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 and then we check
--- End diff --
we don't check anything here.
---
[GitHub] flink pull request #5062: [FLINK-7880][QS] Wait for proper resource cleanup ...
Posted by zentol <gi...@git.apache.org>.
Github user zentol commented on a diff in the pull request:
https://github.com/apache/flink/pull/5062#discussion_r153793789
--- Diff: flink-queryable-state/flink-queryable-state-runtime/src/test/java/org/apache/flink/queryablestate/itcases/NonHAAbstractQueryableStateTestBase.java ---
@@ -68,11 +68,14 @@ public static void setup(int proxyPortRangeStart, int serverPortRangeStart) {
@AfterClass
public static void tearDown() {
try {
+ client.shutdownAndWait();
+
cluster.stop();
- } catch (Exception e) {
+ cluster.awaitTermination();
+
+ } catch (Throwable e) {
--- End diff --
try-catch block is unnecessary
---
[GitHub] flink pull request #5062: [FLINK-7880][QS] Wait for proper resource cleanup ...
Posted by zentol <gi...@git.apache.org>.
Github user zentol commented on a diff in the pull request:
https://github.com/apache/flink/pull/5062#discussion_r154952648
--- Diff: flink-queryable-state/flink-queryable-state-runtime/src/test/java/org/apache/flink/queryablestate/network/AbstractServerTest.java ---
@@ -66,14 +65,15 @@ public void testServerInitializationFailure() throws Throwable {
List<Integer> portList = new ArrayList<>();
portList.add(7777);
- try (
- TestServer server1 = new TestServer("Test Server 1", new DisabledKvStateRequestStats(), portList.iterator());
- TestServer server2 = new TestServer("Test Server 2", new DisabledKvStateRequestStats(), portList.iterator())
- ) {
+ try (TestServer server1 = new TestServer("Test Server 1", new DisabledKvStateRequestStats(), portList.iterator())) {
server1.start();
- Assert.assertEquals(7777L, server1.getServerAddress().getPort());
- server2.start();
+ List<Integer> occupiedPortList = new ArrayList<>();
+ occupiedPortList.add(server1.getServerAddress().getPort());
--- End diff --
could use `Collections.singletonList()` here
---
[GitHub] flink pull request #5062: [FLINK-7880][QS] Wait for proper resource cleanup ...
Posted by zentol <gi...@git.apache.org>.
Github user zentol commented on a diff in the pull request:
https://github.com/apache/flink/pull/5062#discussion_r153804523
--- Diff: flink-queryable-state/flink-queryable-state-runtime/src/test/java/org/apache/flink/queryablestate/itcases/AbstractQueryableStateTestBase.java ---
@@ -434,13 +432,16 @@ public Integer getKey(Tuple2<Integer, Long> value) throws Exception {
* contains a wrong jobId or wrong queryable state name.
*/
@Test
+ @Ignore
--- End diff --
why?
---
[GitHub] flink pull request #5062: [FLINK-7880][QS] Wait for proper resource cleanup ...
Posted by zentol <gi...@git.apache.org>.
Github user zentol commented on a diff in the pull request:
https://github.com/apache/flink/pull/5062#discussion_r153794621
--- Diff: flink-queryable-state/flink-queryable-state-client-java/src/main/java/org/apache/flink/queryablestate/client/QueryableStateClient.java ---
@@ -108,9 +108,33 @@ public QueryableStateClient(final InetAddress remoteAddress, final int remotePor
new DisabledKvStateRequestStats());
}
- /** Shuts down the client. */
- public void shutdown() {
- client.shutdown();
+ /**
+ * Shuts down the client and returns a {@link CompletableFuture} that
+ * will be completed when the shutdown process is completed.
+ *
+ * <p>If an exception is thrown for any reason, then the returned future
+ * will be completed exceptionally with that exception.
+ *
+ * @return A {@link CompletableFuture} for further handling of the
+ * shutdown result.
+ */
+ public CompletableFuture<?> shutdownAndHandle() {
+ return client.shutdown();
+ }
+
+ /**
+ * Shuts down the client and waits until shutdown is completed.
+ *
+ * <p>If an exception is thrown, a warning is printed containing
--- End diff --
the warning is logged, not printed (generally implies stdout).
---
[GitHub] flink pull request #5062: [FLINK-7880][QS] Wait for proper resource cleanup ...
Posted by zentol <gi...@git.apache.org>.
Github user zentol commented on a diff in the pull request:
https://github.com/apache/flink/pull/5062#discussion_r153798768
--- Diff: flink-queryable-state/flink-queryable-state-client-java/src/main/java/org/apache/flink/queryablestate/network/Client.java ---
@@ -166,28 +167,57 @@ public String getClientName() {
* Shuts down the client and closes all connections.
*
* <p>After a call to this method, all returned futures will be failed.
+ *
+ * @return A {@link CompletableFuture} that will be completed when the shutdown process is done.
*/
- public void shutdown() {
- if (shutDown.compareAndSet(false, true)) {
+ public CompletableFuture<Void> shutdown() {
+ final CompletableFuture<Void> newShutdownFuture = new CompletableFuture<>();
--- End diff --
calling shutdown twice in a row will result in a future being returned that never completes.
---
[GitHub] flink pull request #5062: [FLINK-7880][QS] Wait for proper resource cleanup ...
Posted by zentol <gi...@git.apache.org>.
Github user zentol commented on a diff in the pull request:
https://github.com/apache/flink/pull/5062#discussion_r153788979
--- Diff: flink-queryable-state/flink-queryable-state-runtime/src/test/java/org/apache/flink/queryablestate/network/ClientTest.java ---
@@ -95,15 +97,20 @@
private static final Logger LOG = LoggerFactory.getLogger(ClientTest.class);
+ private static final FiniteDuration TEST_TIMEOUT = new FiniteDuration(20L, TimeUnit.SECONDS);
+
// Thread pool for client bootstrap (shared between tests)
- private static final NioEventLoopGroup NIO_GROUP = new NioEventLoopGroup();
+ private NioEventLoopGroup nioGroup;
- private static final FiniteDuration TEST_TIMEOUT = new FiniteDuration(10L, TimeUnit.SECONDS);
+ @Before
+ public void setUp() throws Exception {
+ nioGroup = new NioEventLoopGroup();
--- End diff --
this method isn't strictly necessary; you can also do `private final NioEventLoopGroup nioGroup = new NioEventLoopGroup()`. Then you would also no longer need the null-check in `@After`.
---
[GitHub] flink pull request #5062: [FLINK-7880][QS] Wait for proper resource cleanup ...
Posted by zentol <gi...@git.apache.org>.
Github user zentol commented on a diff in the pull request:
https://github.com/apache/flink/pull/5062#discussion_r153796142
--- Diff: flink-queryable-state/flink-queryable-state-client-java/src/main/java/org/apache/flink/queryablestate/network/AbstractServerBase.java ---
@@ -205,6 +212,11 @@ public void start() throws Throwable {
private boolean attemptToBind(final int port) throws Throwable {
log.debug("Attempting to start {} on port {}.", serverName, port);
+ // here we reset the future every time because in case of failure
+ // to bind, we call shutdown here and this may interfere with future
+ // shutdown attempts.
+ this.serverShutdownFuture.getAndSet(null);
--- End diff --
If we intend to only support a single start of a server this should use compareAndSet and fail immediately if a shutdown is in progress.
If we intend to support multiple starts of a server we should guard start() and shutdown() with a lock since otherwise you easily end up in an inconsistent state.
---
[GitHub] flink pull request #5062: [FLINK-7880][QS] Wait for proper resource cleanup ...
Posted by zentol <gi...@git.apache.org>.
Github user zentol commented on a diff in the pull request:
https://github.com/apache/flink/pull/5062#discussion_r153801913
--- Diff: flink-queryable-state/flink-queryable-state-client-java/src/main/java/org/apache/flink/queryablestate/network/Client.java ---
@@ -422,20 +467,31 @@ void close() {
* @param cause The cause to close the channel with.
* @return Channel close future
*/
- private boolean close(Throwable cause) {
- if (failureCause.compareAndSet(null, cause)) {
- channel.close();
- stats.reportInactiveConnection();
+ private CompletableFuture<Void> close(Throwable cause) {
+ final CompletableFuture<Void> shutdownFuture = new CompletableFuture<>();
- for (long requestId : pendingRequests.keySet()) {
- TimestampedCompletableFuture pending = pendingRequests.remove(requestId);
- if (pending != null && pending.completeExceptionally(cause)) {
- stats.reportFailedRequest();
+ if (connectionShutdownFuture.compareAndSet(null, shutdownFuture) &&
--- End diff --
Let's also log other shutdown attempts as DEBUG.
---
[GitHub] flink pull request #5062: [FLINK-7880][QS] Wait for proper resource cleanup ...
Posted by kl0u <gi...@git.apache.org>.
Github user kl0u commented on a diff in the pull request:
https://github.com/apache/flink/pull/5062#discussion_r154885936
--- Diff: flink-queryable-state/flink-queryable-state-client-java/src/main/java/org/apache/flink/queryablestate/network/Client.java ---
@@ -166,28 +167,57 @@ public String getClientName() {
* Shuts down the client and closes all connections.
*
* <p>After a call to this method, all returned futures will be failed.
+ *
+ * @return A {@link CompletableFuture} that will be completed when the shutdown process is done.
*/
- public void shutdown() {
- if (shutDown.compareAndSet(false, true)) {
+ public CompletableFuture<Void> shutdown() {
+ final CompletableFuture<Void> newShutdownFuture = new CompletableFuture<>();
--- End diff --
This is the same pattern as in the case of the `AbstractServerBase` ;)
---
[GitHub] flink pull request #5062: [FLINK-7880][QS] Wait for proper resource cleanup ...
Posted by zentol <gi...@git.apache.org>.
Github user zentol commented on a diff in the pull request:
https://github.com/apache/flink/pull/5062#discussion_r154955042
--- Diff: flink-queryable-state/flink-queryable-state-runtime/src/test/java/org/apache/flink/queryablestate/itcases/AbstractQueryableStateTestBase.java ---
@@ -1480,7 +1259,52 @@ public String fold(String accumulator, Tuple2<Integer, Long> value) throws Excep
///// General Utility Methods //////
- private CompletableFuture<TestingJobManagerMessages.JobStatusIs> notifyWhenJobStatusIs(
+ /**
+ * A wrapper of the job graph that makes sure to cancell the job and wait for
--- End diff --
typo: cancell -> cancel
---
[GitHub] flink issue #5062: [FLINK-7880][QS] Wait for proper resource cleanup after e...
Posted by zentol <gi...@git.apache.org>.
Github user zentol commented on the issue:
https://github.com/apache/flink/pull/5062
Your follow-up reference points to this very PR :P
---
[GitHub] flink pull request #5062: [FLINK-7880][QS] Wait for proper resource cleanup ...
Posted by kl0u <gi...@git.apache.org>.
Github user kl0u commented on a diff in the pull request:
https://github.com/apache/flink/pull/5062#discussion_r155212885
--- Diff: flink-queryable-state/flink-queryable-state-runtime/src/test/java/org/apache/flink/queryablestate/itcases/NonHAAbstractQueryableStateTestBase.java ---
@@ -68,11 +68,14 @@ public static void setup(int proxyPortRangeStart, int serverPortRangeStart) {
@AfterClass
public static void tearDown() {
try {
+ client.shutdownAndWait();
+
cluster.stop();
- } catch (Exception e) {
+ cluster.awaitTermination();
+
+ } catch (Throwable e) {
--- End diff --
Done.
---
[GitHub] flink pull request #5062: [FLINK-7880][QS] Wait for proper resource cleanup ...
Posted by zentol <gi...@git.apache.org>.
Github user zentol commented on a diff in the pull request:
https://github.com/apache/flink/pull/5062#discussion_r153798113
--- Diff: flink-queryable-state/flink-queryable-state-client-java/src/main/java/org/apache/flink/queryablestate/network/AbstractServerBase.java ---
@@ -251,34 +263,81 @@ private boolean attemptToBind(final int port) throws Throwable {
throw future.cause();
} catch (BindException e) {
log.debug("Failed to start {} on port {}: {}.", serverName, port, e.getMessage());
- shutdown();
+ try {
+ shutdownServer(Time.seconds(10L)).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 and then we check
+
+ log.warn("Problem while shutting down {}: {}", serverName, r.getMessage());
+ }
}
// any other type of exception we let it bubble up.
return false;
}
/**
* Shuts down the server and all related thread pools.
+ * @param timeout The time to wait for the shutdown process to complete.
+ * @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(Time timeout) throws InterruptedException {
+ 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(timeout.toMilliseconds(), TimeUnit.MILLISECONDS, queryExecutor);
+ }
+ });
+
+ CompletableFuture.allOf(
+ queryExecShutdownFuture, groupShutdownFuture, handlerShutdownFuture
+ ).whenComplete((result, throwable) -> {
+ if (throwable != null) {
+ shutdownFuture.completeExceptionally(throwable);
+ } else {
+ shutdownFuture.complete(null);
+ }
+ });
}
- serverAddress = null;
+ return serverShutdownFuture.get();
--- End diff --
if `serverShutdownFuture.get != null` is meant to signal a shutdown in progress, then this method should set it back to null before returning.
---
[GitHub] flink pull request #5062: [FLINK-7880][QS] Wait for proper resource cleanup ...
Posted by zentol <gi...@git.apache.org>.
Github user zentol commented on a diff in the pull request:
https://github.com/apache/flink/pull/5062#discussion_r153799197
--- Diff: flink-queryable-state/flink-queryable-state-client-java/src/main/java/org/apache/flink/queryablestate/network/Client.java ---
@@ -166,28 +167,57 @@ public String getClientName() {
* Shuts down the client and closes all connections.
*
* <p>After a call to this method, all returned futures will be failed.
+ *
+ * @return A {@link CompletableFuture} that will be completed when the shutdown process is done.
*/
- public void shutdown() {
- if (shutDown.compareAndSet(false, true)) {
+ public CompletableFuture<Void> shutdown() {
+ final CompletableFuture<Void> newShutdownFuture = new CompletableFuture<>();
+ if (clientShutdownFuture.compareAndSet(null, newShutdownFuture)) {
+
+ final List<CompletableFuture<Void>> connectionFutures = new ArrayList<>();
+
for (Map.Entry<InetSocketAddress, EstablishedConnection> conn : establishedConnections.entrySet()) {
if (establishedConnections.remove(conn.getKey(), conn.getValue())) {
- conn.getValue().close();
+ connectionFutures.add(conn.getValue().close());
}
}
for (Map.Entry<InetSocketAddress, PendingConnection> conn : pendingConnections.entrySet()) {
if (pendingConnections.remove(conn.getKey()) != null) {
- conn.getValue().close();
+ connectionFutures.add(conn.getValue().close());
}
}
- if (bootstrap != null) {
- EventLoopGroup group = bootstrap.group();
- if (group != null) {
- group.shutdownGracefully(0L, 10L, TimeUnit.SECONDS);
+ CompletableFuture.allOf(
+ connectionFutures.toArray(new CompletableFuture<?>[connectionFutures.size()])
+ ).whenComplete((result, throwable) -> {
+ if (throwable != null) {
+ newShutdownFuture.completeExceptionally(throwable);
+ } else if (bootstrap != null) {
--- End diff --
this means that the bootstrap is not shutdown if a connection cannot be closed for whatever reason, which is rather odd. We should still at least try to shut it down.
---
[GitHub] flink pull request #5062: [FLINK-7880][QS] Wait for proper resource cleanup ...
Posted by zentol <gi...@git.apache.org>.
Github user zentol commented on a diff in the pull request:
https://github.com/apache/flink/pull/5062#discussion_r155225199
--- Diff: flink-queryable-state/flink-queryable-state-runtime/src/test/java/org/apache/flink/queryablestate/itcases/AbstractQueryableStateTestBase.java ---
@@ -260,89 +260,90 @@ public void testDuplicateRegistrationFailsJob() throws Exception {
final Deadline deadline = TEST_TIMEOUT.fromNow();
final int numKeys = 256;
- JobID jobId = null;
+ StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
+ env.setStateBackend(stateBackend);
+ env.setParallelism(maxParallelism);
+ // Very important, because cluster is shared between tests and we
+ // don't explicitly check that all slots are available before
+ // submitting.
+ env.setRestartStrategy(RestartStrategies.fixedDelayRestart(Integer.MAX_VALUE, 1000L));
- try {
- //
- // Test program
- //
- StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
- env.setStateBackend(stateBackend);
- env.setParallelism(maxParallelism);
- // Very important, because cluster is shared between tests and we
- // don't explicitly check that all slots are available before
- // submitting.
- env.setRestartStrategy(RestartStrategies.fixedDelayRestart(Integer.MAX_VALUE, 1000L));
-
- DataStream<Tuple2<Integer, Long>> source = env
- .addSource(new TestKeyRangeSource(numKeys));
-
- // Reducing state
- ReducingStateDescriptor<Tuple2<Integer, Long>> reducingState = new ReducingStateDescriptor<>(
- "any-name",
- new SumReduce(),
- source.getType());
-
- final String queryName = "duplicate-me";
-
- final QueryableStateStream<Integer, Tuple2<Integer, Long>> queryableState =
- source.keyBy(new KeySelector<Tuple2<Integer, Long>, Integer>() {
- private static final long serialVersionUID = -4126824763829132959L;
-
- @Override
- public Integer getKey(Tuple2<Integer, Long> value) throws Exception {
- return value.f0;
- }
- }).asQueryableState(queryName, reducingState);
+ DataStream<Tuple2<Integer, Long>> source = env.addSource(new TestKeyRangeSource(numKeys));
- final QueryableStateStream<Integer, Tuple2<Integer, Long>> duplicate =
- source.keyBy(new KeySelector<Tuple2<Integer, Long>, Integer>() {
- private static final long serialVersionUID = -6265024000462809436L;
+ // Reducing state
+ ReducingStateDescriptor<Tuple2<Integer, Long>> reducingState = new ReducingStateDescriptor<>(
+ "any-name",
+ new SumReduce(),
+ source.getType());
- @Override
- public Integer getKey(Tuple2<Integer, Long> value) throws Exception {
- return value.f0;
- }
- }).asQueryableState(queryName);
+ final String queryName = "duplicate-me";
- // Submit the job graph
- JobGraph jobGraph = env.getStreamGraph().getJobGraph();
- jobId = jobGraph.getJobID();
+ final QueryableStateStream<Integer, Tuple2<Integer, Long>> queryableState =
+ source.keyBy(new KeySelector<Tuple2<Integer, Long>, Integer>() {
+ private static final long serialVersionUID = -4126824763829132959L;
- final CompletableFuture<TestingJobManagerMessages.JobStatusIs> failedFuture =
- notifyWhenJobStatusIs(jobId, JobStatus.FAILED, deadline);
+ @Override
+ public Integer getKey(Tuple2<Integer, Long> value) {
+ return value.f0;
+ }
+ }).asQueryableState(queryName, reducingState);
- cluster.submitJobDetached(jobGraph);
+ final QueryableStateStream<Integer, Tuple2<Integer, Long>> duplicate =
+ source.keyBy(new KeySelector<Tuple2<Integer, Long>, Integer>() {
+ private static final long serialVersionUID = -6265024000462809436L;
- TestingJobManagerMessages.JobStatusIs jobStatus =
- failedFuture.get(deadline.timeLeft().toMillis(), TimeUnit.MILLISECONDS);
- assertEquals(JobStatus.FAILED, jobStatus.state());
+ @Override
+ public Integer getKey(Tuple2<Integer, Long> value) {
+ return value.f0;
+ }
+ }).asQueryableState(queryName);
+
+ // Submit the job graph
+ final JobGraph jobGraph = env.getStreamGraph().getJobGraph();
+ final JobID jobId = jobGraph.getJobID();
+
+ final CompletableFuture<TestingJobManagerMessages.JobStatusIs> failedFuture =
+ notifyWhenJobStatusIs(jobId, JobStatus.FAILED, deadline);
- // Get the job and check the cause
- JobManagerMessages.JobFound jobFound = FutureUtils.toJava(
- cluster.getLeaderGateway(deadline.timeLeft())
- .ask(new JobManagerMessages.RequestJob(jobId), deadline.timeLeft())
- .mapTo(ClassTag$.MODULE$.<JobManagerMessages.JobFound>apply(JobManagerMessages.JobFound.class)))
- .get(deadline.timeLeft().toMillis(), TimeUnit.MILLISECONDS);
+ final CompletableFuture<TestingJobManagerMessages.JobStatusIs> cancellationFuture =
+ notifyWhenJobStatusIs(jobId, JobStatus.CANCELED, deadline);
- String failureCause = jobFound.executionGraph().getFailureCause().getExceptionAsString();
+ cluster.submitJobDetached(jobGraph);
- assertTrue("Not instance of SuppressRestartsException", failureCause.startsWith("org.apache.flink.runtime.execution.SuppressRestartsException"));
- int causedByIndex = failureCause.indexOf("Caused by: ");
- String subFailureCause = failureCause.substring(causedByIndex + "Caused by: ".length());
- assertTrue("Not caused by IllegalStateException", subFailureCause.startsWith("java.lang.IllegalStateException"));
- assertTrue("Exception does not contain registration name", subFailureCause.contains(queryName));
+ try {
+ final TestingJobManagerMessages.JobStatusIs jobStatus =
+ failedFuture.get(deadline.timeLeft().toMillis(), TimeUnit.MILLISECONDS);
+ assertEquals(JobStatus.FAILED, jobStatus.state());
--- End diff --
isn't this always true if the future did not time out? (In which case get() throws a TimeoutException)
---
[GitHub] flink pull request #5062: [FLINK-7880][QS] Wait for proper resource cleanup ...
Posted by zentol <gi...@git.apache.org>.
Github user zentol commented on a diff in the pull request:
https://github.com/apache/flink/pull/5062#discussion_r153794159
--- Diff: flink-queryable-state/flink-queryable-state-runtime/src/main/java/org/apache/flink/queryablestate/server/KvStateServerImpl.java ---
@@ -101,6 +103,12 @@ public InetSocketAddress getServerAddress() {
@Override
public void shutdown() {
- super.shutdown();
+ try {
+ Time timeout = Time.seconds(10L);
+ shutdownServer(timeout).get(timeout.toMilliseconds(), TimeUnit.MILLISECONDS);
--- End diff --
why not just `get(10, TimeUnit.SECONDS)`?
---
[GitHub] flink issue #5062: [FLINK-7880][QS] Wait for proper resource cleanup after e...
Posted by kl0u <gi...@git.apache.org>.
Github user kl0u commented on the issue:
https://github.com/apache/flink/pull/5062
This is a follow-up to https://github.com/apache/flink/pull/5062
---