You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by tz...@apache.org on 2021/08/09 12:59:40 UTC
[flink-statefun] branch master updated: [hotfix] Recreate shared
OkHttpClient after shutdown in DefaultHttpRequestReplyClientFactory
This is an automated email from the ASF dual-hosted git repository.
tzulitai pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink-statefun.git
The following commit(s) were added to refs/heads/master by this push:
new b422bce [hotfix] Recreate shared OkHttpClient after shutdown in DefaultHttpRequestReplyClientFactory
b422bce is described below
commit b422bce0d24ef97a2b2d638d235b99d0f9fbff3f
Author: Tzu-Li (Gordon) Tai <tz...@apache.org>
AuthorDate: Mon Aug 9 20:57:11 2021 +0800
[hotfix] Recreate shared OkHttpClient after shutdown in DefaultHttpRequestReplyClientFactory
---
.../httpfn/DefaultHttpRequestReplyClientFactory.java | 20 ++++++++++++--------
1 file changed, 12 insertions(+), 8 deletions(-)
diff --git a/statefun-flink/statefun-flink-core/src/main/java/org/apache/flink/statefun/flink/core/httpfn/DefaultHttpRequestReplyClientFactory.java b/statefun-flink/statefun-flink-core/src/main/java/org/apache/flink/statefun/flink/core/httpfn/DefaultHttpRequestReplyClientFactory.java
index 2cfadbb..5387c65 100644
--- a/statefun-flink/statefun-flink-core/src/main/java/org/apache/flink/statefun/flink/core/httpfn/DefaultHttpRequestReplyClientFactory.java
+++ b/statefun-flink/statefun-flink-core/src/main/java/org/apache/flink/statefun/flink/core/httpfn/DefaultHttpRequestReplyClientFactory.java
@@ -40,9 +40,7 @@ public final class DefaultHttpRequestReplyClientFactory implements RequestReplyC
private static final ObjectMapper OBJ_MAPPER = StateFunObjectMapper.create();
/** lazily initialized by {@link #createTransportClient} */
- @Nullable private OkHttpClient sharedClient;
-
- private volatile boolean shutdown;
+ @Nullable private volatile OkHttpClient sharedClient;
private DefaultHttpRequestReplyClientFactory() {}
@@ -59,17 +57,18 @@ public final class DefaultHttpRequestReplyClientFactory implements RequestReplyC
@Override
public void cleanup() {
- if (!shutdown) {
- shutdown = true;
- OkHttpUtils.closeSilently(sharedClient);
- }
+ final OkHttpClient sharedClient = this.sharedClient;
+ this.sharedClient = null;
+ OkHttpUtils.closeSilently(sharedClient);
}
private DefaultHttpRequestReplyClient createClient(
ObjectNode transportProperties, URI endpointUrl) {
try (SetContextClassLoader ignored = new SetContextClassLoader(this)) {
+ OkHttpClient sharedClient = this.sharedClient;
if (sharedClient == null) {
sharedClient = OkHttpUtils.newClient();
+ this.sharedClient = sharedClient;
}
final OkHttpClient.Builder clientBuilder = sharedClient.newBuilder();
@@ -97,10 +96,15 @@ public final class DefaultHttpRequestReplyClientFactory implements RequestReplyC
url = HttpUrl.get(endpointUrl);
}
- return new DefaultHttpRequestReplyClient(url, clientBuilder.build(), () -> shutdown);
+ return new DefaultHttpRequestReplyClient(
+ url, clientBuilder.build(), () -> isShutdown(this.sharedClient));
}
}
+ private boolean isShutdown(OkHttpClient previousClient) {
+ return DefaultHttpRequestReplyClientFactory.this.sharedClient != previousClient;
+ }
+
private static DefaultHttpRequestReplyClientSpec parseTransportProperties(
ObjectNode transportClientProperties) {
try {