[improve][client] Implement HTTP client using javax.ws.rs

Signed-off-by: Zixuan Liu <[email protected]>
apache · Jan 28, 2025 · d69ab83 · d69ab83
1 parent 52e8730
commit d69ab83
Show file tree

Hide file tree

Showing 13 changed files with 172 additions and 205 deletions.
diff --git a/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/FunctionsImpl.java b/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/FunctionsImpl.java
@@ -40,8 +40,8 @@
 import org.apache.commons.lang3.StringUtils;
 import org.apache.pulsar.client.admin.Functions;
 import org.apache.pulsar.client.admin.PulsarAdminException;
-import org.apache.pulsar.client.admin.internal.http.AsyncHttpRequestExecutor;
 import org.apache.pulsar.client.api.Authentication;
+import org.apache.pulsar.client.internal.http.AsyncHttpRequestExecutor;
 import org.apache.pulsar.common.functions.FunctionConfig;
 import org.apache.pulsar.common.functions.FunctionDefinition;
 import org.apache.pulsar.common.functions.FunctionState;

diff --git a/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/PackagesImpl.java b/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/PackagesImpl.java
@@ -35,8 +35,8 @@
 import javax.ws.rs.core.Response;
 import org.apache.pulsar.client.admin.Packages;
 import org.apache.pulsar.client.admin.PulsarAdminException;
-import org.apache.pulsar.client.admin.internal.http.AsyncHttpRequestExecutor;
 import org.apache.pulsar.client.api.Authentication;
+import org.apache.pulsar.client.internal.http.AsyncHttpRequestExecutor;
 import org.apache.pulsar.common.naming.NamespaceName;
 import org.apache.pulsar.packages.management.core.common.PackageMetadata;
 import org.apache.pulsar.packages.management.core.common.PackageName;

diff --git a/...r-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/PulsarAdminImpl.java b/...r-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/PulsarAdminImpl.java
@@ -51,13 +51,14 @@
 import org.apache.pulsar.client.admin.Topics;
 import org.apache.pulsar.client.admin.Transactions;
 import org.apache.pulsar.client.admin.Worker;
-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.PulsarServiceNameResolver;
 import org.apache.pulsar.client.impl.auth.AuthenticationDisabled;
 import org.apache.pulsar.client.impl.conf.ClientConfigurationData;
+import org.apache.pulsar.client.internal.http.AsyncHttpConnector;
+import org.apache.pulsar.client.internal.http.AsyncHttpConnectorProvider;
 import org.apache.pulsar.common.net.ServiceURI;
 import org.glassfish.jersey.client.ClientConfig;
 import org.glassfish.jersey.client.ClientProperties;
@@ -124,8 +125,10 @@ public PulsarAdminImpl(String serviceUrl, ClientConfigurationData clientConfigDa
             clientConfigData.setServiceUrl(serviceUrl);
         }
 
+        PulsarServiceNameResolver pulsarServiceNameResolver = new PulsarServiceNameResolver();
+        pulsarServiceNameResolver.updateServiceUrl(serviceUrl);
         AsyncHttpConnectorProvider asyncConnectorProvider = new AsyncHttpConnectorProvider(clientConfigData,
-                clientConfigData.getAutoCertRefreshSeconds(), acceptGzipCompression);
+                clientConfigData.getAutoCertRefreshSeconds(), acceptGzipCompression, pulsarServiceNameResolver);
 
         ClientConfig httpConfig = new ClientConfig();
         httpConfig.property(ClientProperties.FOLLOW_REDIRECTS, true);

diff --git a/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/SinksImpl.java b/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/SinksImpl.java
@@ -34,8 +34,8 @@
 import org.apache.pulsar.client.admin.PulsarAdminException;
 import org.apache.pulsar.client.admin.Sink;
 import org.apache.pulsar.client.admin.Sinks;
-import org.apache.pulsar.client.admin.internal.http.AsyncHttpRequestExecutor;
 import org.apache.pulsar.client.api.Authentication;
+import org.apache.pulsar.client.internal.http.AsyncHttpRequestExecutor;
 import org.apache.pulsar.common.functions.UpdateOptions;
 import org.apache.pulsar.common.functions.UpdateOptionsImpl;
 import org.apache.pulsar.common.io.ConnectorDefinition;

diff --git a/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/SourcesImpl.java b/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/SourcesImpl.java
@@ -33,8 +33,8 @@
 import org.apache.pulsar.client.admin.PulsarAdminException;
 import org.apache.pulsar.client.admin.Source;
 import org.apache.pulsar.client.admin.Sources;
