[fix] Upgrade to AsyncHTTP client 3.0.4

distribution/server/src/assemble/LICENSE.bin.txt

@@ -391,8 +391,7 @@ The Apache Software License, Version 2.0
  * AirCompressor
     - io.airlift-aircompressor-0.27.jar
  * AsyncHttpClient
-    - org.asynchttpclient-async-http-client-2.12.4.jar
-    - org.asynchttpclient-async-http-client-netty-utils-2.12.4.jar
+    - org.asynchttpclient-async-http-client-3.0.4.jar
  * Jetty
     - org.eclipse.jetty-jetty-client-9.4.58.v20250814.jar
     - org.eclipse.jetty-jetty-continuation-9.4.58.v20250814.jar
@@ -428,7 +427,7 @@ The Apache Software License, Version 2.0
  * Kotlin Standard Lib
      - org.jetbrains.kotlin-kotlin-stdlib-1.8.20.jar
      - org.jetbrains.kotlin-kotlin-stdlib-common-1.8.20.jar
-     - org.jetbrains-annotations-13.0.jar
+     - org.jetbrains-annotations-26.0.2.jar
  * gRPC
     - io.grpc-grpc-all-1.75.0.jar
     - io.grpc-grpc-auth-1.75.0.jar

distribution/shell/src/assemble/LICENSE.bin.txt

@@ -401,8 +401,7 @@ The Apache Software License, Version 2.0
   * AirCompressor
      - aircompressor-0.27.jar
  * AsyncHttpClient
-    - async-http-client-2.12.4.jar
-    - async-http-client-netty-utils-2.12.4.jar
+    - async-http-client-3.0.4.jar
  * Jetty
     - jetty-client-9.4.58.v20250814.jar
     - jetty-http-9.4.58.v20250814.jar

pom.xml

@@ -261,7 +261,7 @@ flexible messaging model and an intuitive client API.</description>
     <prometheus-jmx.version>0.16.1</prometheus-jmx.version>
     <confluent.version>7.9.2</confluent.version>
     <aircompressor.version>0.27</aircompressor.version>
-    <asynchttpclient.version>2.12.4</asynchttpclient.version>
+    <asynchttpclient.version>3.0.4</asynchttpclient.version>
     <commons-lang3.version>3.19.0</commons-lang3.version>
     <commons-io.version>2.21.0</commons-io.version>
     <commons-codec.version>1.20.0</commons-codec.version>

pulsar-broker-auth-oidc/src/main/java/org/apache/pulsar/broker/authentication/oidc/AuthenticationProviderOpenID.java

@@ -47,6 +47,7 @@
 import java.security.PublicKey;
 import java.security.interfaces.ECPublicKey;
 import java.security.interfaces.RSAPublicKey;
+import java.time.Duration;
 import java.util.List;
 import java.util.Set;
 import java.util.concurrent.CompletableFuture;
