You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by lh...@apache.org on 2026/10/20 22:21:22 UTC
(pulsar) 01/21: [fix][admin] Use ClientConfigurationData timeout params to set AsyncHttpConnector timeout params (#25040)
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a commit to branch branch-4.0
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit b384c00be9f94789abf89262a227f6edd983d349
Author: Oneby Wang <89...@qq.com>
AuthorDate: Sat Oct 17 20:51:21 2026 +0800
[fix][admin] Use ClientConfigurationData timeout params to set AsyncHttpConnector timeout params (#25040)
Co-authored-by: oneby-wang <on...@qq.com>
Co-authored-by: Matteo Merli <mm...@apache.org>
(cherry picked from commit f78fa57251ab02386910059ec1c8065bf015e373)
---
.../admin/internal/http/AsyncHttpConnector.java | 12 ++------
.../internal/http/AsyncHttpConnectorProvider.java | 2 +-
.../client/admin/internal/PulsarAdminImplTest.java | 35 ++++++++++++++++++++++
3 files changed, 39 insertions(+), 10 deletions(-)
diff --git a/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/http/AsyncHttpConnector.java b/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/http/AsyncHttpConnector.java
index 1ea921f58a6a..ca173a71ebf4 100644
--- a/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/http/AsyncHttpConnector.java
+++ b/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/http/AsyncHttpConnector.java
@@ -52,7 +52,6 @@ import java.util.concurrent.TimeoutException;
import java.util.function.BooleanSupplier;
import java.util.function.Supplier;
import javax.ws.rs.ProcessingException;
-import javax.ws.rs.client.Client;
import javax.ws.rs.core.HttpHeaders;
import javax.ws.rs.core.Response.Status;
import lombok.Getter;
@@ -61,7 +60,6 @@ import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.Validate;
import org.apache.pulsar.PulsarVersion;
-import org.apache.pulsar.client.admin.internal.PulsarAdminImpl;
import org.apache.pulsar.client.api.PulsarClientException;
import org.apache.pulsar.client.impl.PulsarServiceNameResolver;
import org.apache.pulsar.client.impl.ServiceNameResolver;
@@ -83,7 +81,6 @@ import org.asynchttpclient.Response;
import org.asynchttpclient.SslEngineFactory;
import org.asynchttpclient.channel.DefaultKeepAliveStrategy;
import org.asynchttpclient.uri.Uri;
-import org.glassfish.jersey.client.ClientProperties;
import org.glassfish.jersey.client.ClientRequest;
import org.glassfish.jersey.client.ClientResponse;
import org.glassfish.jersey.client.spi.AsyncConnectorCallback;
@@ -112,13 +109,10 @@ public class AsyncHttpConnector implements Connector, AsyncHttpRequestExecutor {
@Setter
private boolean followRedirects = true;
- public AsyncHttpConnector(Client client, ClientConfigurationData conf, int autoCertRefreshTimeSeconds,
+ public AsyncHttpConnector(ClientConfigurationData conf, int autoCertRefreshTimeSeconds,
boolean acceptGzipCompression) {
- this((int) client.getConfiguration().getProperty(ClientProperties.CONNECT_TIMEOUT),
- (int) client.getConfiguration().getProperty(ClientProperties.READ_TIMEOUT),
- PulsarAdminImpl.DEFAULT_REQUEST_TIMEOUT_SECONDS * 1000,
- autoCertRefreshTimeSeconds,
- conf, acceptGzipCompression);
+ this(conf.getConnectionTimeoutMs(), conf.getReadTimeoutMs(), conf.getRequestTimeoutMs(),
+ autoCertRefreshTimeSeconds, conf, acceptGzipCompression);
}
@SneakyThrows
diff --git a/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/http/AsyncHttpConnectorProvider.java b/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/http/AsyncHttpConnectorProvider.java
index 0ceab24ac000..af7dea8847ec 100644
--- a/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/http/AsyncHttpConnectorProvider.java
+++ b/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/http/AsyncHttpConnectorProvider.java
@@ -46,7 +46,7 @@ public class AsyncHttpConnectorProvider implements ConnectorProvider {
@Override
public Connector getConnector(Client client, Configuration runtimeConfig) {
if (connector == null) {
- connector = new AsyncHttpConnector(client, conf, autoCertRefreshTimeSeconds, acceptGzipCompression);
+ connector = new AsyncHttpConnector(conf, autoCertRefreshTimeSeconds, acceptGzipCompression);
connector.setFollowRedirects(followRedirects);
}
return connector;
diff --git a/pulsar-client-admin/src/test/java/org/apache/pulsar/client/admin/internal/PulsarAdminImplTest.java b/pulsar-client-admin/src/test/java/org/apache/pulsar/client/admin/internal/PulsarAdminImplTest.java
index 27c6fd96079f..94949d58462c 100644
--- a/pulsar-client-admin/src/test/java/org/apache/pulsar/client/admin/internal/PulsarAdminImplTest.java
+++ b/pulsar-client-admin/src/test/java/org/apache/pulsar/client/admin/internal/PulsarAdminImplTest.java
@@ -18,8 +18,20 @@
*/
package org.apache.pulsar.client.admin.internal;
+import static com.github.tomakehurst.wiremock.client.WireMock.get;
+import static com.github.tomakehurst.wiremock.client.WireMock.okJson;
+import static com.github.tomakehurst.wiremock.client.WireMock.urlEqualTo;
import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
import static org.mockito.Mockito.mock;
+import com.github.tomakehurst.wiremock.WireMockServer;
+import com.github.tomakehurst.wiremock.core.WireMockConfiguration;
+import java.util.List;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+import lombok.Cleanup;
import lombok.SneakyThrows;
import org.apache.pulsar.client.api.Authentication;
import org.apache.pulsar.client.impl.auth.AuthenticationDisabled;
@@ -45,6 +57,29 @@ public class PulsarAdminImplTest {
assertThat(createAdminAndGetAuth(conf)).isSameAs(auth);
}
+ @Test
+ public void testAsyncRequestIsBoundedByRequestTimeout() throws Exception {
+ @Cleanup("stop")
+ WireMockServer server = new WireMockServer(WireMockConfiguration.wireMockConfig().port(0));
+ server.start();
+ server.stubFor(get(urlEqualTo("/admin/v2/clusters"))
+ .willReturn(okJson("[\"test-cluster\"]").withFixedDelay(5000)));
+
+ ClientConfigurationData conf = new ClientConfigurationData();
+ conf.setServiceUrl("http://localhost:" + server.port());
+ conf.setReadTimeoutMs(10000);
+ conf.setRequestTimeoutMs(1000);
+
+ @Cleanup
+ PulsarAdminImpl admin = new PulsarAdminImpl(conf.getServiceUrl(), conf, null);
+ CompletableFuture<List<String>> clusters = admin.clusters().getClustersAsync();
+ // The server answers after 5 s, within the read timeout: only the 1 s request timeout can fail the call
+ // before then
+ assertThatThrownBy(() -> clusters.get(4, TimeUnit.SECONDS))
+ .isInstanceOf(ExecutionException.class)
+ .hasRootCauseInstanceOf(TimeoutException.class);
+ }
+
@SneakyThrows
private Authentication createAdminAndGetAuth(ClientConfigurationData conf) {
try (PulsarAdminImpl admin = new PulsarAdminImpl("http://localhost:8080", conf, null)) {