-import org.apache.pulsar.client.admin.internal.http.AsyncHttpRequestExecutor;
 import org.apache.pulsar.client.api.Authentication;
+import org.apache.pulsar.client.internal.http.AsyncHttpRequestExecutor;
 import org.apache.pulsar.common.functions.UpdateOptions;
 import org.apache.pulsar.common.functions.UpdateOptionsImpl;
 import org.apache.pulsar.common.io.ConnectorDefinition;

diff --git a/...internal/http/AsyncHttpConnectorTest.java → ...internal/http/AsyncHttpConnectorTest.java b/...internal/http/AsyncHttpConnectorTest.java → ...internal/http/AsyncHttpConnectorTest.java
@@ -16,7 +16,7 @@
  * specific language governing permissions and limitations
  * under the License.
  */
-package org.apache.pulsar.client.admin.internal.http;
+package org.apache.pulsar.client.internal.http;
 
 import static com.github.tomakehurst.wiremock.client.WireMock.aResponse;
 import static com.github.tomakehurst.wiremock.client.WireMock.get;
@@ -44,6 +44,8 @@
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicInteger;
 import lombok.Cleanup;
+import org.apache.pulsar.client.api.PulsarClientException.InvalidServiceURL;
+import org.apache.pulsar.client.impl.PulsarServiceNameResolver;
 import org.apache.pulsar.client.impl.conf.ClientConfigurationData;
 import org.apache.pulsar.common.util.FutureUtil;
 import org.asynchttpclient.Request;
@@ -174,9 +176,11 @@ public void testShouldStopRetriesWhenTimeoutOccurs() throws IOException, Executi
         Executor delayedExecutor = runnable -> {
             scheduledExecutor.schedule(runnable, requestTimeout, TimeUnit.MILLISECONDS);
         };
