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/20 15:15:31 UTC
[GitHub] flink pull request #5038: [FLINK-7880][FLINK-7975][FLINK-7974][QS] QS test i...
GitHub user kl0u opened a pull request:
https://github.com/apache/flink/pull/5038
[FLINK-7880][FLINK-7975][FLINK-7974][QS] QS test instability fix.
This is a follow-up on https://github.com/apache/flink/pull/4993.
It contains one additional commit that makes the QS ITcases to wait for proper clean-up before exiting.
R: @aljoscha @zentol
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/5038.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 #5038
----
commit 15c39807f560f6ee4f34e4ad1ba245b66dd10b09
Author: kkloudas <kk...@gmail.com>
Date: 2017-11-09T18:21:43Z
[FLINK-7975][QS] Wait for QS client to shutdown.
commit 689502e70cced18a9cb5b2b6435ff500b7bf70dd
Author: kkloudas <kk...@gmail.com>
Date: 2017-11-09T18:30:29Z
[FLINK-7974][QS] Wait for QS abstract server to shutdown.
commit 5c0e3fc82a88768caa750367fbf02e24848b3b53
Author: kkloudas <kk...@gmail.com>
Date: 2017-11-20T13:28:24Z
[FLINK-7880][QS] Wait for proper resource cleanup after each ITCase.
----
---
[GitHub] flink pull request #5038: [FLINK-7880][FLINK-7975][FLINK-7974][QS] QS test i...
Posted by zentol <gi...@git.apache.org>.
Github user zentol commented on a diff in the pull request:
https://github.com/apache/flink/pull/5038#discussion_r152299398
--- 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 --
you could just write `private NioEventLoopGroup nioGroup = new NioEventLoopGroup();` and remove the `@Before` method
---
[GitHub] flink pull request #5038: [FLINK-7880][FLINK-7975][FLINK-7974][QS] QS test i...
Posted by zentol <gi...@git.apache.org>.
Github user zentol commented on a diff in the pull request:
https://github.com/apache/flink/pull/5038#discussion_r152294984
--- 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<?> shutdown() {
--- End diff --
should be typed to `Void`.
---
[GitHub] flink pull request #5038: [FLINK-7880][FLINK-7975][FLINK-7974][QS] QS test i...
Posted by kl0u <gi...@git.apache.org>.
Github user kl0u closed the pull request at:
https://github.com/apache/flink/pull/5038
---
[GitHub] flink pull request #5038: [FLINK-7880][FLINK-7975][FLINK-7974][QS] QS test i...
Posted by zentol <gi...@git.apache.org>.
Github user zentol commented on a diff in the pull request:
https://github.com/apache/flink/pull/5038#discussion_r152300724
--- 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<?> shutdown() {
+ final CompletableFuture<?> newShutdownFuture = new CompletableFuture<>();
+ if (clientShutdownFuture.compareAndSet(null, newShutdownFuture)) {
+
+ final List<CompletableFuture<?>> 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 --
where in close() do we set the shutdown future to null? I only see that being done in sendRequest. (which seems fishy)
---
[GitHub] flink pull request #5038: [FLINK-7880][FLINK-7975][FLINK-7974][QS] QS test i...
Posted by zentol <gi...@git.apache.org>.
Github user zentol commented on a diff in the pull request:
https://github.com/apache/flink/pull/5038#discussion_r152298240
--- Diff: flink-queryable-state/flink-queryable-state-client-java/src/main/java/org/apache/flink/queryablestate/network/Client.java ---
@@ -312,32 +345,43 @@ private void handInChannel(Channel channel) {
/**
* Close the connecting channel with a ClosedChannelException.
*/
- private void close() {
- close(new ClosedChannelException());
+ private CompletableFuture<?> 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) {
- if (failureCause == null) {
- failureCause = cause;
- }
+ private CompletableFuture<?> close(Throwable cause) {
+ CompletableFuture<?> future = new CompletableFuture<>();
+ if (connectionShutdownFuture.compareAndSet(null, future)) {
+ synchronized (connectLock) {
+ if (!closed) {
--- End diff --
this seems unnecessary, doesn't the check at L358 guarantee that the entire branch is only executed once?
---
[GitHub] flink pull request #5038: [FLINK-7880][FLINK-7975][FLINK-7974][QS] QS test i...
Posted by zentol <gi...@git.apache.org>.
Github user zentol commented on a diff in the pull request:
https://github.com/apache/flink/pull/5038#discussion_r152297601
--- Diff: flink-queryable-state/flink-queryable-state-client-java/src/main/java/org/apache/flink/queryablestate/network/Client.java ---
@@ -312,32 +345,43 @@ private void handInChannel(Channel channel) {
/**
* Close the connecting channel with a ClosedChannelException.
*/
- private void close() {
- close(new ClosedChannelException());
+ private CompletableFuture<?> 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) {
- if (failureCause == null) {
- failureCause = cause;
- }
+ private CompletableFuture<?> close(Throwable cause) {
--- End diff --
same as above
---
[GitHub] flink pull request #5038: [FLINK-7880][FLINK-7975][FLINK-7974][QS] QS test i...
Posted by zentol <gi...@git.apache.org>.
Github user zentol commented on a diff in the pull request:
https://github.com/apache/flink/pull/5038#discussion_r152298612
--- Diff: flink-queryable-state/flink-queryable-state-client-java/src/main/java/org/apache/flink/queryablestate/network/Client.java ---
@@ -386,6 +430,9 @@ private PendingRequest(REQ request) {
/** Reference to a failure that was reported by the channel. */
private final AtomicReference<Throwable> failureCause = new AtomicReference<>();
+ /** Atomic shut down future. */
+ private final AtomicReference<CompletableFuture<Boolean>> connectionShutdownFuture = new AtomicReference<>(null);
--- End diff --
why does this one suddenly return a boolean?
---