You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by "patricklucas (via GitHub)" <gi...@apache.org> on 2023/07/12 13:19:15 UTC
[GitHub] [flink] patricklucas opened a new pull request, #22987: [FLINK-32583][rest] Fix deadlock in RestClient
patricklucas opened a new pull request, #22987:
URL: https://github.com/apache/flink/pull/22987
## What is the purpose of the change
Fix the issue described in FLINK-32583 wherein a request made to a RestClient that has already been shutdown can deadlock due to the returned future never resolving.
## Brief change log
- Add check if the Netty channel future has already failed after attempting to add a listener to it
## Verifying this change
This change added tests and can be verified as follows:
- Added test for this behavior in RestClientTest, `testCloseClientBeforeRequest`
- This test calls `restClient.close()` and then attempts to make a request that would eventually fail with a connection timeout
- The test blocks on the future returned by the call with a timeout of 0 so it can test that the expected exception (and not TimeoutException) is thrown
## Does this pull request potentially affect one of the following parts:
- Dependencies (does it add or upgrade a dependency): no
- The public API, i.e., is any changed class annotated with `@Public(Evolving)`: no
- The serializers: no
- The runtime per-record code paths (performance sensitive): no
- Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Kubernetes/Yarn, ZooKeeper: no
- The S3 file system connector: no
## Documentation
- Does this pull request introduce a new feature? no
- If yes, how is the feature documented? not applicable
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] XComp commented on a diff in pull request #22987: [FLINK-32583][rest] Fix deadlock in RestClient
Posted by "XComp (via GitHub)" <gi...@apache.org>.
XComp commented on code in PR #22987:
URL: https://github.com/apache/flink/pull/22987#discussion_r1272076683
##########
flink-runtime/src/test/java/org/apache/flink/runtime/rest/RestClientTest.java:
##########
@@ -207,6 +210,40 @@ public void testRestClientClosedHandling() throws Exception {
}
}
+ /**
+ * Tests that the futures returned by {@link RestClient} fail immediately if the client is
+ * already closed.
+ *
+ * <p>See FLINK-32583
+ */
+ @Test
+ public void testCloseClientBeforeRequest() throws Exception {
Review Comment:
I looked into the multi-threading of netty a bit more and found the `SelectStrategy` being a viable tool to control the state of the request. I came up with the following test:
```
@Test
public void testCloseClientWhileProcessingRequest() throws Exception {
final OneShotLatch connectTriggered = new OneShotLatch();
final OneShotLatch closeTriggered = new OneShotLatch();
final SelectStrategy fallbackSelectStrategy =
DefaultSelectStrategyFactory.INSTANCE.newSelectStrategy();
final SelectStrategy selectStrategy =
(selectSupplier, hasTasks) -> {
connectTriggered.trigger();
closeTriggered.awaitQuietly();
return fallbackSelectStrategy.calculateStrategy(selectSupplier, hasTasks);
};
// Note that the executor passed to the RestClient constructor is not the same as the
// executor used by Netty
try (final RestClient restClient =
new RestClient(
new Configuration(), Executors.directExecutor(), () -> selectStrategy)) {
final CompletableFuture<?> requestFuture =
restClient.sendRequest(
unroutableIp,
80,
new TestMessageHeaders(),
EmptyMessageParameters.getInstance(),
EmptyRequestBody.getInstance());
connectTriggered.await();
final CompletableFuture<Void> closeFuture = restClient.closeAsync();
closeTriggered.trigger();
// close should complete successfully
closeFuture.get();
final Throwable cause =
assertThrows(
ExecutionException.class,
() -> requestFuture.get(0, TimeUnit.SECONDS))
.getCause();
assertThat(cause, instanceOf(IllegalStateException.class));
}
}
```
It would require an extension of the `RestClient` adding a package-private constructor:
```
// ...
public RestClient(Configuration configuration, Executor executor)
throws ConfigurationException {
this(configuration, executor, DefaultSelectStrategyFactory.INSTANCE);
}
@VisibleForTesting
RestClient(
Configuration configuration,
Executor executor,
SelectStrategyFactory selectStrategyFactory)
//...
NioEventLoopGroup group =
new NioEventLoopGroup(
1,
new ExecutorThreadFactory("flink-rest-client-netty"),
SelectorProvider.provider(),
selectStrategyFactory);
// ...
```
WDYT? That should test the `isRunning.get()` code path.
##########
flink-runtime/src/test/java/org/apache/flink/runtime/rest/RestClientTest.java:
##########
@@ -207,6 +210,40 @@ public void testRestClientClosedHandling() throws Exception {
}
}
+ /**
+ * Tests that the futures returned by {@link RestClient} fail immediately if the client is
+ * already closed.
+ *
+ * <p>See FLINK-32583
+ */
+ @Test
+ public void testCloseClientBeforeRequest() throws Exception {
Review Comment:
I looked into the multi-threading of netty a bit more and found the `SelectStrategy` being a viable tool to control the state of the request. I came up with the following test:
```
@Test
public void testCloseClientWhileProcessingRequest() throws Exception {
final OneShotLatch connectTriggered = new OneShotLatch();
final OneShotLatch closeTriggered = new OneShotLatch();
final SelectStrategy fallbackSelectStrategy =
DefaultSelectStrategyFactory.INSTANCE.newSelectStrategy();
final SelectStrategy selectStrategy =
(selectSupplier, hasTasks) -> {
connectTriggered.trigger();
closeTriggered.awaitQuietly();
return fallbackSelectStrategy.calculateStrategy(selectSupplier, hasTasks);
};
// Note that the executor passed to the RestClient constructor is not the same as the
// executor used by Netty
try (final RestClient restClient =
new RestClient(
new Configuration(), Executors.directExecutor(), () -> selectStrategy)) {
final CompletableFuture<?> requestFuture =
restClient.sendRequest(
unroutableIp,
80,
new TestMessageHeaders(),
EmptyMessageParameters.getInstance(),
EmptyRequestBody.getInstance());
connectTriggered.await();
final CompletableFuture<Void> closeFuture = restClient.closeAsync();
closeTriggered.trigger();
// close should complete successfully
closeFuture.get();
final Throwable cause =
assertThrows(
ExecutionException.class,
() -> requestFuture.get(0, TimeUnit.SECONDS))
.getCause();
assertThat(cause, instanceOf(IllegalStateException.class));
}
}
```
It would require an extension of the `RestClient` adding a package-private constructor:
```
// ...
public RestClient(Configuration configuration, Executor executor)
throws ConfigurationException {
this(configuration, executor, DefaultSelectStrategyFactory.INSTANCE);
}
@VisibleForTesting
RestClient(
Configuration configuration,
Executor executor,
SelectStrategyFactory selectStrategyFactory)
//...
NioEventLoopGroup group =
new NioEventLoopGroup(
1,
new ExecutorThreadFactory("flink-rest-client-netty"),
SelectorProvider.provider(),
selectStrategyFactory);
// ...
```
WDYT? That should test the `isRunning.get()` code path.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] patricklucas commented on a diff in pull request #22987: [FLINK-32583][rest] Fix deadlock in RestClient
Posted by "patricklucas (via GitHub)" <gi...@apache.org>.
patricklucas commented on code in PR #22987:
URL: https://github.com/apache/flink/pull/22987#discussion_r1263446778
##########
flink-runtime/src/main/java/org/apache/flink/runtime/rest/RestClient.java:
##########
@@ -501,6 +501,22 @@ private <P extends ResponseBody> CompletableFuture<P> submitRequest(
}
});
+ // [FLINK-32583] If connectFuture failed instantly but channelFuture is unresolved, it may
+ // mean the executor service Netty is using has shut down, in which case the above listener
+ // to complete channelFuture will never run
+ if (connectFuture.isDone() && !connectFuture.isSuccess() && !channelFuture.isDone()) {
+ final String message;
+ if (!isRunning.get()) {
Review Comment:
Can do. Since the client being closed is the likeliest reason we hit this, I preferred to put that one first. "if not running"
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] patricklucas commented on pull request #22987: [FLINK-32583][rest] Fix deadlock in RestClient
Posted by "patricklucas (via GitHub)" <gi...@apache.org>.
patricklucas commented on PR #22987:
URL: https://github.com/apache/flink/pull/22987#issuecomment-1652189416
Great, I'll get to this tomorrow, thanks for the reviews @XComp.
Regarding the commit organization, I'm happy to break out another hotfix commit—wasn't sure if the branch would get squashed before going into master. Can you confirm that fast-forwarding is the merge strategy? If that's the case, I assume you want me to also fixup the "Changes per review" commits into the primary commit?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] XComp merged pull request #22987: [FLINK-32583][rest] Fix deadlock in RestClient
Posted by "XComp (via GitHub)" <gi...@apache.org>.
XComp merged PR #22987:
URL: https://github.com/apache/flink/pull/22987
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] XComp commented on pull request #22987: [FLINK-32583][rest] Fix deadlock in RestClient
Posted by "XComp (via GitHub)" <gi...@apache.org>.
XComp commented on PR #22987:
URL: https://github.com/apache/flink/pull/22987#issuecomment-1653099478
It's going to be rebased and merged through Github (so essentially fast-forward merge, yeah). The commits could be something like:
* [hotfix][test] Improve RestClientTest test case
* [hotfix][test] Make package-private fields private and accessible through getters instead in RestClient and RestServerEndpoint
* [FLINK-32583] ...
About the CI failure is related to FLINK-30719.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] XComp commented on a diff in pull request #22987: [FLINK-32583][rest] Fix deadlock in RestClient
Posted by "XComp (via GitHub)" <gi...@apache.org>.
XComp commented on code in PR #22987:
URL: https://github.com/apache/flink/pull/22987#discussion_r1262773979
##########
flink-runtime/src/main/java/org/apache/flink/runtime/rest/RestClient.java:
##########
@@ -501,6 +501,22 @@ private <P extends ResponseBody> CompletableFuture<P> submitRequest(
}
});
+ // [FLINK-32583] If connectFuture failed instantly but channelFuture is unresolved, it may
+ // mean the executor service Netty is using has shut down, in which case the above listener
+ // to complete channelFuture will never run
+ if (connectFuture.isDone() && !connectFuture.isSuccess() && !channelFuture.isDone()) {
Review Comment:
Should we add a comment why we're not handling the success case? Essentially, we're trying to work around a bug in the netty code, aren't we?
```
/**
* Adds the specified listener to this future. The
* specified listener is notified when this future is
* {@linkplain #isDone() done}. If this future is already
* completed, the specified listener is notified immediately.
*/
```
The JavaDoc of `Future.addListener` states that the listener would be informed if the future is already completed (which includes the successful and the exceptional case). But that doesn't match the implementation, apparently.
Right now, we're missing the `isCancelled` and `isSuccess` case handling. It's not clear whether these cases can actually happen. But since we're (at least) covering the success case in the listener implementation above, wouldn't it be reasonable to cover it here as well?
##########
flink-runtime/src/test/java/org/apache/flink/runtime/rest/RestClientTest.java:
##########
@@ -207,6 +209,42 @@ public void testRestClientClosedHandling() throws Exception {
}
}
+ /**
+ * Tests that the futures returned by {@link RestClient} fail immediately if the client is
+ * already closed.
+ *
+ * <p>See FLINK-32583
+ */
+ @Test
+ public void testCloseClientBeforeRequest() throws Exception {
+ // Note that the executor passed to the RestClient constructor is not the same as the
+ // executor used by Netty
+ try (final RestClient restClient =
+ new RestClient(new Configuration(), Executors.directExecutor())) {
+ // Intentionally close the client (and thus also the executor used by Netty)
+ restClient.close();
+
+ CompletableFuture<?> future =
+ restClient.sendRequest(
+ unroutableIp,
+ 80,
+ new TestMessageHeaders(),
+ EmptyMessageParameters.getInstance(),
+ EmptyRequestBody.getInstance());
+
+ // Call get() on the future with a timeout of 0 so we can test that the exception thrown
+ // is not a TimeoutException, which is what would be thrown if restClient were not
+ // already closed
+ final ThrowingRunnable getFuture = () -> future.get(0, TimeUnit.SECONDS);
+
+ final ExecutionException executionException =
+ assertThrows(ExecutionException.class, getFuture);
+ final Throwable throwable = ExceptionUtils.stripExecutionException(executionException);
+ assertThat(throwable, instanceOf(IOException.class));
+ assertThat(throwable.getMessage(), containsString("RestClient is closed"));
Review Comment:
```suggestion
final Throwable cause = assertThrows(ExecutionException.class, getFuture).getCause();
assertThat(cause, instanceOf(IOException.class));
assertThat(cause.getMessage(), containsString("RestClient is closed"));
```
nit: What about extracting the cause right away and doing the assertions on the cause? That makes the test more restrictive. We shouldn't expect a multiple-layer stacktrace here, should we?
##########
flink-runtime/src/main/java/org/apache/flink/runtime/rest/RestClient.java:
##########
@@ -501,6 +501,22 @@ private <P extends ResponseBody> CompletableFuture<P> submitRequest(
}
});
+ // [FLINK-32583] If connectFuture failed instantly but channelFuture is unresolved, it may
+ // mean the executor service Netty is using has shut down, in which case the above listener
+ // to complete channelFuture will never run
+ if (connectFuture.isDone() && !connectFuture.isSuccess() && !channelFuture.isDone()) {
+ final String message;
+ if (!isRunning.get()) {
Review Comment:
nit: Could we invert the if condition? ...just for readability purposes
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] XComp commented on a diff in pull request #22987: [FLINK-32583][rest] Fix deadlock in RestClient
Posted by "XComp (via GitHub)" <gi...@apache.org>.
XComp commented on code in PR #22987:
URL: https://github.com/apache/flink/pull/22987#discussion_r1274576162
##########
flink-runtime/src/test/java/org/apache/flink/runtime/rest/RestClientTest.java:
##########
@@ -207,6 +210,40 @@ public void testRestClientClosedHandling() throws Exception {
}
}
+ /**
+ * Tests that the futures returned by {@link RestClient} fail immediately if the client is
+ * already closed.
+ *
+ * <p>See FLINK-32583
+ */
+ @Test
+ public void testCloseClientBeforeRequest() throws Exception {
Review Comment:
that's a fair point - I know what you mean. :-D
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] XComp commented on pull request #22987: [FLINK-32583][rest] Fix deadlock in RestClient
Posted by "XComp (via GitHub)" <gi...@apache.org>.
XComp commented on PR #22987:
URL: https://github.com/apache/flink/pull/22987#issuecomment-1645543909
the [k8s build failure](https://dev.azure.com/apache-flink/apache-flink/_build/results?buildId=51496&view=logs&j=bea52777-eaf8-5663-8482-18fbc3630e81&s=ae4f8708-9994-57d3-c2d7-b892156e7812&t=43ba8ce7-ebbf-57cd-9163-444305d74117&l=6199) is happening consistently right now. I'm looking into FLINK-32632 and will get back to you on that PR afterwards
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] patricklucas commented on a diff in pull request #22987: [FLINK-32583][rest] Fix deadlock in RestClient
Posted by "patricklucas (via GitHub)" <gi...@apache.org>.
patricklucas commented on code in PR #22987:
URL: https://github.com/apache/flink/pull/22987#discussion_r1261177569
##########
flink-runtime/src/main/java/org/apache/flink/runtime/rest/RestClient.java:
##########
@@ -501,6 +501,14 @@ private <P extends ResponseBody> CompletableFuture<P> submitRequest(
}
});
+ // [FLINK-32583] If connectFuture failed instantly but channelFuture is unresolved, it may
+ // mean the executor service Netty is using has shut down, in which case the above listener
+ // to complete channelFuture will never run
+ if (connectFuture.isDone() && !connectFuture.isSuccess() && !channelFuture.isDone()) {
+ channelFuture.completeExceptionally(
+ new IOException("Netty connection failed instantly. Shutting down?"));
+ }
Review Comment:
Realistically the `channelFuture.isDone()` check will always pass, as it's unlikely the listener would ever have had time to run—I can remove it if a reviewer wishes.
Also, I'm open to suggestions for a different exception to use than IOException.
##########
flink-runtime/src/main/java/org/apache/flink/runtime/rest/RestClient.java:
##########
@@ -501,6 +501,14 @@ private <P extends ResponseBody> CompletableFuture<P> submitRequest(
}
});
+ // [FLINK-32583] If connectFuture failed instantly but channelFuture is unresolved, it may
+ // mean the executor service Netty is using has shut down, in which case the above listener
+ // to complete channelFuture will never run
+ if (connectFuture.isDone() && !connectFuture.isSuccess() && !channelFuture.isDone()) {
+ channelFuture.completeExceptionally(
+ new IOException("Netty connection failed instantly. Shutting down?"));
+ }
Review Comment:
Realistically the `channelFuture.isDone()` check will always pass, as it's unlikely the listener would ever have had time to run—I can remove it if a reviewer wishes.
Also, I'm open to suggestions for a different exception to use than IOException.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] patricklucas commented on pull request #22987: [FLINK-32583][rest] Fix deadlock in RestClient
Posted by "patricklucas (via GitHub)" <gi...@apache.org>.
patricklucas commented on PR #22987:
URL: https://github.com/apache/flink/pull/22987#issuecomment-1645488393
@flinkbot run azure
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] patricklucas commented on pull request #22987: [FLINK-32583][rest] Fix deadlock in RestClient
Posted by "patricklucas (via GitHub)" <gi...@apache.org>.
patricklucas commented on PR #22987:
URL: https://github.com/apache/flink/pull/22987#issuecomment-1651315710
@XComp Made the following changes:
- Updated `testCloseClientWhileProcessingRequest` to check that `responseChannelFutures` is empty before and size 1 after the call to `sendRequest`
- Added a test that preloads a response channel future and makes sure it's resolved as expected
- To accomplish that, added `@VisibleForTesting RestClient#getResponseChannelFutures()`
- To keep consistency, I changed one other field which was `@VisibleForTesting`, but non-final and without a getter, hope this is okay
- Making the field final should be a strict improvement, but being consistent about adding a getter rather than exposing a bare field also seems good
- I also made a symmetric change in `RestServerEndpoint`—the diff should be clear as to why that made sense
It occurred to me that there is actually no test in `RestClientTest` that actually tests the happy path. If there were, I would have copied it to test that the response channel future is added and also cleared when the client is not closed, but the request completes successfully. I'm not super motivated to do that right now as I'm very busy and have already put a good bit of time into this change—what do you think?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot commented on pull request #22987: [FLINK-32583][rest] Fix deadlock in RestClient
Posted by "flinkbot (via GitHub)" <gi...@apache.org>.
flinkbot commented on PR #22987:
URL: https://github.com/apache/flink/pull/22987#issuecomment-1632531200
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "4a78684a6d5d9655a92550ff8f5132f9047a7ddf",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "4a78684a6d5d9655a92550ff8f5132f9047a7ddf",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 4a78684a6d5d9655a92550ff8f5132f9047a7ddf UNKNOWN
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run azure` re-run the last Azure build
</details>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] patricklucas commented on a diff in pull request #22987: [FLINK-32583][rest] Fix deadlock in RestClient
Posted by "patricklucas (via GitHub)" <gi...@apache.org>.
patricklucas commented on code in PR #22987:
URL: https://github.com/apache/flink/pull/22987#discussion_r1263443555
##########
flink-runtime/src/test/java/org/apache/flink/runtime/rest/RestClientTest.java:
##########
@@ -207,6 +209,42 @@ public void testRestClientClosedHandling() throws Exception {
}
}
+ /**
+ * Tests that the futures returned by {@link RestClient} fail immediately if the client is
+ * already closed.
+ *
+ * <p>See FLINK-32583
+ */
+ @Test
+ public void testCloseClientBeforeRequest() throws Exception {
+ // Note that the executor passed to the RestClient constructor is not the same as the
+ // executor used by Netty
+ try (final RestClient restClient =
+ new RestClient(new Configuration(), Executors.directExecutor())) {
+ // Intentionally close the client (and thus also the executor used by Netty)
+ restClient.close();
+
+ CompletableFuture<?> future =
+ restClient.sendRequest(
+ unroutableIp,
+ 80,
+ new TestMessageHeaders(),
+ EmptyMessageParameters.getInstance(),
+ EmptyRequestBody.getInstance());
+
+ // Call get() on the future with a timeout of 0 so we can test that the exception thrown
+ // is not a TimeoutException, which is what would be thrown if restClient were not
+ // already closed
+ final ThrowingRunnable getFuture = () -> future.get(0, TimeUnit.SECONDS);
+
+ final ExecutionException executionException =
+ assertThrows(ExecutionException.class, getFuture);
+ final Throwable throwable = ExceptionUtils.stripExecutionException(executionException);
+ assertThat(throwable, instanceOf(IOException.class));
+ assertThat(throwable.getMessage(), containsString("RestClient is closed"));
Review Comment:
Sounds good, though the same should be true of `testConnectionTimeout` in this same file which I followed the pattern of—do Flink contribution expectations say I should make the same change to that test or leave it alone?
##########
flink-runtime/src/test/java/org/apache/flink/runtime/rest/RestClientTest.java:
##########
@@ -207,6 +209,42 @@ public void testRestClientClosedHandling() throws Exception {
}
}
+ /**
+ * Tests that the futures returned by {@link RestClient} fail immediately if the client is
+ * already closed.
+ *
+ * <p>See FLINK-32583
+ */
+ @Test
+ public void testCloseClientBeforeRequest() throws Exception {
+ // Note that the executor passed to the RestClient constructor is not the same as the
+ // executor used by Netty
+ try (final RestClient restClient =
+ new RestClient(new Configuration(), Executors.directExecutor())) {
+ // Intentionally close the client (and thus also the executor used by Netty)
+ restClient.close();
+
+ CompletableFuture<?> future =
+ restClient.sendRequest(
+ unroutableIp,
+ 80,
+ new TestMessageHeaders(),
+ EmptyMessageParameters.getInstance(),
+ EmptyRequestBody.getInstance());
+
+ // Call get() on the future with a timeout of 0 so we can test that the exception thrown
+ // is not a TimeoutException, which is what would be thrown if restClient were not
+ // already closed
+ final ThrowingRunnable getFuture = () -> future.get(0, TimeUnit.SECONDS);
+
+ final ExecutionException executionException =
+ assertThrows(ExecutionException.class, getFuture);
+ final Throwable throwable = ExceptionUtils.stripExecutionException(executionException);
+ assertThat(throwable, instanceOf(IOException.class));
+ assertThat(throwable.getMessage(), containsString("RestClient is closed"));
Review Comment:
Sounds good, though the same should be true of `testConnectionTimeout` in this same file which I followed the pattern of—do Flink contribution expectations say I should make the same change to that test or leave it alone?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] patricklucas commented on pull request #22987: [FLINK-32583][rest] Fix deadlock in RestClient
Posted by "patricklucas (via GitHub)" <gi...@apache.org>.
patricklucas commented on PR #22987:
URL: https://github.com/apache/flink/pull/22987#issuecomment-1653394915
> @XComp CI looks promising, starting on the backport. Just 1.17 or also 1.16?
Hmm, it seems that there are various other changes that haven't been backported to release-1.17 that this PR implicitly relies on, including e.g. your change to add `OneShotLatch#awaitQuietly`. What is the general expectation for backports in terms of how deep to go? For instance, should I also backport your addition to add `awaitQuietly` or just wrap my call with the same try/catch?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] XComp commented on a diff in pull request #22987: [FLINK-32583][rest] Fix deadlock in RestClient
Posted by "XComp (via GitHub)" <gi...@apache.org>.
XComp commented on code in PR #22987:
URL: https://github.com/apache/flink/pull/22987#discussion_r1274946566
##########
flink-runtime/src/test/java/org/apache/flink/runtime/rest/RestClientTest.java:
##########
@@ -207,6 +218,120 @@ public void testRestClientClosedHandling() throws Exception {
}
}
+ /**
+ * Tests that the futures returned by {@link RestClient} fail immediately if the client is
+ * already closed.
+ *
+ * <p>See FLINK-32583
+ */
+ @Test
+ public void testCloseClientBeforeRequest() throws Exception {
+ try (final RestClient restClient =
+ new RestClient(new Configuration(), Executors.directExecutor())) {
+ restClient.close(); // Intentionally close the client prior to the request
+
+ CompletableFuture<?> future =
+ restClient.sendRequest(
+ unroutableIp,
+ 80,
+ new TestMessageHeaders(),
+ EmptyMessageParameters.getInstance(),
+ EmptyRequestBody.getInstance());
+
+ // Call get() on the future with a timeout of 0s so we can test that the exception
+ // thrown is not a TimeoutException, which is what would be thrown if restClient were
+ // not already closed
+ final ThrowingRunnable getFuture = () -> future.get(0, TimeUnit.SECONDS);
+
+ final Throwable cause = assertThrows(ExecutionException.class, getFuture).getCause();
+ assertThat(cause, instanceOf(IllegalStateException.class));
+ assertThat(cause.getMessage(), equalTo("RestClient is already closed"));
+ }
+ }
+
+ @Test
+ public void testCloseClientWhileProcessingRequest() throws Exception {
+ // Set up a Netty SelectStrategy with latches that allow us to step forward through Netty's
+ // request state machine, closing the client at a particular moment
+ final OneShotLatch connectTriggered = new OneShotLatch();
+ final OneShotLatch closeTriggered = new OneShotLatch();
+ final SelectStrategy fallbackSelectStrategy =
+ DefaultSelectStrategyFactory.INSTANCE.newSelectStrategy();
+ final SelectStrategyFactory selectStrategyFactory =
+ () ->
+ (selectSupplier, hasTasks) -> {
+ connectTriggered.trigger();
+ closeTriggered.awaitQuietly(1, TimeUnit.SECONDS);
Review Comment:
```suggestion
closeTriggered.awaitQuietly();
```
The community agreed on not using timeouts in tests (see [Flink Coding Conventions](https://flink.apache.org/how-to-contribute/code-style-and-quality-common/#avoid-timeouts-in-junit-tests)) since they are a source of instabilities. Instead the test would run until the overall CI build is cancelled after ~4hrs.
##########
flink-runtime/src/test/java/org/apache/flink/runtime/rest/RestClientTest.java:
##########
@@ -207,6 +218,120 @@ public void testRestClientClosedHandling() throws Exception {
}
}
+ /**
+ * Tests that the futures returned by {@link RestClient} fail immediately if the client is
+ * already closed.
+ *
+ * <p>See FLINK-32583
+ */
+ @Test
+ public void testCloseClientBeforeRequest() throws Exception {
+ try (final RestClient restClient =
+ new RestClient(new Configuration(), Executors.directExecutor())) {
+ restClient.close(); // Intentionally close the client prior to the request
+
+ CompletableFuture<?> future =
+ restClient.sendRequest(
+ unroutableIp,
+ 80,
+ new TestMessageHeaders(),
+ EmptyMessageParameters.getInstance(),
+ EmptyRequestBody.getInstance());
+
+ // Call get() on the future with a timeout of 0s so we can test that the exception
+ // thrown is not a TimeoutException, which is what would be thrown if restClient were
+ // not already closed
+ final ThrowingRunnable getFuture = () -> future.get(0, TimeUnit.SECONDS);
+
+ final Throwable cause = assertThrows(ExecutionException.class, getFuture).getCause();
+ assertThat(cause, instanceOf(IllegalStateException.class));
+ assertThat(cause.getMessage(), equalTo("RestClient is already closed"));
+ }
+ }
+
+ @Test
+ public void testCloseClientWhileProcessingRequest() throws Exception {
+ // Set up a Netty SelectStrategy with latches that allow us to step forward through Netty's
+ // request state machine, closing the client at a particular moment
+ final OneShotLatch connectTriggered = new OneShotLatch();
+ final OneShotLatch closeTriggered = new OneShotLatch();
+ final SelectStrategy fallbackSelectStrategy =
+ DefaultSelectStrategyFactory.INSTANCE.newSelectStrategy();
+ final SelectStrategyFactory selectStrategyFactory =
+ () ->
+ (selectSupplier, hasTasks) -> {
+ connectTriggered.trigger();
+ closeTriggered.awaitQuietly(1, TimeUnit.SECONDS);
+
+ return fallbackSelectStrategy.calculateStrategy(
+ selectSupplier, hasTasks);
+ };
+
+ try (final RestClient restClient =
+ new RestClient(
+ new Configuration(), Executors.directExecutor(), selectStrategyFactory)) {
+ // Check that client's internal collection of pending response futures is empty prior to
+ // the request
+ assertThat(restClient.getResponseChannelFutures(), empty());
+
+ final CompletableFuture<?> requestFuture =
+ restClient.sendRequest(
+ unroutableIp,
+ 80,
+ new TestMessageHeaders(),
+ EmptyMessageParameters.getInstance(),
+ EmptyRequestBody.getInstance());
+
+ // Check that client's internal collection of pending response futures now has one
+ // entry, presumably due to the call to sendRequest
+ assertThat(restClient.getResponseChannelFutures(), hasSize(1));
+
+ // Wait for Netty to start connecting, then while it's paused in the SelectStrategy,
+ // close the client before unpausing Netty
+ connectTriggered.await(1, TimeUnit.SECONDS);
Review Comment:
same here: No timeout should be used.
##########
flink-runtime/src/main/java/org/apache/flink/runtime/rest/RestClient.java:
##########
@@ -150,11 +159,30 @@ public static RestClient forUrl(Configuration configuration, Executor executor,
public RestClient(Configuration configuration, Executor executor)
throws ConfigurationException {
- this(configuration, executor, null, -1);
+ this(configuration, executor, null, -1, DefaultSelectStrategyFactory.INSTANCE);
Review Comment:
True, I overlooked that we're doing it your way in for the default port `-1`. Feel free to revert it if you want to be more consistent in this class :innocent:
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] patricklucas commented on pull request #22987: [FLINK-32583][rest] Fix deadlock in RestClient
Posted by "patricklucas (via GitHub)" <gi...@apache.org>.
patricklucas commented on PR #22987:
URL: https://github.com/apache/flink/pull/22987#issuecomment-1651696448
Lost a minor change during a rebase, sorry for the noise.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] patricklucas commented on a diff in pull request #22987: [FLINK-32583][rest] Fix deadlock in RestClient
Posted by "patricklucas (via GitHub)" <gi...@apache.org>.
patricklucas commented on code in PR #22987:
URL: https://github.com/apache/flink/pull/22987#discussion_r1275249702
##########
flink-runtime/src/test/java/org/apache/flink/runtime/rest/RestClientTest.java:
##########
@@ -207,6 +218,120 @@ public void testRestClientClosedHandling() throws Exception {
}
}
+ /**
+ * Tests that the futures returned by {@link RestClient} fail immediately if the client is
+ * already closed.
+ *
+ * <p>See FLINK-32583
+ */
+ @Test
+ public void testCloseClientBeforeRequest() throws Exception {
+ try (final RestClient restClient =
+ new RestClient(new Configuration(), Executors.directExecutor())) {
+ restClient.close(); // Intentionally close the client prior to the request
+
+ CompletableFuture<?> future =
+ restClient.sendRequest(
+ unroutableIp,
+ 80,
+ new TestMessageHeaders(),
+ EmptyMessageParameters.getInstance(),
+ EmptyRequestBody.getInstance());
+
+ // Call get() on the future with a timeout of 0s so we can test that the exception
+ // thrown is not a TimeoutException, which is what would be thrown if restClient were
+ // not already closed
+ final ThrowingRunnable getFuture = () -> future.get(0, TimeUnit.SECONDS);
+
+ final Throwable cause = assertThrows(ExecutionException.class, getFuture).getCause();
+ assertThat(cause, instanceOf(IllegalStateException.class));
+ assertThat(cause.getMessage(), equalTo("RestClient is already closed"));
+ }
+ }
+
+ @Test
+ public void testCloseClientWhileProcessingRequest() throws Exception {
+ // Set up a Netty SelectStrategy with latches that allow us to step forward through Netty's
+ // request state machine, closing the client at a particular moment
+ final OneShotLatch connectTriggered = new OneShotLatch();
+ final OneShotLatch closeTriggered = new OneShotLatch();
+ final SelectStrategy fallbackSelectStrategy =
+ DefaultSelectStrategyFactory.INSTANCE.newSelectStrategy();
+ final SelectStrategyFactory selectStrategyFactory =
+ () ->
+ (selectSupplier, hasTasks) -> {
+ connectTriggered.trigger();
+ closeTriggered.awaitQuietly(1, TimeUnit.SECONDS);
Review Comment:
Alright, no problem following that guidance—I actually hit it locally and it took a few minutes to track down this instance which should always resolve in milliseconds. I considered the risks but am happy to follow the convention.
##########
flink-runtime/src/test/java/org/apache/flink/runtime/rest/RestClientTest.java:
##########
@@ -207,6 +218,120 @@ public void testRestClientClosedHandling() throws Exception {
}
}
+ /**
+ * Tests that the futures returned by {@link RestClient} fail immediately if the client is
+ * already closed.
+ *
+ * <p>See FLINK-32583
+ */
+ @Test
+ public void testCloseClientBeforeRequest() throws Exception {
+ try (final RestClient restClient =
+ new RestClient(new Configuration(), Executors.directExecutor())) {
+ restClient.close(); // Intentionally close the client prior to the request
+
+ CompletableFuture<?> future =
+ restClient.sendRequest(
+ unroutableIp,
+ 80,
+ new TestMessageHeaders(),
+ EmptyMessageParameters.getInstance(),
+ EmptyRequestBody.getInstance());
+
+ // Call get() on the future with a timeout of 0s so we can test that the exception
+ // thrown is not a TimeoutException, which is what would be thrown if restClient were
+ // not already closed
+ final ThrowingRunnable getFuture = () -> future.get(0, TimeUnit.SECONDS);
+
+ final Throwable cause = assertThrows(ExecutionException.class, getFuture).getCause();
+ assertThat(cause, instanceOf(IllegalStateException.class));
+ assertThat(cause.getMessage(), equalTo("RestClient is already closed"));
+ }
+ }
+
+ @Test
+ public void testCloseClientWhileProcessingRequest() throws Exception {
+ // Set up a Netty SelectStrategy with latches that allow us to step forward through Netty's
+ // request state machine, closing the client at a particular moment
+ final OneShotLatch connectTriggered = new OneShotLatch();
+ final OneShotLatch closeTriggered = new OneShotLatch();
+ final SelectStrategy fallbackSelectStrategy =
+ DefaultSelectStrategyFactory.INSTANCE.newSelectStrategy();
+ final SelectStrategyFactory selectStrategyFactory =
+ () ->
+ (selectSupplier, hasTasks) -> {
+ connectTriggered.trigger();
+ closeTriggered.awaitQuietly(1, TimeUnit.SECONDS);
Review Comment:
Alright, no problem following that guidance—I actually hit it locally and it took a few minutes to track down this instance which should always resolve in milliseconds. I considered the risks but am happy to follow the convention.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] XComp commented on a diff in pull request #22987: [FLINK-32583][rest] Fix deadlock in RestClient
Posted by "XComp (via GitHub)" <gi...@apache.org>.
XComp commented on code in PR #22987:
URL: https://github.com/apache/flink/pull/22987#discussion_r1263387814
##########
flink-runtime/src/main/java/org/apache/flink/runtime/rest/RestClient.java:
##########
@@ -501,6 +501,14 @@ private <P extends ResponseBody> CompletableFuture<P> submitRequest(
}
});
+ // [FLINK-32583] If connectFuture failed instantly but channelFuture is unresolved, it may
+ // mean the executor service Netty is using has shut down, in which case the above listener
+ // to complete channelFuture will never run
+ if (connectFuture.isDone() && !connectFuture.isSuccess() && !channelFuture.isDone()) {
+ channelFuture.completeExceptionally(
+ new IOException("Netty connection failed instantly. Shutting down?"));
+ }
Review Comment:
`IOException` seems to be reasonable. At least, I can't come up with a better alternative :thinking:
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] XComp commented on a diff in pull request #22987: [FLINK-32583][rest] Fix deadlock in RestClient
Posted by "XComp (via GitHub)" <gi...@apache.org>.
XComp commented on code in PR #22987:
URL: https://github.com/apache/flink/pull/22987#discussion_r1274576162
##########
flink-runtime/src/test/java/org/apache/flink/runtime/rest/RestClientTest.java:
##########
@@ -207,6 +210,40 @@ public void testRestClientClosedHandling() throws Exception {
}
}
+ /**
+ * Tests that the futures returned by {@link RestClient} fail immediately if the client is
+ * already closed.
+ *
+ * <p>See FLINK-32583
+ */
+ @Test
+ public void testCloseClientBeforeRequest() throws Exception {
Review Comment:
that's a fair point - I know what you mean. :-D You could use JavaDoc to underline that. But using the compiler as a safeguard is a better option, I guess.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] patricklucas commented on pull request #22987: [FLINK-32583][rest] Fix deadlock in RestClient
Posted by "patricklucas (via GitHub)" <gi...@apache.org>.
patricklucas commented on PR #22987:
URL: https://github.com/apache/flink/pull/22987#issuecomment-1647388754
Rebased with master, including the fix for FLINK-32632.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] patricklucas commented on pull request #22987: [FLINK-32583][rest] Fix deadlock in RestClient
Posted by "patricklucas (via GitHub)" <gi...@apache.org>.
patricklucas commented on PR #22987:
URL: https://github.com/apache/flink/pull/22987#issuecomment-1653115641
Alrighty, should be good to go, hopefully CI goes through.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] patricklucas commented on pull request #22987: [FLINK-32583][rest] Fix deadlock in RestClient
Posted by "patricklucas (via GitHub)" <gi...@apache.org>.
patricklucas commented on PR #22987:
URL: https://github.com/apache/flink/pull/22987#issuecomment-1653380951
@XComp CI looks promising, starting on the backport. Just 1.17 or also 1.16?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] XComp commented on a diff in pull request #22987: [FLINK-32583][rest] Fix deadlock in RestClient
Posted by "XComp (via GitHub)" <gi...@apache.org>.
XComp commented on code in PR #22987:
URL: https://github.com/apache/flink/pull/22987#discussion_r1269401834
##########
flink-runtime/src/test/java/org/apache/flink/runtime/rest/RestClientTest.java:
##########
@@ -207,6 +209,42 @@ public void testRestClientClosedHandling() throws Exception {
}
}
+ /**
+ * Tests that the futures returned by {@link RestClient} fail immediately if the client is
+ * already closed.
+ *
+ * <p>See FLINK-32583
+ */
+ @Test
+ public void testCloseClientBeforeRequest() throws Exception {
+ // Note that the executor passed to the RestClient constructor is not the same as the
+ // executor used by Netty
+ try (final RestClient restClient =
+ new RestClient(new Configuration(), Executors.directExecutor())) {
+ // Intentionally close the client (and thus also the executor used by Netty)
+ restClient.close();
+
+ CompletableFuture<?> future =
+ restClient.sendRequest(
+ unroutableIp,
+ 80,
+ new TestMessageHeaders(),
+ EmptyMessageParameters.getInstance(),
+ EmptyRequestBody.getInstance());
+
+ // Call get() on the future with a timeout of 0 so we can test that the exception thrown
+ // is not a TimeoutException, which is what would be thrown if restClient were not
+ // already closed
+ final ThrowingRunnable getFuture = () -> future.get(0, TimeUnit.SECONDS);
+
+ final ExecutionException executionException =
+ assertThrows(ExecutionException.class, getFuture);
+ final Throwable throwable = ExceptionUtils.stripExecutionException(executionException);
+ assertThat(throwable, instanceOf(IOException.class));
+ assertThat(throwable.getMessage(), containsString("RestClient is closed"));
Review Comment:
You could change the other test(s) in a hotfix commit...
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] patricklucas commented on a diff in pull request #22987: [FLINK-32583][rest] Fix deadlock in RestClient
Posted by "patricklucas (via GitHub)" <gi...@apache.org>.
patricklucas commented on code in PR #22987:
URL: https://github.com/apache/flink/pull/22987#discussion_r1273174325
##########
flink-runtime/src/test/java/org/apache/flink/runtime/rest/RestClientTest.java:
##########
@@ -207,6 +210,40 @@ public void testRestClientClosedHandling() throws Exception {
}
}
+ /**
+ * Tests that the futures returned by {@link RestClient} fail immediately if the client is
+ * already closed.
+ *
+ * <p>See FLINK-32583
+ */
+ @Test
+ public void testCloseClientBeforeRequest() throws Exception {
Review Comment:
With this additional test, the `isRunning` codepath is now tested, but not the branch where the listener is unable to be attached and is thus handled in `notifyResponseFuturesOfShutdown()`. I think there is just a microscopic window for this to happen, between the calls to `bootstrap.connect()` and `connectFuture.addListener()`, which are not executed atomically.
The only way that comes to mind to test that would be to inject a latch to enable calling `close()` between those two calls.
Do you think that's necessary? I could also add a note that that eventuality was tested manually, with `notifyResponseFuturesOfShutdown()` successfully resolving the future.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] XComp commented on a diff in pull request #22987: [FLINK-32583][rest] Fix deadlock in RestClient
Posted by "XComp (via GitHub)" <gi...@apache.org>.
XComp commented on code in PR #22987:
URL: https://github.com/apache/flink/pull/22987#discussion_r1273425760
##########
flink-runtime/src/main/java/org/apache/flink/runtime/rest/RestClient.java:
##########
@@ -150,11 +159,30 @@ public static RestClient forUrl(Configuration configuration, Executor executor,
public RestClient(Configuration configuration, Executor executor)
throws ConfigurationException {
- this(configuration, executor, null, -1);
+ this(configuration, executor, null, -1, DefaultSelectStrategyFactory.INSTANCE);
Review Comment:
```suggestion
this(configuration, executor, null, -1);
```
Why do you think that change is necessary? Isn't it better to have the default behavior defined in only one code location?
##########
flink-runtime/src/main/java/org/apache/flink/runtime/rest/RestClient.java:
##########
@@ -234,7 +262,11 @@ protected void initChannel(SocketChannel socketChannel) {
}
};
NioEventLoopGroup group =
- new NioEventLoopGroup(1, new ExecutorThreadFactory("flink-rest-client-netty"));
+ new NioEventLoopGroup(
+ 1,
+ new ExecutorThreadFactory("flink-rest-client-netty"),
+ SelectorProvider.provider(),
Review Comment:
```suggestion
// mimicking the default for NioEventLoopGroup(int, Executor) here because there is no constructor that allows us to change the SelectStrategy without specifying the SelectorProvider
SelectorProvider.provider(),
```
Shall we add a comment here that describes where `SelectorProvider.provider()` is coming from?
##########
flink-runtime/src/test/java/org/apache/flink/runtime/rest/RestClientTest.java:
##########
@@ -207,6 +210,40 @@ public void testRestClientClosedHandling() throws Exception {
}
}
+ /**
+ * Tests that the futures returned by {@link RestClient} fail immediately if the client is
+ * already closed.
+ *
+ * <p>See FLINK-32583
+ */
+ @Test
+ public void testCloseClientBeforeRequest() throws Exception {
Review Comment:
hm, you're right - that's a tricky one. We could extend the `RestClient` constructor that's labeled as `@VisibleForTesting` to not only provide the `SelectyStrategyFactory` but also a `Collection` to initialize the `responseChannelFutures` field. The default value would be `ConcurrentHashMap.newKeySet()`.
That would enable us to check in `testCloseClientWhileProcessingRequest` that the future is, indeed, added to the collection when calling connect (i.e. when executing the custom `SelectStrategy` callback) and cleared afterwards.
And then, we could add another test that adds a non-empty collection via the constructor to the `RestClient` and calls `closeAsync` on that client to verify that the future was completed exceptionally and the collection cleared afterwards. :thinking:
Maybe, I got too nitpicky on that one :innocent: WDYT?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] patricklucas commented on a diff in pull request #22987: [FLINK-32583][rest] Fix deadlock in RestClient
Posted by "patricklucas (via GitHub)" <gi...@apache.org>.
patricklucas commented on code in PR #22987:
URL: https://github.com/apache/flink/pull/22987#discussion_r1272567509
##########
flink-runtime/src/test/java/org/apache/flink/runtime/rest/RestClientTest.java:
##########
@@ -207,6 +210,40 @@ public void testRestClientClosedHandling() throws Exception {
}
}
+ /**
+ * Tests that the futures returned by {@link RestClient} fail immediately if the client is
+ * already closed.
+ *
+ * <p>See FLINK-32583
+ */
+ @Test
+ public void testCloseClientBeforeRequest() throws Exception {
Review Comment:
Cool, good find, I'll try it out.
Regarding the exception, I've started to feel that maybe `IOException` actually _isn't_ the best choice, since no true IO (e.g. via syscalls) is happening at the time of the failure. Even if we take an analogous example of trying to write to a file that's closed, the error would often come from the failure of the syscall to write.
In this situation, the client is in fact in an illegal state (closed) for making requests.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] XComp commented on a diff in pull request #22987: [FLINK-32583][rest] Fix deadlock in RestClient
Posted by "XComp (via GitHub)" <gi...@apache.org>.
XComp commented on code in PR #22987:
URL: https://github.com/apache/flink/pull/22987#discussion_r1271900214
##########
flink-runtime/src/main/java/org/apache/flink/runtime/rest/RestClient.java:
##########
@@ -289,6 +298,16 @@ private CompletableFuture<Void> shutdownInternally(Time timeout) {
return terminationFuture;
}
+ private void notifyResponseFuturesOfShutdown() {
+ responseChannelFutures
+ .keySet()
+ .forEach(
+ future ->
+ future.completeExceptionally(
+ new IOException("RestClient is closed")));
Review Comment:
```suggestion
new IOException("RestClient was closed while processing the request.")));
```
proposal in contrast to the `IOException` error message I proposed in [the related comment above](https://github.com/apache/flink/pull/22987#discussion_r1271898836).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] XComp commented on a diff in pull request #22987: [FLINK-32583][rest] Fix deadlock in RestClient
Posted by "XComp (via GitHub)" <gi...@apache.org>.
XComp commented on code in PR #22987:
URL: https://github.com/apache/flink/pull/22987#discussion_r1269485273
##########
flink-runtime/src/main/java/org/apache/flink/runtime/rest/RestClient.java:
##########
@@ -501,6 +501,22 @@ private <P extends ResponseBody> CompletableFuture<P> submitRequest(
}
});
+ // [FLINK-32583] If connectFuture failed instantly but channelFuture is unresolved, it may
+ // mean the executor service Netty is using has shut down, in which case the above listener
+ // to complete channelFuture will never run
+ if (connectFuture.isDone() && !connectFuture.isSuccess() && !channelFuture.isDone()) {
Review Comment:
I guess your last proposal makes sense. Could you add this as a proper commit to make it reviewable? And it's not too complicated.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] patricklucas commented on a diff in pull request #22987: [FLINK-32583][rest] Fix deadlock in RestClient
Posted by "patricklucas (via GitHub)" <gi...@apache.org>.
patricklucas commented on code in PR #22987:
URL: https://github.com/apache/flink/pull/22987#discussion_r1263539507
##########
flink-runtime/src/main/java/org/apache/flink/runtime/rest/RestClient.java:
##########
@@ -501,6 +501,22 @@ private <P extends ResponseBody> CompletableFuture<P> submitRequest(
}
});
+ // [FLINK-32583] If connectFuture failed instantly but channelFuture is unresolved, it may
+ // mean the executor service Netty is using has shut down, in which case the above listener
+ // to complete channelFuture will never run
+ if (connectFuture.isDone() && !connectFuture.isSuccess() && !channelFuture.isDone()) {
Review Comment:
The diff for RestClient if we just have it preemptively check `isRunning` (the test can remain the same):
```diff
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 2055a2227cf..e9f57c2f4ce 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
@@ -488,18 +488,22 @@ public class RestClient implements AutoCloseableAsync {
private <P extends ResponseBody> CompletableFuture<P> submitRequest(
String targetAddress, int targetPort, Request httpRequest, JavaType responseType) {
- final ChannelFuture connectFuture = bootstrap.connect(targetAddress, targetPort);
-
final CompletableFuture<Channel> channelFuture = new CompletableFuture<>();
- connectFuture.addListener(
- (ChannelFuture future) -> {
- if (future.isSuccess()) {
- channelFuture.complete(future.channel());
- } else {
- channelFuture.completeExceptionally(future.cause());
- }
- });
+ if (isRunning.get()) {
+ final ChannelFuture connectFuture = bootstrap.connect(targetAddress, targetPort);
+
+ connectFuture.addListener(
+ (ChannelFuture future) -> {
+ if (future.isSuccess()) {
+ channelFuture.complete(future.channel());
+ } else {
+ channelFuture.completeExceptionally(future.cause());
+ }
+ });
+ } else {
+ channelFuture.completeExceptionally(new IOException("RestClient is closed"));
+ }
return channelFuture
.thenComposeAsync(
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] patricklucas commented on a diff in pull request #22987: [FLINK-32583][rest] Fix deadlock in RestClient
Posted by "patricklucas (via GitHub)" <gi...@apache.org>.
patricklucas commented on code in PR #22987:
URL: https://github.com/apache/flink/pull/22987#discussion_r1272124153
##########
flink-runtime/src/main/java/org/apache/flink/runtime/rest/RestClient.java:
##########
@@ -131,6 +133,11 @@ public class RestClient implements AutoCloseableAsync {
private final String urlPrefix;
+ // Used to track unresolved request futures in case they need to be resolved when the client is
+ // closed
+ private final Map<CompletableFuture<Channel>, Void> responseChannelFutures =
+ Collections.synchronizedMap(new IdentityHashMap<>());
Review Comment:
I had gone with `IdentityHashMap` since we want to maintain references to the actual future objects, however since `CompletableFuture`'s `equals` and `hashCode` just delegate to `Object`, the behavior is the same. I'll make the change.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] XComp commented on a diff in pull request #22987: [FLINK-32583][rest] Fix deadlock in RestClient
Posted by "XComp (via GitHub)" <gi...@apache.org>.
XComp commented on code in PR #22987:
URL: https://github.com/apache/flink/pull/22987#discussion_r1271887156
##########
flink-runtime/src/main/java/org/apache/flink/runtime/rest/RestClient.java:
##########
@@ -131,6 +133,11 @@ public class RestClient implements AutoCloseableAsync {
private final String urlPrefix;
+ // Used to track unresolved request futures in case they need to be resolved when the client is
+ // closed
+ private final Map<CompletableFuture<Channel>, Void> responseChannelFutures =
+ Collections.synchronizedMap(new IdentityHashMap<>());
Review Comment:
```suggestion
private final Collection<CompletableFuture<Channel>> responseChannelFutures =
ConcurrentHashMap.newKeySet();
```
nit: What about using `ConcurrentHashMap.newKeySet()` here? That simplifies the change a little bit.
##########
flink-runtime/src/main/java/org/apache/flink/runtime/rest/RestClient.java:
##########
@@ -514,18 +533,25 @@ private static Request createRequest(
private <P extends ResponseBody> CompletableFuture<P> submitRequest(
String targetAddress, int targetPort, Request httpRequest, JavaType responseType) {
- final ChannelFuture connectFuture = bootstrap.connect(targetAddress, targetPort);
-
final CompletableFuture<Channel> channelFuture = new CompletableFuture<>();
+ responseChannelFutures.put(channelFuture, null);
- connectFuture.addListener(
- (ChannelFuture future) -> {
- if (future.isSuccess()) {
- channelFuture.complete(future.channel());
- } else {
- channelFuture.completeExceptionally(future.cause());
- }
- });
+ if (isRunning.get()) {
Review Comment:
```suggestion
if (!isRunning.get()) {
return FutureUtils.completedExceptionally(new IOException("RestClient is closed"));
}
```
Inverting the condition here and moving this if block to the top of the method makes the diff smaller. And it makes the code also more readable due to less indentations, I feel.
##########
flink-runtime/src/main/java/org/apache/flink/runtime/rest/RestClient.java:
##########
@@ -289,6 +298,16 @@ private CompletableFuture<Void> shutdownInternally(Time timeout) {
return terminationFuture;
}
+ private void notifyResponseFuturesOfShutdown() {
+ responseChannelFutures
+ .keySet()
+ .forEach(
+ future ->
+ future.completeExceptionally(
+ new IOException("RestClient is closed")));
Review Comment:
```suggestion
new IOException("RestClient was closed while processing the request.")));
```
proposal in contrast to the other `IOException` error message.
##########
flink-runtime/src/test/java/org/apache/flink/runtime/rest/RestClientTest.java:
##########
@@ -207,6 +210,40 @@ public void testRestClientClosedHandling() throws Exception {
}
}
+ /**
+ * Tests that the futures returned by {@link RestClient} fail immediately if the client is
+ * already closed.
+ *
+ * <p>See FLINK-32583
+ */
+ @Test
+ public void testCloseClientBeforeRequest() throws Exception {
Review Comment:
I looked into the multi-threading of netty a bit more and found the `SelectStrategy` a viable tool to control the state of the request. I came up with the following test:
```
@Test
public void testCloseClientWhileProcessingRequest() throws Exception {
final OneShotLatch connectTriggered = new OneShotLatch();
final OneShotLatch closeTriggered = new OneShotLatch();
final SelectStrategy fallbackSelectStrategy =
DefaultSelectStrategyFactory.INSTANCE.newSelectStrategy();
final SelectStrategy selectStrategy =
(selectSupplier, hasTasks) -> {
connectTriggered.trigger();
closeTriggered.awaitQuietly();
return fallbackSelectStrategy.calculateStrategy(selectSupplier, hasTasks);
};
// Note that the executor passed to the RestClient constructor is not the same as the
// executor used by Netty
try (final RestClient restClient =
new RestClient(
new Configuration(), Executors.directExecutor(), () -> selectStrategy)) {
final CompletableFuture<?> requestFuture =
restClient.sendRequest(
unroutableIp,
80,
new TestMessageHeaders(),
EmptyMessageParameters.getInstance(),
EmptyRequestBody.getInstance());
connectTriggered.await();
final CompletableFuture<Void> closeFuture = restClient.closeAsync();
closeTriggered.trigger();
// close should complete successfully
closeFuture.get();
final Throwable cause =
assertThrows(
ExecutionException.class,
() -> requestFuture.get(0, TimeUnit.SECONDS))
.getCause();
assertThat(cause, instanceOf(IllegalStateException.class));
}
}
```
It would require an extension of the `RestClient` adding a package-private constructor:
```
// ...
public RestClient(Configuration configuration, Executor executor)
throws ConfigurationException {
this(configuration, executor, DefaultSelectStrategyFactory.INSTANCE);
}
@VisibleForTesting
RestClient(
Configuration configuration,
Executor executor,
SelectStrategyFactory selectStrategyFactory)
//...
NioEventLoopGroup group =
new NioEventLoopGroup(
1,
new ExecutorThreadFactory("flink-rest-client-netty"),
SelectorProvider.provider(),
selectStrategyFactory);
// ...
```
WDYT? That should test the `isRunning.get()` code path.
##########
flink-runtime/src/test/java/org/apache/flink/runtime/rest/RestClientTest.java:
##########
@@ -207,6 +210,40 @@ public void testRestClientClosedHandling() throws Exception {
}
}
+ /**
+ * Tests that the futures returned by {@link RestClient} fail immediately if the client is
+ * already closed.
+ *
+ * <p>See FLINK-32583
+ */
+ @Test
+ public void testCloseClientBeforeRequest() throws Exception {
+ // Note that the executor passed to the RestClient constructor is not the same as the
+ // executor used by Netty
+ try (final RestClient restClient =
+ new RestClient(new Configuration(), Executors.directExecutor())) {
+ // Intentionally close the client (and thus also the executor used by Netty)
+ restClient.close();
+
+ CompletableFuture<?> future =
+ restClient.sendRequest(
+ unroutableIp,
+ 80,
+ new TestMessageHeaders(),
+ EmptyMessageParameters.getInstance(),
+ EmptyRequestBody.getInstance());
+
+ // Call get() on the future with a timeout of 0 so we can test that the exception thrown
Review Comment:
```suggestion
// Call get() on the future with a timeout of 0s so we can test that the exception thrown
```
##########
flink-runtime/src/main/java/org/apache/flink/runtime/rest/RestClient.java:
##########
@@ -514,18 +533,25 @@ private static Request createRequest(
private <P extends ResponseBody> CompletableFuture<P> submitRequest(
String targetAddress, int targetPort, Request httpRequest, JavaType responseType) {
- final ChannelFuture connectFuture = bootstrap.connect(targetAddress, targetPort);
-
final CompletableFuture<Channel> channelFuture = new CompletableFuture<>();
+ responseChannelFutures.put(channelFuture, null);
- connectFuture.addListener(
- (ChannelFuture future) -> {
- if (future.isSuccess()) {
- channelFuture.complete(future.channel());
- } else {
- channelFuture.completeExceptionally(future.cause());
- }
- });
+ if (isRunning.get()) {
+ final ChannelFuture connectFuture = bootstrap.connect(targetAddress, targetPort);
+
+ connectFuture.addListener(
+ (ChannelFuture future) -> {
+ responseChannelFutures.remove(channelFuture);
+
+ if (future.isSuccess()) {
+ channelFuture.complete(future.channel());
+ } else {
+ channelFuture.completeExceptionally(future.cause());
+ }
+ });
+ } else {
+ channelFuture.completeExceptionally(new IOException("RestClient is closed"));
Review Comment:
```suggestion
channelFuture.completeExceptionally(new IOException("RestClient was closed while submitting the request."));
```
Should we give more context to differentiate the two newly introduced `IOException`s?
##########
flink-runtime/src/test/java/org/apache/flink/runtime/rest/RestClientTest.java:
##########
@@ -207,6 +210,40 @@ public void testRestClientClosedHandling() throws Exception {
}
}
+ /**
+ * Tests that the futures returned by {@link RestClient} fail immediately if the client is
+ * already closed.
+ *
+ * <p>See FLINK-32583
+ */
+ @Test
+ public void testCloseClientBeforeRequest() throws Exception {
Review Comment:
Btw. in that case, the future is failing with an `IllegalStateException`. I'm wondering whether we should switch to that one within the `RestClient` as well to make it consistent. Or we wrap the `IllegalStateException` in an `IOException` because I think that `IOException` still seems to be the best fit. :thinking:
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] patricklucas commented on a diff in pull request #22987: [FLINK-32583][rest] Fix deadlock in RestClient
Posted by "patricklucas (via GitHub)" <gi...@apache.org>.
patricklucas commented on code in PR #22987:
URL: https://github.com/apache/flink/pull/22987#discussion_r1274549521
##########
flink-runtime/src/test/java/org/apache/flink/runtime/rest/RestClientTest.java:
##########
@@ -207,6 +210,40 @@ public void testRestClientClosedHandling() throws Exception {
}
}
+ /**
+ * Tests that the futures returned by {@link RestClient} fail immediately if the client is
+ * already closed.
+ *
+ * <p>See FLINK-32583
+ */
+ @Test
+ public void testCloseClientBeforeRequest() throws Exception {
Review Comment:
Might be able to get the same behavior with just a `@VisibleForTesting` getter for `responseChannelFutures`, I'll give it a shot.
Always makes me uneasy allowing injection or adding extra steps to initializing `ConcurrentXyz` fields, even for testing, since you can't enforce the concurrency-safety with the type system.
##########
flink-runtime/src/main/java/org/apache/flink/runtime/rest/RestClient.java:
##########
@@ -150,11 +159,30 @@ public static RestClient forUrl(Configuration configuration, Executor executor,
public RestClient(Configuration configuration, Executor executor)
throws ConfigurationException {
- this(configuration, executor, null, -1);
+ this(configuration, executor, null, -1, DefaultSelectStrategyFactory.INSTANCE);
Review Comment:
Once again, just wasn't sure what Flink's preferred style is—reducing the number of constructor indirections or reducing the number of default values materialized. :)
Happy to change.
##########
flink-runtime/src/main/java/org/apache/flink/runtime/rest/RestClient.java:
##########
@@ -150,11 +159,30 @@ public static RestClient forUrl(Configuration configuration, Executor executor,
public RestClient(Configuration configuration, Executor executor)
throws ConfigurationException {
- this(configuration, executor, null, -1);
+ this(configuration, executor, null, -1, DefaultSelectStrategyFactory.INSTANCE);
Review Comment:
Once again, just wasn't sure what Flink's preferred style is—reducing the number of constructor indirections or reducing the number of default values materialized. :)
Happy to change.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] patricklucas commented on a diff in pull request #22987: [FLINK-32583][rest] Fix deadlock in RestClient
Posted by "patricklucas (via GitHub)" <gi...@apache.org>.
patricklucas commented on code in PR #22987:
URL: https://github.com/apache/flink/pull/22987#discussion_r1264995814
##########
flink-runtime/src/main/java/org/apache/flink/runtime/rest/RestClient.java:
##########
@@ -501,6 +501,22 @@ private <P extends ResponseBody> CompletableFuture<P> submitRequest(
}
});
+ // [FLINK-32583] If connectFuture failed instantly but channelFuture is unresolved, it may
+ // mean the executor service Netty is using has shut down, in which case the above listener
+ // to complete channelFuture will never run
+ if (connectFuture.isDone() && !connectFuture.isSuccess() && !channelFuture.isDone()) {
Review Comment:
@XComp After manually trying out various orderings of the threaded operations, I've come to the conclusion that as long as `addListener` is called before the client is closed, the listener will be run. This means there is only a minuscule period between the non-blocking call to `bootstrap.connect()` and `connectFuture.addListener()` where this situation could arise.
So again, I think the vast likelihood of hitting this issue is resolved with the simple `isRunning` check from the diff in my previous comment, while 100% coverage would require tracking the futures and resolving them on client shutdown, though with a higher chance of introducing another bug. Its behavior would also be difficult to test.
Including the future tracking would look something like this:
```diff
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 2055a2227cf..c986f7c6292 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
@@ -96,8 +96,10 @@ import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.Comparator;
+import java.util.IdentityHashMap;
import java.util.Iterator;
import java.util.List;
+import java.util.Map;
import java.util.Optional;
import java.util.ServiceLoader;
import java.util.concurrent.CompletableFuture;
@@ -127,6 +129,9 @@ public class RestClient implements AutoCloseableAsync {
private final String urlPrefix;
+ private final Map<CompletableFuture<Channel>, Void> responseChannelFutures =
+ Collections.synchronizedMap(new IdentityHashMap<>());
+
@VisibleForTesting List<OutboundChannelHandlerFactory> outboundChannelHandlerFactories;
public RestClient(Configuration configuration, Executor executor)
@@ -252,6 +257,8 @@ public class RestClient implements AutoCloseableAsync {
.shutdownGracefully(0L, timeout.toMilliseconds(), TimeUnit.MILLISECONDS)
.addListener(
finished -> {
+ notifyResponseFuturesOfShutdown();
+
if (finished.isSuccess()) {
terminationFuture.complete(null);
} else {
@@ -265,6 +272,16 @@ public class RestClient implements AutoCloseableAsync {
return terminationFuture;
}
+ private void notifyResponseFuturesOfShutdown() {
+ responseChannelFutures
+ .keySet()
+ .forEach(
+ future ->
+ future.completeExceptionally(
+ new IOException("RestClient is closed")));
+ responseChannelFutures.clear();
+ }
+
public <
M extends MessageHeaders<EmptyRequestBody, P, EmptyMessageParameters>,
P extends ResponseBody>
@@ -488,18 +505,25 @@ public class RestClient implements AutoCloseableAsync {
private <P extends ResponseBody> CompletableFuture<P> submitRequest(
String targetAddress, int targetPort, Request httpRequest, JavaType responseType) {
- final ChannelFuture connectFuture = bootstrap.connect(targetAddress, targetPort);
-
final CompletableFuture<Channel> channelFuture = new CompletableFuture<>();
+ responseChannelFutures.put(channelFuture, null);
- connectFuture.addListener(
- (ChannelFuture future) -> {
- if (future.isSuccess()) {
- channelFuture.complete(future.channel());
- } else {
- channelFuture.completeExceptionally(future.cause());
- }
- });
+ if (isRunning.get()) {
+ final ChannelFuture connectFuture = bootstrap.connect(targetAddress, targetPort);
+
+ connectFuture.addListener(
+ (ChannelFuture future) -> {
+ responseChannelFutures.remove(channelFuture);
+
+ if (future.isSuccess()) {
+ channelFuture.complete(future.channel());
+ } else {
+ channelFuture.completeExceptionally(future.cause());
+ }
+ });
+ } else {
+ channelFuture.completeExceptionally(new IOException("RestClient is closed"));
+ }
return channelFuture
.thenComposeAsync(
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] patricklucas commented on a diff in pull request #22987: [FLINK-32583][rest] Fix deadlock in RestClient
Posted by "patricklucas (via GitHub)" <gi...@apache.org>.
patricklucas commented on code in PR #22987:
URL: https://github.com/apache/flink/pull/22987#discussion_r1263527019
##########
flink-runtime/src/main/java/org/apache/flink/runtime/rest/RestClient.java:
##########
@@ -501,6 +501,22 @@ private <P extends ResponseBody> CompletableFuture<P> submitRequest(
}
});
+ // [FLINK-32583] If connectFuture failed instantly but channelFuture is unresolved, it may
+ // mean the executor service Netty is using has shut down, in which case the above listener
+ // to complete channelFuture will never run
+ if (connectFuture.isDone() && !connectFuture.isSuccess() && !channelFuture.isDone()) {
Review Comment:
While trying to get a better picture of the various cases we might need to handle to get full coverage here I took a step back and tried to get a broader view of this.
I verified that the event loop/executor for both the connection as well as its listeners is the same, the NioEventLoopGroup `bootstrap` is configured with elsewhere in RestClient, and I saw that when the client is shut down, this group is asked to shut down gracefully with a "quiet period" of 0 and a default timeout of 10 seconds. As far as I can tell, already-scheduled tasks are allowed to run to completion.
I believe we could cover the vast majority of cases by simply checking `isRunning` at the top of `submitRequest`—if the client has already been closed and graceful shutdown of the group initiated, we should immediately reject any new requests.
The challenge in particular is covering the case where the client is shut down after the call to `bootstrap.connect()` but before the listeners are notified. Depending on configured timeouts, the connect task may run to completion or time out, but not be able to notify its listeners and thus resolve `channelFuture`. I don't think we have any way to detect this within `submitRequest` itself.
The only solution that comes to mind here would be to:
- Add each `channelFuture` created in `submitRequest` to a class-level collection
- In the listener attached to `connectFuture`, remove the `channelFuture` from that collection after calling `complete`/`completeExceptionally`
- In the listener in `shutdownInternally` that is notified when the event loop shuts down, resolve all remaining futures in the collection exceptionally
Having written it down it doesn't sound as complex as I first thought, but it's still clearly beyond what I had originally hoped for this change.
@XComp Thoughts? Should I go with simply checking whether the client has already been closed at the top of `submitRequest`, implement this "future cleanup" logic, or both?
##########
flink-runtime/src/main/java/org/apache/flink/runtime/rest/RestClient.java:
##########
@@ -501,6 +501,22 @@ private <P extends ResponseBody> CompletableFuture<P> submitRequest(
}
});
+ // [FLINK-32583] If connectFuture failed instantly but channelFuture is unresolved, it may
+ // mean the executor service Netty is using has shut down, in which case the above listener
+ // to complete channelFuture will never run
+ if (connectFuture.isDone() && !connectFuture.isSuccess() && !channelFuture.isDone()) {
Review Comment:
While trying to get a better picture of the various cases we might need to handle to get full coverage here I took a step back and tried to get a broader view of this.
I verified that the event loop/executor for both the connection as well as its listeners is the same, the NioEventLoopGroup `bootstrap` is configured with elsewhere in RestClient, and I saw that when the client is shut down, this group is asked to shut down gracefully with a "quiet period" of 0 and a default timeout of 10 seconds. As far as I can tell, already-scheduled tasks are allowed to run to completion.
I believe we could cover the vast majority of cases by simply checking `isRunning` at the top of `submitRequest`—if the client has already been closed and graceful shutdown of the group initiated, we should immediately reject any new requests.
The challenge in particular is covering the case where the client is shut down after the call to `bootstrap.connect()` but before the listeners are notified. Depending on configured timeouts, the connect task may run to completion or time out, but not be able to notify its listeners and thus resolve `channelFuture`. I don't think we have any way to detect this within `submitRequest` itself.
The only solution that comes to mind here would be to:
- Add each `channelFuture` created in `submitRequest` to a class-level collection
- In the listener attached to `connectFuture`, remove the `channelFuture` from that collection after calling `complete`/`completeExceptionally`
- In the listener in `shutdownInternally` that is notified when the event loop shuts down, resolve all remaining futures in the collection exceptionally
Having written it down it doesn't sound as complex as I first thought, but it's still clearly beyond what I had originally hoped for this change.
@XComp Thoughts? Should I go with simply checking whether the client has already been closed at the top of `submitRequest`, implement this "future cleanup" logic, or both?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org