+        PulsarServiceNameResolver pulsarServiceNameResolver = new PulsarServiceNameResolver();
+        pulsarServiceNameResolver.updateServiceUrl(conf.getServiceUrl());
         @Cleanup
         AsyncHttpConnector connector = new AsyncHttpConnector(5000, requestTimeout,
-                requestTimeout, 0, conf, false) {
+                requestTimeout, 0, conf, false, pulsarServiceNameResolver) {
             @Override
             protected CompletableFuture<Response> oneShot(InetSocketAddress host, ClientRequest request) {
                 // delay the response to simulate a timeout
@@ -214,7 +218,7 @@ public void failure(Throwable failure) {
     }
 
     @Test
-    void testMaxRedirects() {
+    void testMaxRedirects() throws InvalidServiceURL {
         // Redirect to itself to test max redirects
         server.stubFor(get(urlEqualTo("/admin/v2/clusters"))
                 .willReturn(aResponse()
@@ -224,9 +228,11 @@ void testMaxRedirects() {
         ClientConfigurationData conf = new ClientConfigurationData();
         conf.setServiceUrl("http://localhost:" + server.port());
 
+        PulsarServiceNameResolver pulsarServiceNameResolver = new PulsarServiceNameResolver();
+        pulsarServiceNameResolver.updateServiceUrl(conf.getServiceUrl());
         @Cleanup
         AsyncHttpConnector connector = new AsyncHttpConnector(5000, 5000,
-                5000, 0, conf, false);
+                5000, 0, conf, false, pulsarServiceNameResolver);
 
         Request request = new RequestBuilder("GET")
                 .setUrl("http://localhost:" + server.port() + "/admin/v2/clusters")
@@ -243,21 +249,21 @@ void testMaxRedirects() {
     }
 
     @Test
-    void testRelativeRedirect() throws ExecutionException, InterruptedException {
+    void testRelativeRedirect() throws ExecutionException, InterruptedException, InvalidServiceURL {
         doTestRedirect("path2");
     }
 
     @Test
-    void testAbsoluteRedirect() throws ExecutionException, InterruptedException {
+    void testAbsoluteRedirect() throws ExecutionException, InterruptedException, InvalidServiceURL {
         doTestRedirect("/path2");
     }
 
     @Test
-    void testUrlRedirect() throws ExecutionException, InterruptedException {
+    void testUrlRedirect() throws ExecutionException, InterruptedException, InvalidServiceURL {
         doTestRedirect("http://localhost:" + server.port() + "/path2");
     }
 
-    private void doTestRedirect(String location) throws InterruptedException, ExecutionException {
+    private void doTestRedirect(String location) throws InterruptedException, ExecutionException, InvalidServiceURL {
         server.stubFor(get(urlEqualTo("/path1"))
                 .willReturn(aResponse()
                         .withStatus(301)
@@ -270,9 +276,11 @@ private void doTestRedirect(String location) throws InterruptedException, Execut
         ClientConfigurationData conf = new ClientConfigurationData();
         conf.setServiceUrl("http://localhost:" + server.port());
 
+        PulsarServiceNameResolver pulsarServiceNameResolver = new PulsarServiceNameResolver();
+        pulsarServiceNameResolver.updateServiceUrl(conf.getServiceUrl());
         @Cleanup
         AsyncHttpConnector connector = new AsyncHttpConnector(5000, 5000,
-                5000, 0, conf, false);
+                5000, 0, conf, false, pulsarServiceNameResolver);
 
         Request request = new RequestBuilder("GET")
                 .setUrl("http://localhost:" + server.port() + "/path1")
@@ -283,7 +291,7 @@ private void doTestRedirect(String location) throws InterruptedException, Execut
     }
 
     @Test
-    void testRedirectWithBody() throws ExecutionException, InterruptedException {
+    void testRedirectWithBody() throws ExecutionException, InterruptedException, InvalidServiceURL {
         server.stubFor(post(urlEqualTo("/path1"))
                 .willReturn(aResponse()
                         .withStatus(307)
@@ -296,9 +304,12 @@ void testRedirectWithBody() throws ExecutionException, InterruptedException {
         ClientConfigurationData conf = new ClientConfigurationData();
         conf.setServiceUrl("http://localhost:" + server.port());
 
+        PulsarServiceNameResolver pulsarServiceNameResolver = new PulsarServiceNameResolver();
+        pulsarServiceNameResolver.updateServiceUrl(conf.getServiceUrl());
         @Cleanup
         AsyncHttpConnector connector = new AsyncHttpConnector(5000, 5000,
-                5000, 0, conf, false);
+                5000, 0, conf, false, pulsarServiceNameResolver);
+
 
         Request request = new RequestBuilder("POST")
                 .setUrl("http://localhost:" + server.port() + "/path1")
@@ -310,7 +321,7 @@ void testRedirectWithBody() throws ExecutionException, InterruptedException {
     }
 
     @Test
-    void testMaxConnections() throws ExecutionException, InterruptedException {
+    void testMaxConnections() throws ExecutionException, InterruptedException, InvalidServiceURL {
         server.stubFor(post(urlEqualTo("/concurrency-test"))
                 .willReturn(aResponse()
                         .withTransformers("concurrency-test")));
@@ -320,9 +331,11 @@ void testMaxConnections() throws ExecutionException, InterruptedException {
         conf.setConnectionsPerBroker(maxConnections);
         conf.setServiceUrl("http://localhost:" + server.port());
 
+        PulsarServiceNameResolver pulsarServiceNameResolver = new PulsarServiceNameResolver();
+        pulsarServiceNameResolver.updateServiceUrl(conf.getServiceUrl());
         @Cleanup
         AsyncHttpConnector connector = new AsyncHttpConnector(5000, 5000,
-                5000, 0, conf, false);
+                5000, 0, conf, false, pulsarServiceNameResolver);
 
         Request request = new RequestBuilder("POST")
                 .setUrl("http://localhost:" + server.port() + "/concurrency-test")

diff --git a/pulsar-client/pom.xml b/pulsar-client/pom.xml
@@ -186,6 +186,21 @@
       <optional>true</optional>
     </dependency>
 
+    <dependency>
+      <groupId>jakarta.ws.rs</groupId>
+      <artifactId>jakarta.ws.rs-api</artifactId>
+    </dependency>
+
+    <dependency>
+      <groupId>org.glassfish.jersey.core</groupId>
+      <artifactId>jersey-client</artifactId>
+    </dependency>
+
+    <dependency>
+      <groupId>org.glassfish.jersey.media</groupId>
+      <artifactId>jersey-media-multipart</artifactId>
+    </dependency>
+
     <!-- Testing dependencies -->
     <dependency>
       <groupId>${project.groupId}</groupId>
@@ -217,6 +232,13 @@
       <artifactId>fastutil</artifactId>
     </dependency>
 
+    <dependency>
+      <groupId>com.github.tomakehurst</groupId>
+      <artifactId>wiremock-jre8</artifactId>
+      <version>${wiremock.version}</version>
+      <scope>test</scope>
+    </dependency>
+
   </dependencies>
 
   <build>