[improve][build] Upgrade async-http-client to 3.0.10

build-logic/conventions/src/main/kotlin/pulsar.java-conventions.gradle.kts

@@ -75,6 +75,35 @@ dependencies {
                 }
             }
         }
+        // com.sun.activation:jakarta.activation bundles the jakarta.activation API classes together
+        // with the implementation and was not republished past 2.0.x, so it conflicts with
+        // jakarta.activation:jakarta.activation-api 2.1.x. Replace it everywhere with the API artifact
+        // plus the Eclipse Angus implementation (the EE10 successor). async-http-client still depends
+        // on it (https://github.com/AsyncHttpClient/async-http-client/issues/2190) and this rule also
+        // guards against any future dependency pulling it in. Versions are pinned by the
+        // pulsar-dependencies platform.
+        all {
+            allVariants {
+                withDependencies {
+                    if (removeAll { it.group == "com.sun.activation" && it.name == "jakarta.activation" }) {
+                        add("jakarta.activation:jakarta.activation-api")
+                        add("org.eclipse.angus:angus-activation")
+                    }
+                }
+            }
+        }
+        // async-http-client depends on the classic io.netty:netty-codec module, which in Netty 4.2
+        // is an empty backwards-compatibility aggregator that only adds the unused
+        // netty-codec-marshalling and netty-codec-protobuf modules to the classpath. The codec
+        // modules async-http-client actually needs (netty-codec-base, netty-codec-compression)
+        // come in through its netty-codec-http dependency.
+        withModule("org.asynchttpclient:async-http-client") {
+            allVariants {
+                withDependencies {
+                    removeAll { it.group == "io.netty" && it.name == "netty-codec" }
+                }
+            }
+        }
         // libthrift is a transitive dependency of distributedlog-core.
         // libthrift 0.23.0 upgraded to jakarta.* and HttpComponents 5 deps for its HTTP/servlet
         // transports, which distributedlog-core does not use (only TJSON/TMemory serialization is needed).

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

@@ -389,8 +389,7 @@ The Apache Software License, Version 2.0
  * AirCompressor
     - io.airlift-aircompressor-2.0.3.jar
  * AsyncHttpClient
-    - org.asynchttpclient-async-http-client-2.15.0.jar
-    - org.asynchttpclient-async-http-client-netty-utils-2.15.0.jar
+    - org.asynchttpclient-async-http-client-3.0.10.jar
  * Jetty
     - org.eclipse.jetty-jetty-alpn-client-12.1.10.jar
     - org.eclipse.jetty-jetty-alpn-conscrypt-server-12.1.10.jar
@@ -441,7 +440,8 @@ The Apache Software License, Version 2.0
  * Javassist -- org.javassist-javassist-3.25.0-GA.jar
  * Kotlin Standard Lib
      - org.jetbrains.kotlin-kotlin-stdlib-2.2.21.jar
-     - org.jetbrains-annotations-13.0.jar
+ * JetBrains Annotations
+     - org.jetbrains-annotations-26.1.0.jar
  * gRPC
     - io.grpc-grpc-context-1.79.0.jar
     - io.grpc-grpc-core-1.79.0.jar
@@ -606,7 +606,6 @@ Eclipse Distribution License 1.0 -- ../licenses/LICENSE-EDL-1.0.txt
  * Jakarta Activation
     - jakarta.activation-jakarta.activation-api-2.1.3.jar
     - org.eclipse.angus-angus-activation-2.0.2.jar
-    - com.sun.activation-jakarta.activation-1.2.2.jar
  * Jakarta XML Binding -- jakarta.xml.bind-jakarta.xml.bind-api-4.0.2.jar
 
 Eclipse Public License - v2.0 -- ../licenses/LICENSE-EPL-2.0.txt

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

@@ -350,6 +350,7 @@ The Apache Software License, Version 2.0
     - netty-codec-compression-4.2.15.Final.jar
     - netty-codec-dns-4.2.15.Final.jar
     - netty-codec-http-4.2.15.Final.jar
+    - netty-codec-http2-4.2.15.Final.jar
     - netty-codec-socks-4.2.15.Final.jar
     - netty-codec-haproxy-4.2.15.Final.jar
     - netty-common-4.2.15.Final.jar