@@ -169,9 +170,9 @@ public void initialize(Context context) throws IOException {
         this.issuers = validateIssuers(getConfigValueAsSet(config, ALLOWED_TOKEN_ISSUERS), requireHttps,
                 fallbackDiscoveryMode != FallbackDiscoveryMode.DISABLED);
 
-        int connectionTimeout = getConfigValueAsInt(config, HTTP_CONNECTION_TIMEOUT_MILLIS,
+        int connectionTimeoutMs = getConfigValueAsInt(config, HTTP_CONNECTION_TIMEOUT_MILLIS,
                 HTTP_CONNECTION_TIMEOUT_MILLIS_DEFAULT);
-        int readTimeout = getConfigValueAsInt(config, HTTP_READ_TIMEOUT_MILLIS, HTTP_READ_TIMEOUT_MILLIS_DEFAULT);
+        int readTimeoutMs = getConfigValueAsInt(config, HTTP_READ_TIMEOUT_MILLIS, HTTP_READ_TIMEOUT_MILLIS_DEFAULT);
         String trustCertsFilePath = getConfigValueAsString(config, ISSUER_TRUST_CERTS_FILE_PATH, null);
         SslContext sslContext = null;
         // When config is in the conf file but is empty, it defaults to the empty string, which is not meaningful and
@@ -184,8 +185,8 @@ public void initialize(Context context) throws IOException {
         }
         AsyncHttpClientConfig clientConfig = new DefaultAsyncHttpClientConfig.Builder()
                 .setCookieStore(null)
-                .setConnectTimeout(connectionTimeout)
-                .setReadTimeout(readTimeout)
+                .setConnectTimeout(Duration.ofMillis(connectionTimeoutMs))
+                .setReadTimeout(Duration.ofMillis(readTimeoutMs))
                 .setSslContext(sslContext)
                 .build();
         httpClient = new DefaultAsyncHttpClient(clientConfig);

pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/http/AsyncHttpConnector.java

@@ -203,7 +203,7 @@ private void configureAsyncHttpClientConfig(ClientConfigurationData conf, int co
             confBuilder.setAcquireFreeChannelTimeout(conf.getRequestTimeoutMs());
         }
         if (conf.getConnectionMaxIdleSeconds() > 0) {
-            confBuilder.setPooledConnectionIdleTimeout(conf.getConnectionMaxIdleSeconds() * 1000);
+            confBuilder.setPooledConnectionIdleTimeout(Duration.ofSeconds(conf.getConnectionMaxIdleSeconds()));
         }
         if (sharedResources != null) {
             if (this.eventLoopGroup != null) {
@@ -216,14 +216,14 @@ private void configureAsyncHttpClientConfig(ClientConfigurationData conf, int co
         confBuilder.setCookieStore(null);
         confBuilder.setUseProxyProperties(true);
         confBuilder.setFollowRedirect(false);
-        confBuilder.setRequestTimeout(conf.getRequestTimeoutMs());
-        confBuilder.setConnectTimeout(connectTimeoutMs);
-        confBuilder.setReadTimeout(readTimeoutMs);
+        confBuilder.setRequestTimeout(Duration.ofMillis(conf.getRequestTimeoutMs()));
+        confBuilder.setConnectTimeout(Duration.ofMillis(connectTimeoutMs));
+        confBuilder.setReadTimeout(Duration.ofMillis(readTimeoutMs));
         confBuilder.setUserAgent(String.format("Pulsar-Java-v%s%s",
                 PulsarVersion.getVersion(),
                 (conf.getDescription() == null ? "" : ("-" + conf.getDescription()))
         ));
-        confBuilder.setRequestTimeout(requestTimeoutMs);
+        confBuilder.setRequestTimeout(Duration.ofMillis(requestTimeoutMs));
         confBuilder.setIoThreadsCount(conf.getNumIoThreads());
         confBuilder.setKeepAliveStrategy(new DefaultKeepAliveStrategy() {
             @Override

pulsar-client/src/main/java/org/apache/pulsar/client/impl/ControlledClusterFailover.java

@@ -27,6 +27,7 @@
 import java.io.IOException;
 import java.net.InetSocketAddress;
 import java.nio.charset.StandardCharsets;
+import java.time.Duration;
 import java.util.Collections;
 import java.util.LinkedHashMap;
 import java.util.Map;
@@ -91,8 +92,8 @@ private AsyncHttpClient buildHttpClient() {
         confBuilder.setUseProxyProperties(true);
         confBuilder.setFollowRedirect(true);
         confBuilder.setMaxRedirects(DEFAULT_MAX_REDIRECTS);
-        confBuilder.setConnectTimeout(DEFAULT_CONNECT_TIMEOUT_IN_SECONDS * 1000);
-        confBuilder.setReadTimeout(DEFAULT_READ_TIMEOUT_IN_SECONDS * 1000);
+        confBuilder.setConnectTimeout(Duration.ofSeconds(DEFAULT_CONNECT_TIMEOUT_IN_SECONDS));
+        confBuilder.setReadTimeout(Duration.ofSeconds(DEFAULT_READ_TIMEOUT_IN_SECONDS));
         confBuilder.setUserAgent(String.format("Pulsar-Java-v%s", PulsarVersion.getVersion()));
         confBuilder.setKeepAliveStrategy(new DefaultKeepAliveStrategy() {
             @Override

pulsar-client/src/main/java/org/apache/pulsar/client/impl/HttpClient.java

@@ -30,6 +30,7 @@
 import java.net.InetSocketAddress;
 import java.net.URI;
 import java.net.URL;
+import java.time.Duration;
 import java.util.Map;
 import java.util.Map.Entry;
 import java.util.Set;
@@ -89,8 +90,8 @@ protected HttpClient(ClientConfigurationData conf, EventLoopGroup eventLoopGroup
         confBuilder.setUseProxyProperties(true);
         confBuilder.setFollowRedirect(true);
         confBuilder.setMaxRedirects(conf.getMaxLookupRedirects());
-        confBuilder.setConnectTimeout(DEFAULT_CONNECT_TIMEOUT_IN_SECONDS * 1000);
-        confBuilder.setReadTimeout(DEFAULT_READ_TIMEOUT_IN_SECONDS * 1000);
+        confBuilder.setConnectTimeout(Duration.ofSeconds(DEFAULT_CONNECT_TIMEOUT_IN_SECONDS));
+        confBuilder.setReadTimeout(Duration.ofSeconds(DEFAULT_READ_TIMEOUT_IN_SECONDS));
         confBuilder.setUserAgent(String.format("Pulsar-Java-v%s%s",
                 PulsarVersion.getVersion(),
                 (conf.getDescription() == null ? "" : ("-" + conf.getDescription()))

pulsar-client/src/main/java/org/apache/pulsar/client/impl/auth/oauth2/FlowBase.java

@@ -69,11 +69,9 @@ private AsyncHttpClient defaultHttpClient(Duration readTimeout, Duration connect
         confBuilder.setCookieStore(null);
         confBuilder.setUseProxyProperties(true);
         confBuilder.setFollowRedirect(true);
-        confBuilder.setConnectTimeout(
-                getParameterDurationToMillis(CONFIG_PARAM_CONNECT_TIMEOUT, connectTimeout,
+        confBuilder.setConnectTimeout(getParameterDuration(CONFIG_PARAM_CONNECT_TIMEOUT, connectTimeout,
                         DEFAULT_CONNECT_TIMEOUT));
-        confBuilder.setReadTimeout(
-                getParameterDurationToMillis(CONFIG_PARAM_READ_TIMEOUT, readTimeout, DEFAULT_READ_TIMEOUT));
+        confBuilder.setReadTimeout(getParameterDuration(CONFIG_PARAM_READ_TIMEOUT, readTimeout, DEFAULT_READ_TIMEOUT));
         confBuilder.setUserAgent(String.format("Pulsar-Java-v%s", PulsarVersion.getVersion()));
         if (StringUtils.isNotBlank(trustCertsFilePath)) {
             try {
@@ -87,17 +85,14 @@ private AsyncHttpClient defaultHttpClient(Duration readTimeout, Duration connect
         return new DefaultAsyncHttpClient(confBuilder.build());
     }
 
-    private int getParameterDurationToMillis(String name, Duration value, Duration defaultValue) {
-        Duration duration;
+    private Duration getParameterDuration(String name, Duration value, Duration defaultValue) {
         if (value == null) {
             log.info("Configuration for [{}] is using the default value: [{}]", name, defaultValue);
-            duration = defaultValue;
+            return defaultValue;
         } else {
             log.info("Configuration for [{}] is: [{}]", name, value);
-            duration = value;
+            return value;
         }
-
-        return (int) duration.toMillis();
     }
 
     public void initialize() throws PulsarClientException {