You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by kk...@apache.org on 2018/01/16 10:29:58 UTC
[1/2] flink git commit: [FLINK-8050] [rest] REST server reports netty
exceptions on shutdown.
Repository: flink
Updated Branches:
refs/heads/master 907361d86 -> 7da32d19f
[FLINK-8050] [rest] REST server reports netty exceptions on shutdown.
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/642e11a9
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/642e11a9
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/642e11a9
Branch: refs/heads/master
Commit: 642e11a9cd31c83fbbabe860871c714f3d4ca981
Parents: 907361d
Author: kkloudas <kk...@gmail.com>
Authored: Wed Nov 22 18:16:04 2017 +0100
Committer: kkloudas <kk...@gmail.com>
Committed: Tue Jan 16 11:27:37 2018 +0100
----------------------------------------------------------------------
.../flink/runtime/rest/RestServerEndpoint.java | 28 ++++++++++++++++----
1 file changed, 23 insertions(+), 5 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/642e11a9/flink-runtime/src/main/java/org/apache/flink/runtime/rest/RestServerEndpoint.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/rest/RestServerEndpoint.java b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/RestServerEndpoint.java
index 38f7e25..c2acb8b 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/rest/RestServerEndpoint.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/RestServerEndpoint.java
@@ -233,7 +233,13 @@ public abstract class RestServerEndpoint {
CompletableFuture<?> channelFuture = new CompletableFuture<>();
if (this.serverChannel != null) {
- this.serverChannel.close().addListener(ignored -> channelFuture.complete(null));
+ this.serverChannel.close().addListener(finished -> {
+ if (finished.isSuccess()) {
+ channelFuture.complete(null);
+ } else {
+ channelFuture.completeExceptionally(finished.cause());
+ }
+ });
serverChannel = null;
}
CompletableFuture<?> groupFuture = new CompletableFuture<>();
@@ -242,12 +248,24 @@ public abstract class RestServerEndpoint {
channelFuture.thenRun(() -> {
if (bootstrap != null) {
if (bootstrap.group() != null) {
- bootstrap.group().shutdownGracefully(0, timeout.toMilliseconds(), TimeUnit.MILLISECONDS)
- .addListener(ignored -> groupFuture.complete(null));
+ bootstrap.group().shutdownGracefully(0L, timeout.toMilliseconds(), TimeUnit.MILLISECONDS)
+ .addListener(finished -> {
+ if (finished.isSuccess()) {
+ groupFuture.complete(null);
+ } else {
+ groupFuture.completeExceptionally(finished.cause());
+ }
+ });
}
if (bootstrap.childGroup() != null) {
- bootstrap.childGroup().shutdownGracefully(0, timeout.toMilliseconds(), TimeUnit.MILLISECONDS)
- .addListener(ignored -> childGroupFuture.complete(null));
+ bootstrap.childGroup().shutdownGracefully(0L, timeout.toMilliseconds(), TimeUnit.MILLISECONDS)
+ .addListener(finished -> {
+ if (finished.isSuccess()) {
+ childGroupFuture.complete(null);
+ } else {
+ childGroupFuture.completeExceptionally(finished.cause());
+ }
+ });
}
bootstrap = null;
} else {
[2/2] flink git commit: [FLINK-8049] [rest] REST client reports netty
exceptions on shutdown.
Posted by kk...@apache.org.
[FLINK-8049] [rest] REST client reports netty exceptions on shutdown.
This closes #5057.
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/7da32d19
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/7da32d19
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/7da32d19
Branch: refs/heads/master
Commit: 7da32d19f9623ca98c8a4ba76e7c406bf9318d4d
Parents: 642e11a
Author: kkloudas <kk...@gmail.com>
Authored: Wed Nov 22 18:25:06 2017 +0100
Committer: kkloudas <kk...@gmail.com>
Committed: Tue Jan 16 11:27:38 2018 +0100
----------------------------------------------------------------------
.../java/org/apache/flink/runtime/rest/RestClient.java | 10 ++++++++--
1 file changed, 8 insertions(+), 2 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/7da32d19/flink-runtime/src/main/java/org/apache/flink/runtime/rest/RestClient.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/rest/RestClient.java b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/RestClient.java
index cd80083..71891de 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/rest/RestClient.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/RestClient.java
@@ -122,8 +122,14 @@ public class RestClient {
CompletableFuture<?> groupFuture = new CompletableFuture<>();
if (bootstrap != null) {
if (bootstrap.group() != null) {
- bootstrap.group().shutdownGracefully(0, timeout.toMilliseconds(), TimeUnit.MILLISECONDS)
- .addListener(ignored -> groupFuture.complete(null));
+ bootstrap.group().shutdownGracefully(0L, timeout.toMilliseconds(), TimeUnit.MILLISECONDS)
+ .addListener(finished -> {
+ if (finished.isSuccess()) {
+ groupFuture.complete(null);
+ } else {
+ groupFuture.completeExceptionally(finished.cause());
+ }
+ });
}
}