@@ -403,8 +404,7 @@ The Apache Software License, Version 2.0
   * AirCompressor
      - aircompressor-2.0.3.jar
  * AsyncHttpClient
-    - async-http-client-2.15.0.jar
-    - async-http-client-netty-utils-2.15.0.jar
+    - async-http-client-3.0.10.jar
  * Jetty
     - jetty-alpn-client-12.1.10.jar
     - jetty-client-12.1.10.jar
@@ -428,6 +428,7 @@ The Apache Software License, Version 2.0
  * Spotify completable-futures -- completable-futures-0.3.6.jar
  * RoaringBitmap -- RoaringBitmap-1.6.9.jar
  * JSpecify -- jspecify-1.0.0.jar
+ * JetBrains Annotations -- annotations-26.1.0.jar
 
 BSD 3-clause "New" or "Revised" License
  * JLine3 -- jline-3.21.0.jar -- ../licenses/LICENSE-JLine.txt
@@ -457,7 +458,6 @@ Eclipse Distribution License 1.0 -- ../licenses/LICENSE-EDL-1.0.txt
  * Jakarta Activation
     - jakarta.activation-api-2.1.3.jar
     - angus-activation-2.0.2.jar
-    - jakarta.activation-1.2.2.jar
     - validation-api-1.1.0.Final.jar
  * Jakarta XML Binding -- jakarta.xml.bind-api-4.0.2.jar
 

gradle/libs.versions.toml

@@ -66,13 +66,15 @@ snakeyaml = "2.0"
 # Vert.x
 vertx = "4.5.28"
 # Networking / HTTP
-asynchttpclient = "2.15.0"
+asynchttpclient = "3.0.10"
 conscrypt = "2.5.2"
 okhttp3 = "5.3.2"
 okio = "3.17.0"
 netty-tcnative = "2.0.77.Final"
 httpcomponents-httpclient = "4.5.14"
 httpcomponents-httpcore = "4.4.16"
+# Pinned so that async-http-client's 1.0.3 dependency does not downgrade it
+reactive-streams = "1.0.4"
 # Google libraries (transitive deps)
 opentelemetry-gcp-resources = "1.57.0-alpha"
 # Data structures / Utils
@@ -354,6 +356,7 @@ okio-bom = { module = "com.squareup.okio:okio-bom", version.ref = "okio" }
 okio = { module = "com.squareup.okio:okio", version.ref = "okio" }
 # Transitive dep version pins (enforced by pulsar-dependencies platform)
 httpcomponents-httpclient = { module = "org.apache.httpcomponents:httpclient", version.ref = "httpcomponents-httpclient" }
+reactive-streams = { module = "org.reactivestreams:reactive-streams", version.ref = "reactive-streams" }
 httpcomponents-httpcore = { module = "org.apache.httpcomponents:httpcore", version.ref = "httpcomponents-httpcore" }
 jakarta-annotation-api = { module = "jakarta.annotation:jakarta.annotation-api", version.ref = "jakarta-annotation" }
 snakeyaml = { module = "org.yaml:snakeyaml", version.ref = "snakeyaml" }

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;
@@ -168,9 +169,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
@@ -183,8 +184,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

@@ -212,7 +212,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) {
@@ -225,14 +225,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
@@ -243,7 +243,8 @@ public boolean keepAlive(InetSocketAddress remoteAddress, Request ahcRequest,
                         && super.keepAlive(remoteAddress, ahcRequest, request, response);
             }
         });
-        confBuilder.setDisableHttpsEndpointIdentificationAlgorithm(!conf.isTlsHostnameVerificationEnable());
+        confBuilder.setSslEngineFactory(
+                new PulsarHttpAsyncSslEngineFactory(sslFactory, null, conf.isTlsHostnameVerificationEnable()));
         configureSocks5ProxyIfNeeded(confBuilder, conf);
     }
 
@@ -270,7 +271,8 @@ private void configureAsyncHttpClientSslEngineFactory(ClientConfigurationData co
         }
         String hostname = conf.isTlsHostnameVerificationEnable() ? null : serviceNameResolver
                 .resolveHostUri().getHost();
