You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by zentol <gi...@git.apache.org> on 2017/09/05 11:01:57 UTC
[GitHub] flink pull request #4645: [FLINK-7582] [REST] Netty thread immediately parse...
GitHub user zentol opened a pull request:
https://github.com/apache/flink/pull/4645
[FLINK-7582] [REST] Netty thread immediately parses responses
## What is the purpose of the change
This PR modifies the RestClient to make the netty receiver threads parse responses (and complete the associated future) immediately after receiving a response, instead of deferring the parsing to a separate executor.
This guarantees that we don't buffer responses that haven't been parsed yet. This was previously the case if the executor could not keep up with the response rate, for example because it was used exclusively for sending requests.
## Verifying this change
This change is a trivial rework / code cleanup without any test coverage.
You can merge this pull request into a Git repository by running:
$ git pull https://github.com/zentol/flink 7582
Alternatively you can review and apply these changes as the patch at:
https://github.com/apache/flink/pull/4645.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 #4645
----
commit bf910ff69fa2754c235f8f0d957cadefbc37c53f
Author: zentol <ch...@apache.org>
Date: 2017-09-05T10:58:57Z
[FLINK-7582] [REST] Netty thread immediately parses responses
----
---
[GitHub] flink pull request #4645: [FLINK-7582] [REST] Netty thread immediately parse...
Posted by kl0u <gi...@git.apache.org>.
Github user kl0u commented on a diff in the pull request:
https://github.com/apache/flink/pull/4645#discussion_r137210136
--- Diff: flink-runtime/src/main/java/org/apache/flink/runtime/rest/RestClient.java ---
@@ -159,24 +155,17 @@ public void shutdown(Time timeout) {
}
private <P extends ResponseBody> CompletableFuture<P> submitRequest(String targetAddress, int targetPort, FullHttpRequest httpRequest, Class<P> responseClass) {
- return CompletableFuture.supplyAsync(() -> bootstrap.connect(targetAddress, targetPort), executor)
- .thenApply((channel) -> {
- try {
- return channel.sync();
- } catch (InterruptedException e) {
- throw new FlinkRuntimeException(e);
- }
- })
- .thenApply((ChannelFuture::channel))
- .thenCompose(channel -> {
- ClientHandler handler = channel.pipeline().get(ClientHandler.class);
- CompletableFuture<JsonResponse> future = handler.getJsonFuture();
- channel.writeAndFlush(httpRequest);
- return future;
- }).thenComposeAsync(
- (JsonResponse rawResponse) -> parseResponse(rawResponse, responseClass),
- executor
- );
+ ChannelFuture connect = bootstrap.connect(targetAddress, targetPort);
+ Channel channel;
+ try {
+ channel = connect.sync().channel();
+ } catch (InterruptedException e) {
+ throw new FlinkRuntimeException(e);
--- End diff --
This should be a `RestClientException`.
---
[GitHub] flink pull request #4645: [FLINK-7582] [REST] Netty thread immediately parse...
Posted by uce <gi...@git.apache.org>.
Github user uce commented on a diff in the pull request:
https://github.com/apache/flink/pull/4645#discussion_r137269070
--- Diff: flink-runtime/src/main/java/org/apache/flink/runtime/rest/RestClient.java ---
@@ -159,24 +155,17 @@ public void shutdown(Time timeout) {
}
private <P extends ResponseBody> CompletableFuture<P> submitRequest(String targetAddress, int targetPort, FullHttpRequest httpRequest, Class<P> responseClass) {
- return CompletableFuture.supplyAsync(() -> bootstrap.connect(targetAddress, targetPort), executor)
- .thenApply((channel) -> {
- try {
- return channel.sync();
- } catch (InterruptedException e) {
- throw new FlinkRuntimeException(e);
- }
- })
- .thenApply((ChannelFuture::channel))
- .thenCompose(channel -> {
- ClientHandler handler = channel.pipeline().get(ClientHandler.class);
- CompletableFuture<JsonResponse> future = handler.getJsonFuture();
- channel.writeAndFlush(httpRequest);
- return future;
- }).thenComposeAsync(
- (JsonResponse rawResponse) -> parseResponse(rawResponse, responseClass),
- executor
- );
+ ChannelFuture connect = bootstrap.connect(targetAddress, targetPort);
+ Channel channel;
--- End diff --
I understand, but it can be confusing that a method which returns a future actually has a (potentially long) blocking operation.
---
[GitHub] flink pull request #4645: [FLINK-7582] [REST] Netty thread immediately parse...
Posted by zentol <gi...@git.apache.org>.
Github user zentol commented on a diff in the pull request:
https://github.com/apache/flink/pull/4645#discussion_r139403702
--- Diff: flink-runtime/src/main/java/org/apache/flink/runtime/rest/RestClient.java ---
@@ -159,24 +155,17 @@ public void shutdown(Time timeout) {
}
private <P extends ResponseBody> CompletableFuture<P> submitRequest(String targetAddress, int targetPort, FullHttpRequest httpRequest, Class<P> responseClass) {
- return CompletableFuture.supplyAsync(() -> bootstrap.connect(targetAddress, targetPort), executor)
- .thenApply((channel) -> {
- try {
- return channel.sync();
- } catch (InterruptedException e) {
- throw new FlinkRuntimeException(e);
- }
- })
- .thenApply((ChannelFuture::channel))
- .thenCompose(channel -> {
- ClientHandler handler = channel.pipeline().get(ClientHandler.class);
- CompletableFuture<JsonResponse> future = handler.getJsonFuture();
- channel.writeAndFlush(httpRequest);
- return future;
- }).thenComposeAsync(
- (JsonResponse rawResponse) -> parseResponse(rawResponse, responseClass),
- executor
- );
+ ChannelFuture connect = bootstrap.connect(targetAddress, targetPort);
+ Channel channel;
--- End diff --
That's true, generally I would also prefer making the entire method async. For now I would still like to make this change though, and happily revisit this later on when we have some more robust mechanism in place to deal with heavy loads.
---
[GitHub] flink pull request #4645: [FLINK-7582] [REST] Netty thread immediately parse...
Posted by zentol <gi...@git.apache.org>.
Github user zentol commented on a diff in the pull request:
https://github.com/apache/flink/pull/4645#discussion_r137065666
--- Diff: flink-runtime/src/main/java/org/apache/flink/runtime/rest/RestClient.java ---
@@ -159,24 +155,17 @@ public void shutdown(Time timeout) {
}
private <P extends ResponseBody> CompletableFuture<P> submitRequest(String targetAddress, int targetPort, FullHttpRequest httpRequest, Class<P> responseClass) {
- return CompletableFuture.supplyAsync(() -> bootstrap.connect(targetAddress, targetPort), executor)
- .thenApply((channel) -> {
- try {
- return channel.sync();
- } catch (InterruptedException e) {
- throw new FlinkRuntimeException(e);
- }
- })
- .thenApply((ChannelFuture::channel))
- .thenCompose(channel -> {
- ClientHandler handler = channel.pipeline().get(ClientHandler.class);
- CompletableFuture<JsonResponse> future = handler.getJsonFuture();
- channel.writeAndFlush(httpRequest);
- return future;
- }).thenComposeAsync(
- (JsonResponse rawResponse) -> parseResponse(rawResponse, responseClass),
- executor
- );
+ ChannelFuture connect = bootstrap.connect(targetAddress, targetPort);
+ Channel channel;
--- End diff --
Technically yes, this alone would fix the problem at hand.
The sending part was modified as a band-aid to alleviate issues that can occur under heavy load since we don't have any throttling mechanism for accepting requests. For example, in one of our tests we had thousands of futures suddenly fail because all ports were in use, not to mention that you can allocate a virtually infinite number of futures (way faster than we could ever process them).
I couldn't quickly come up with a good scheme to throttle requests though. Suggestions are welcome ;)
---
[GitHub] flink pull request #4645: [FLINK-7582] [REST] Netty thread immediately parse...
Posted by kl0u <gi...@git.apache.org>.
Github user kl0u commented on a diff in the pull request:
https://github.com/apache/flink/pull/4645#discussion_r137209983
--- Diff: flink-runtime/src/main/java/org/apache/flink/runtime/rest/RestClient.java ---
@@ -77,14 +77,10 @@
private static final ObjectMapper objectMapper = RestMapperUtils.getStrictObjectMapper();
- // used to open connections to a rest server endpoint
- private final Executor executor;
-
private Bootstrap bootstrap;
- public RestClient(RestClientConfiguration configuration, Executor executor) {
+ public RestClient(RestClientConfiguration configuration) {
--- End diff --
Given that now we let netty do all the parsing, I think it makes sense to allow the user to specify the size of the `NioEventLoopGroup ` (currently set to 1 - which can be the default).
---
[GitHub] flink pull request #4645: [FLINK-7582] [REST] Netty thread immediately parse...
Posted by uce <gi...@git.apache.org>.
Github user uce commented on a diff in the pull request:
https://github.com/apache/flink/pull/4645#discussion_r137038459
--- Diff: flink-runtime/src/main/java/org/apache/flink/runtime/rest/RestClient.java ---
@@ -159,24 +155,17 @@ public void shutdown(Time timeout) {
}
private <P extends ResponseBody> CompletableFuture<P> submitRequest(String targetAddress, int targetPort, FullHttpRequest httpRequest, Class<P> responseClass) {
- return CompletableFuture.supplyAsync(() -> bootstrap.connect(targetAddress, targetPort), executor)
- .thenApply((channel) -> {
- try {
- return channel.sync();
- } catch (InterruptedException e) {
- throw new FlinkRuntimeException(e);
- }
- })
- .thenApply((ChannelFuture::channel))
- .thenCompose(channel -> {
- ClientHandler handler = channel.pipeline().get(ClientHandler.class);
- CompletableFuture<JsonResponse> future = handler.getJsonFuture();
- channel.writeAndFlush(httpRequest);
- return future;
- }).thenComposeAsync(
- (JsonResponse rawResponse) -> parseResponse(rawResponse, responseClass),
- executor
- );
+ ChannelFuture connect = bootstrap.connect(targetAddress, targetPort);
+ Channel channel;
--- End diff --
Couldn't you keep the async connect and only change the last `composeAsync` to `compose`?
---