fix apache#7437 (apache#7687)

wanghhao · Jul 29, 2020 · f2532c8 · f2532c8
1 parent c76fd37
commit f2532c8
Show file tree

Hide file tree

Showing 2 changed files with 9 additions and 12 deletions.
diff --git a/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/PulsarAdmin.java b/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/PulsarAdmin.java
@@ -47,13 +47,13 @@
 import org.apache.pulsar.client.admin.internal.TenantsImpl;
 import org.apache.pulsar.client.admin.internal.TopicsImpl;
 import org.apache.pulsar.client.admin.internal.WorkerImpl;
+import org.apache.pulsar.client.admin.internal.http.AsyncHttpConnector;
 import org.apache.pulsar.client.admin.internal.http.AsyncHttpConnectorProvider;
 import org.apache.pulsar.client.api.Authentication;
 import org.apache.pulsar.client.api.AuthenticationFactory;
 import org.apache.pulsar.client.api.PulsarClientException;
 import org.apache.pulsar.client.impl.auth.AuthenticationDisabled;
 import org.apache.pulsar.client.impl.conf.ClientConfigurationData;
-import org.asynchttpclient.AsyncHttpClient;
 import org.glassfish.jersey.client.ClientConfig;
 import org.glassfish.jersey.client.ClientProperties;
 import org.glassfish.jersey.jackson.JacksonFeature;
@@ -86,7 +86,7 @@ public class PulsarAdmin implements Closeable {
     private final ResourceQuotas resourceQuotas;
     private final ClientConfigurationData clientConfigData;
     private final Client client;
-    private final AsyncHttpClient httpAsyncClient;
+    private final AsyncHttpConnector asyncHttpConnector;
     private final String serviceUrl;
     private final Lookup lookups;
     private final Functions functions;
@@ -184,10 +184,10 @@ public PulsarAdmin(String serviceUrl,
         this.serviceUrl = serviceUrl;
         root = client.target(serviceUrl);
 
-        this.httpAsyncClient = asyncConnectorProvider.getConnector(
+        this.asyncHttpConnector = asyncConnectorProvider.getConnector(
                 Math.toIntExact(connectTimeoutUnit.toMillis(this.connectTimeout)),
                 Math.toIntExact(readTimeoutUnit.toMillis(this.readTimeout)),
-                Math.toIntExact(requestTimeoutUnit.toMillis(this.requestTimeout))).getHttpClient();
+                Math.toIntExact(requestTimeoutUnit.toMillis(this.requestTimeout)));
 
         long readTimeoutMs = readTimeoutUnit.toMillis(this.readTimeout);
         this.clusters = new ClustersImpl(root, auth, readTimeoutMs);
@@ -201,9 +201,9 @@ public PulsarAdmin(String serviceUrl,
         this.nonPersistentTopics = new NonPersistentTopicsImpl(root, auth, readTimeoutMs);
         this.resourceQuotas = new ResourceQuotasImpl(root, auth, readTimeoutMs);
         this.lookups = new LookupImpl(root, auth, useTls, readTimeoutMs);
-        this.functions = new FunctionsImpl(root, auth, httpAsyncClient, readTimeoutMs);
-        this.sources = new SourcesImpl(root, auth, httpAsyncClient, readTimeoutMs);
-        this.sinks = new SinksImpl(root, auth, httpAsyncClient, readTimeoutMs);
+        this.functions = new FunctionsImpl(root, auth, asyncHttpConnector.getHttpClient(), readTimeoutMs);
+        this.sources = new SourcesImpl(root, auth, asyncHttpConnector.getHttpClient(), readTimeoutMs);
+        this.sinks = new SinksImpl(root, auth, asyncHttpConnector.getHttpClient(), readTimeoutMs);
         this.worker = new WorkerImpl(root, auth, readTimeoutMs);
         this.schemas = new SchemasImpl(root, auth, readTimeoutMs);
         this.bookies = new BookiesImpl(root, auth, readTimeoutMs);
@@ -433,10 +433,6 @@ public void close() {
         }
         client.close();
 
-        try {
-            httpAsyncClient.close();
-        } catch (IOException e) {
-           LOG.error("Failed to close http async client", e);
-        }
+        asyncHttpConnector.close();
     }
 }
diff --git a/...-admin/src/main/java/org/apache/pulsar/client/admin/internal/http/AsyncHttpConnector.java b/...-admin/src/main/java/org/apache/pulsar/client/admin/internal/http/AsyncHttpConnector.java
@@ -310,6 +310,7 @@ public String getName() {
     public void close() {
         try {
             httpClient.close();
+            delayer.shutdownNow();
         } catch (IOException e) {
             log.warn("Failed to close http client", e);
         }