-        SslEngineFactory sslEngineFactory = new PulsarHttpAsyncSslEngineFactory(sslFactory, hostname);
+        SslEngineFactory sslEngineFactory =
+                new PulsarHttpAsyncSslEngineFactory(sslFactory, hostname, conf.isTlsHostnameVerificationEnable());
         confBuilder.setSslEngineFactory(sslEngineFactory);
         confBuilder.setUseInsecureTrustManager(conf.isTlsAllowInsecureConnection());
         confBuilder.setDisableHttpsEndpointIdentificationAlgorithm(!conf.isTlsHostnameVerificationEnable());

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;
@@ -100,8 +101,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

@@ -31,6 +31,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;
@@ -101,8 +102,8 @@ protected HttpClient(ClientConfigurationData conf, EventLoopGroup eventLoopGroup
         // host/port, i.e. cross-origin.
         confBuilder.setFollowRedirect(false);
         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()))
@@ -136,7 +137,9 @@ public boolean keepAlive(InetSocketAddress remoteAddress, Request ahcRequest,
                 }
                 String hostname = conf.isTlsHostnameVerificationEnable() ? null : serviceNameResolver
                         .resolveHostUri().getHost();
-                SslEngineFactory sslEngineFactory = new PulsarHttpAsyncSslEngineFactory(this.sslFactory, hostname);
+                SslEngineFactory sslEngineFactory =
+                        new PulsarHttpAsyncSslEngineFactory(this.sslFactory, hostname,
+                                conf.isTlsHostnameVerificationEnable());
                 confBuilder.setSslEngineFactory(sslEngineFactory);
 
 

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

@@ -101,11 +101,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()));
         boolean hasCertFile = StringUtils.isNotBlank(certFile);
         boolean hasKeyFile = StringUtils.isNotBlank(keyFile);
@@ -126,7 +124,7 @@ private AsyncHttpClient defaultHttpClient(Duration readTimeout, Duration connect
                 sslFactory = new org.apache.pulsar.common.util.DefaultPulsarSslFactory();
                 sslFactory.initialize(sslConfiguration);
                 sslFactory.createInternalSslContext();
-                SslEngineFactory sslEngineFactory = new PulsarHttpAsyncSslEngineFactory(sslFactory, null);
+                SslEngineFactory sslEngineFactory = new PulsarHttpAsyncSslEngineFactory(sslFactory, null, true);
                 confBuilder.setSslEngineFactory(sslEngineFactory);
             } catch (Exception e) {
                 throw new IllegalArgumentException("Invalid TLS client certificate configuration", e);
@@ -174,10 +172,6 @@ private void refreshSslContext() {
         }
     }
 
-    private int getParameterDurationToMillis(String name, Duration value, Duration defaultValue) {
-        return (int) getParameterDuration(name, value, defaultValue).toMillis();
-    }
-
     private long getParameterDurationToSeconds(String name, Duration value, Duration defaultValue) {
         return getParameterDuration(name, value, defaultValue).getSeconds();
     }

pulsar-client/src/main/java/org/apache/pulsar/client/util/PulsarHttpAsyncSslEngineFactory.java

@@ -32,20 +32,28 @@ public class PulsarHttpAsyncSslEngineFactory extends DefaultSslEngineFactory {
 
     private final PulsarSslFactory pulsarSslFactory;
     private final String host;
+    private final boolean enableHostnameVerification;
 
-    public PulsarHttpAsyncSslEngineFactory(PulsarSslFactory pulsarSslFactory, String host) {
+    public PulsarHttpAsyncSslEngineFactory(PulsarSslFactory pulsarSslFactory, String host,
+                                           boolean enableHostnameVerification) {
         this.pulsarSslFactory = pulsarSslFactory;
         this.host = host;
+        this.enableHostnameVerification = enableHostnameVerification;
     }
 
     @Override
     protected void configureSslEngine(SSLEngine sslEngine, AsyncHttpClientConfig config) {
         super.configureSslEngine(sslEngine, config);
+        SSLParameters parameters = sslEngine.getSSLParameters();
         if (StringUtils.isNotBlank(host)) {
-            SSLParameters parameters = sslEngine.getSSLParameters();
             parameters.setServerNames(Collections.singletonList(new SNIHostName(host)));
-            sslEngine.setSSLParameters(parameters);
         }
+
+        if (enableHostnameVerification) {
+            parameters.setEndpointIdentificationAlgorithm("HTTPS");
+        }
+
+        sslEngine.setSSLParameters(parameters);
     }
 
     @Override