diff --git a/distribution/server/src/assemble/LICENSE.bin.txt b/distribution/server/src/assemble/LICENSE.bin.txt index a0327d4d3e828..c212432c2be30 100644 --- a/distribution/server/src/assemble/LICENSE.bin.txt +++ b/distribution/server/src/assemble/LICENSE.bin.txt @@ -273,7 +273,7 @@ The Apache Software License, Version 2.0 - com.google.guava-failureaccess-1.0.1.jar - com.google.guava-listenablefuture-9999.0-empty-to-avoid-conflict-with-guava.jar * J2ObjC Annotations -- com.google.j2objc-j2objc-annotations-1.3.jar - * Netty Reactive Streams -- com.typesafe.netty-netty-reactive-streams-2.0.6.jar + * Netty Reactive Streams -- com.typesafe.netty-netty-reactive-streams-2.0.20.jar * Swagger - io.swagger-swagger-annotations-1.6.2.jar - io.swagger-swagger-core-1.6.2.jar @@ -294,26 +294,27 @@ The Apache Software License, Version 2.0 - org.apache.commons-commons-lang3-3.19.0.jar - org.apache.commons-commons-text-1.15.0.jar * Netty - - io.netty-netty-buffer-4.1.139.Final.jar - - io.netty-netty-codec-4.1.139.Final.jar - - io.netty-netty-codec-dns-4.1.139.Final.jar - - io.netty-netty-codec-http-4.1.139.Final.jar - - io.netty-netty-codec-http2-4.1.139.Final.jar - - io.netty-netty-codec-socks-4.1.139.Final.jar - - io.netty-netty-codec-haproxy-4.1.139.Final.jar - - io.netty-netty-common-4.1.139.Final.jar - - io.netty-netty-handler-4.1.139.Final.jar - - io.netty-netty-handler-proxy-4.1.139.Final.jar - - io.netty-netty-resolver-4.1.139.Final.jar - - io.netty-netty-resolver-dns-4.1.139.Final.jar - - io.netty-netty-resolver-dns-classes-macos-4.1.139.Final.jar - - io.netty-netty-resolver-dns-native-macos-4.1.139.Final-osx-aarch_64.jar - - io.netty-netty-resolver-dns-native-macos-4.1.139.Final-osx-x86_64.jar - - io.netty-netty-transport-4.1.139.Final.jar - - io.netty-netty-transport-classes-epoll-4.1.139.Final.jar - - io.netty-netty-transport-native-epoll-4.1.139.Final-linux-aarch_64.jar - - io.netty-netty-transport-native-epoll-4.1.139.Final-linux-x86_64.jar - - io.netty-netty-transport-native-unix-common-4.1.139.Final.jar + - io.netty-netty-buffer-4.2.19.Final.jar + - io.netty-netty-codec-base-4.2.19.Final.jar + - io.netty-netty-codec-compression-4.2.19.Final.jar + - io.netty-netty-codec-dns-4.2.19.Final.jar + - io.netty-netty-codec-http-4.2.19.Final.jar + - io.netty-netty-codec-http2-4.2.19.Final.jar + - io.netty-netty-codec-socks-4.2.19.Final.jar + - io.netty-netty-codec-haproxy-4.2.19.Final.jar + - io.netty-netty-common-4.2.19.Final.jar + - io.netty-netty-handler-4.2.19.Final.jar + - io.netty-netty-handler-proxy-4.2.19.Final.jar + - io.netty-netty-resolver-4.2.19.Final.jar + - io.netty-netty-resolver-dns-4.2.19.Final.jar + - io.netty-netty-resolver-dns-classes-macos-4.2.19.Final.jar + - io.netty-netty-resolver-dns-native-macos-4.2.19.Final-osx-aarch_64.jar + - io.netty-netty-resolver-dns-native-macos-4.2.19.Final-osx-x86_64.jar + - io.netty-netty-transport-4.2.19.Final.jar + - io.netty-netty-transport-classes-epoll-4.2.19.Final.jar + - io.netty-netty-transport-native-epoll-4.2.19.Final-linux-aarch_64.jar + - io.netty-netty-transport-native-epoll-4.2.19.Final-linux-x86_64.jar + - io.netty-netty-transport-native-unix-common-4.2.19.Final.jar - io.netty-netty-tcnative-boringssl-static-2.0.84.Final.jar - io.netty-netty-tcnative-boringssl-static-2.0.84.Final-linux-aarch_64.jar - io.netty-netty-tcnative-boringssl-static-2.0.84.Final-linux-x86_64.jar @@ -321,6 +322,9 @@ The Apache Software License, Version 2.0 - io.netty-netty-tcnative-boringssl-static-2.0.84.Final-osx-x86_64.jar - io.netty-netty-tcnative-boringssl-static-2.0.84.Final-windows-x86_64.jar - io.netty-netty-tcnative-classes-2.0.84.Final.jar + - io.netty-netty-transport-classes-io_uring-4.2.19.Final.jar + - io.netty-netty-transport-native-io_uring-4.2.19.Final-linux-x86_64.jar + - io.netty-netty-transport-native-io_uring-4.2.19.Final-linux-aarch_64.jar - io.netty.incubator-netty-incubator-transport-classes-io_uring-0.0.26.Final.jar - io.netty.incubator-netty-incubator-transport-native-io_uring-0.0.26.Final-linux-x86_64.jar - io.netty.incubator-netty-incubator-transport-native-io_uring-0.0.26.Final-linux-aarch_64.jar @@ -390,8 +394,7 @@ The Apache Software License, Version 2.0 * AirCompressor - io.airlift-aircompressor-2.0.3.jar * AsyncHttpClient - - org.asynchttpclient-async-http-client-2.16.1.jar - - org.asynchttpclient-async-http-client-netty-utils-2.16.1.jar + - org.asynchttpclient-async-http-client-3.0.14.jar * Jetty - org.eclipse.jetty-jetty-alpn-client-12.1.13.jar - org.eclipse.jetty-jetty-alpn-conscrypt-server-12.1.13.jar @@ -444,7 +447,7 @@ 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 + - org.jetbrains-annotations-26.1.0.jar * gRPC - io.grpc-grpc-all-1.75.0.jar - io.grpc-grpc-auth-1.75.0.jar diff --git a/distribution/shell/src/assemble/LICENSE.bin.txt b/distribution/shell/src/assemble/LICENSE.bin.txt index fb95fcf544478..ffb7a28aea159 100644 --- a/distribution/shell/src/assemble/LICENSE.bin.txt +++ b/distribution/shell/src/assemble/LICENSE.bin.txt @@ -333,7 +333,7 @@ The Apache Software License, Version 2.0 - failureaccess-1.0.1.jar - listenablefuture-9999.0-empty-to-avoid-conflict-with-guava.jar * J2ObjC Annotations -- j2objc-annotations-1.3.jar - * Netty Reactive Streams -- netty-reactive-streams-2.0.6.jar + * Netty Reactive Streams -- netty-reactive-streams-2.0.20.jar * Swagger -- swagger-annotations-1.6.2.jar * DataSketches - memory-0.8.3.jar @@ -345,22 +345,24 @@ The Apache Software License, Version 2.0 - commons-text-1.15.0.jar - commons-compress-1.28.0.jar * Netty - - netty-buffer-4.1.139.Final.jar - - netty-codec-4.1.139.Final.jar - - netty-codec-dns-4.1.139.Final.jar - - netty-codec-http-4.1.139.Final.jar - - netty-codec-socks-4.1.139.Final.jar - - netty-codec-haproxy-4.1.139.Final.jar - - netty-common-4.1.139.Final.jar - - netty-handler-4.1.139.Final.jar - - netty-handler-proxy-4.1.139.Final.jar - - netty-resolver-4.1.139.Final.jar - - netty-resolver-dns-4.1.139.Final.jar - - netty-transport-4.1.139.Final.jar - - netty-transport-classes-epoll-4.1.139.Final.jar - - netty-transport-native-epoll-4.1.139.Final-linux-aarch_64.jar - - netty-transport-native-epoll-4.1.139.Final-linux-x86_64.jar - - netty-transport-native-unix-common-4.1.139.Final.jar + - netty-buffer-4.2.19.Final.jar + - netty-codec-base-4.2.19.Final.jar + - netty-codec-compression-4.2.19.Final.jar + - netty-codec-dns-4.2.19.Final.jar + - netty-codec-http-4.2.19.Final.jar + - netty-codec-http2-4.2.19.Final.jar + - netty-codec-socks-4.2.19.Final.jar + - netty-codec-haproxy-4.2.19.Final.jar + - netty-common-4.2.19.Final.jar + - netty-handler-4.2.19.Final.jar + - netty-handler-proxy-4.2.19.Final.jar + - netty-resolver-4.2.19.Final.jar + - netty-resolver-dns-4.2.19.Final.jar + - netty-transport-4.2.19.Final.jar + - netty-transport-classes-epoll-4.2.19.Final.jar + - netty-transport-native-epoll-4.2.19.Final-linux-aarch_64.jar + - netty-transport-native-epoll-4.2.19.Final-linux-x86_64.jar + - netty-transport-native-unix-common-4.2.19.Final.jar - netty-tcnative-boringssl-static-2.0.84.Final.jar - netty-tcnative-boringssl-static-2.0.84.Final-linux-aarch_64.jar - netty-tcnative-boringssl-static-2.0.84.Final-linux-x86_64.jar @@ -368,12 +370,12 @@ The Apache Software License, Version 2.0 - netty-tcnative-boringssl-static-2.0.84.Final-osx-x86_64.jar - netty-tcnative-boringssl-static-2.0.84.Final-windows-x86_64.jar - netty-tcnative-classes-2.0.84.Final.jar - - netty-incubator-transport-classes-io_uring-0.0.26.Final.jar - - netty-incubator-transport-native-io_uring-0.0.26.Final-linux-aarch_64.jar - - netty-incubator-transport-native-io_uring-0.0.26.Final-linux-x86_64.jar - - netty-resolver-dns-classes-macos-4.1.139.Final.jar - - netty-resolver-dns-native-macos-4.1.139.Final-osx-aarch_64.jar - - netty-resolver-dns-native-macos-4.1.139.Final-osx-x86_64.jar + - netty-transport-classes-io_uring-4.2.19.Final.jar + - netty-transport-native-io_uring-4.2.19.Final-linux-aarch_64.jar + - netty-transport-native-io_uring-4.2.19.Final-linux-x86_64.jar + - netty-resolver-dns-classes-macos-4.2.19.Final.jar + - netty-resolver-dns-native-macos-4.2.19.Final-osx-aarch_64.jar + - netty-resolver-dns-native-macos-4.2.19.Final-osx-x86_64.jar * Prometheus client - simpleclient-0.16.0.jar - simpleclient_log4j2-0.16.0.jar @@ -398,8 +400,7 @@ The Apache Software License, Version 2.0 * AirCompressor - aircompressor-2.0.3.jar * AsyncHttpClient - - async-http-client-2.16.1.jar - - async-http-client-netty-utils-2.16.1.jar + - async-http-client-3.0.14.jar * Jetty - jetty-alpn-client-12.1.13.jar - jetty-client-12.1.13.jar diff --git a/pom.xml b/pom.xml index 43c529d29ef25..10aae1f062ef3 100644 --- a/pom.xml +++ b/pom.xml @@ -80,7 +80,8 @@ flexible messaging model and an intuitive client API. 17 17 ${maven.compiler.target} - 8 + + 11 3.13.0 @@ -184,8 +185,7 @@ flexible messaging model and an intuitive client API. 1.1.10.8 4.1.12.1 5.7.1 - 4.1.139.Final - 0.0.26.Final + 4.2.19.Final 12.1.13 9.4.58.v20250814 @@ -262,7 +262,7 @@ flexible messaging model and an intuitive client API. 0.16.1 7.9.2 2.0.3 - 2.16.1 + 3.0.14 3.19.0 2.22.0 1.22.1 @@ -306,7 +306,7 @@ flexible messaging model and an intuitive client API. 1.5.7-3 - 2.0.6 + 2.0.20 1.62.0 ${opentelemetry.version}-alpha 2.28.1 @@ -774,29 +774,6 @@ flexible messaging model and an intuitive client API. import - - io.netty.incubator - netty-incubator-transport-classes-io_uring - ${netty-iouring.version} - - - io.netty.incubator - netty-incubator-transport-native-io_uring - ${netty-iouring.version} - - - io.netty.incubator - netty-incubator-transport-native-io_uring - ${netty-iouring.version} - linux-x86_64 - - - io.netty.incubator - netty-incubator-transport-native-io_uring - ${netty-iouring.version} - linux-aarch_64 - - info.picocli picocli @@ -2491,20 +2468,6 @@ flexible messaging model and an intuitive client API. - - - integration-test-java8 - - 1.8 - - - - 8 - 8 - - - - coverage diff --git a/pulsar-broker-auth-oidc/pom.xml b/pulsar-broker-auth-oidc/pom.xml index e9a66a677495f..561683f9cae9d 100644 --- a/pulsar-broker-auth-oidc/pom.xml +++ b/pulsar-broker-auth-oidc/pom.xml @@ -40,6 +40,12 @@ + + + io.netty + netty-codec-http2 + + ${project.groupId} pulsar-broker-common diff --git a/pulsar-broker-auth-oidc/src/main/java/org/apache/pulsar/broker/authentication/oidc/AuthenticationProviderOpenID.java b/pulsar-broker-auth-oidc/src/main/java/org/apache/pulsar/broker/authentication/oidc/AuthenticationProviderOpenID.java index 00aa9d7af137a..6f19c15843c0d 100644 --- a/pulsar-broker-auth-oidc/src/main/java/org/apache/pulsar/broker/authentication/oidc/AuthenticationProviderOpenID.java +++ b/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; @@ -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(connectionTimeout)) + .setReadTimeout(Duration.ofMillis(readTimeout)) .setSslContext(sslContext) .build(); httpClient = new DefaultAsyncHttpClient(clientConfig); diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/PulsarChannelInitializer.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/PulsarChannelInitializer.java index dffd8260d463c..9a6d36634f194 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/PulsarChannelInitializer.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/PulsarChannelInitializer.java @@ -22,7 +22,6 @@ import io.netty.channel.ChannelHandler; import io.netty.channel.ChannelInitializer; import io.netty.channel.socket.SocketChannel; -import io.netty.handler.flow.FlowControlHandler; import io.netty.handler.flush.FlushConsolidationHandler; import io.netty.handler.ssl.SslHandler; import java.util.concurrent.TimeUnit; @@ -36,6 +35,7 @@ import org.apache.pulsar.common.protocol.OptionalProxyProtocolDecoder; import org.apache.pulsar.common.util.PulsarSslConfiguration; import org.apache.pulsar.common.util.PulsarSslFactory; +import org.apache.pulsar.common.util.netty.PulsarFlowControlHandler; @Slf4j public class PulsarChannelInitializer extends ChannelInitializer { @@ -98,7 +98,7 @@ protected void initChannel(SocketChannel ch) throws Exception { // as they like for any given input. so, disabling auto-read on `ByteToMessageDecoder` doesn't work properly and // ServerCnx ends up reading higher number of messages and broker can not throttle the messages by disabling // auto-read. - ch.pipeline().addLast("flowController", new FlowControlHandler()); + ch.pipeline().addLast("flowController", new PulsarFlowControlHandler()); // using "ChannelHandler" type to workaround an IntelliJ bug that shows a false positive error ChannelHandler cnx = newServerCnx(pulsar, listenerName); ch.pipeline().addLast("handler", cnx); 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 ca173a71ebf4b..aa8e9973a026b 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 @@ -157,16 +157,16 @@ 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())); } 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", PulsarVersion.getVersion())); - confBuilder.setRequestTimeout(requestTimeoutMs); + confBuilder.setRequestTimeout(Duration.ofMillis(requestTimeoutMs)); confBuilder.setIoThreadsCount(conf.getNumIoThreads()); confBuilder.setKeepAliveStrategy(new DefaultKeepAliveStrategy() { @Override diff --git a/pulsar-client-messagecrypto-bc/pom.xml b/pulsar-client-messagecrypto-bc/pom.xml index 13db04788fe4a..d44801e1a92f7 100644 --- a/pulsar-client-messagecrypto-bc/pom.xml +++ b/pulsar-client-messagecrypto-bc/pom.xml @@ -34,6 +34,18 @@ Pulsar Client Messagecrypto BC + + io.netty + netty-common + provided + + + + io.netty + netty-buffer + provided + + ${project.groupId} pulsar-common diff --git a/pulsar-client/pom.xml b/pulsar-client/pom.xml index 4539ccb90e789..ca55b0305d44f 100644 --- a/pulsar-client/pom.xml +++ b/pulsar-client/pom.xml @@ -32,6 +32,12 @@ Pulsar Client Java Original + + + + io.netty + netty-codec-http2 + ${project.groupId} pulsar-client-api diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ControlledClusterFailover.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ControlledClusterFailover.java index db3d7939d3b54..4cbdfdb4978e4 100644 --- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ControlledClusterFailover.java +++ b/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.Map; import java.util.Objects; import java.util.concurrent.ExecutionException; @@ -98,8 +99,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 diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/HttpClient.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/HttpClient.java index 0465d25be8f3b..ed96838c16723 100644 --- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/HttpClient.java +++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/HttpClient.java @@ -28,6 +28,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; @@ -88,8 +89,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", PulsarVersion.getVersion())); confBuilder.setKeepAliveStrategy(new DefaultKeepAliveStrategy() { @Override diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/auth/oauth2/FlowBase.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/auth/oauth2/FlowBase.java index b6d7b39f236ff..b039dc7918487 100644 --- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/auth/oauth2/FlowBase.java +++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/auth/oauth2/FlowBase.java @@ -101,10 +101,10 @@ private AsyncHttpClient defaultHttpClient(Duration readTimeout, Duration connect confBuilder.setUseProxyProperties(true); confBuilder.setFollowRedirect(true); confBuilder.setConnectTimeout( - getParameterDurationToMillis(CONFIG_PARAM_CONNECT_TIMEOUT, connectTimeout, + getParameterDuration(CONFIG_PARAM_CONNECT_TIMEOUT, connectTimeout, DEFAULT_CONNECT_TIMEOUT)); confBuilder.setReadTimeout( - getParameterDurationToMillis(CONFIG_PARAM_READ_TIMEOUT, readTimeout, DEFAULT_READ_TIMEOUT)); + 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); @@ -173,10 +173,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(); } diff --git a/pulsar-client/src/test/java/org/apache/pulsar/client/impl/AcknowledgementsGroupingTrackerTest.java b/pulsar-client/src/test/java/org/apache/pulsar/client/impl/AcknowledgementsGroupingTrackerTest.java index 410819ef732b6..e0d44a32d3082 100644 --- a/pulsar-client/src/test/java/org/apache/pulsar/client/impl/AcknowledgementsGroupingTrackerTest.java +++ b/pulsar-client/src/test/java/org/apache/pulsar/client/impl/AcknowledgementsGroupingTrackerTest.java @@ -44,7 +44,6 @@ import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicBoolean; -import lombok.var; import org.apache.pulsar.client.impl.conf.ClientConfigurationData; import org.apache.pulsar.client.impl.conf.ConsumerConfigurationData; import org.apache.pulsar.client.impl.metrics.InstrumentProvider; diff --git a/pulsar-client/src/test/java/org/apache/pulsar/client/impl/HttpClientTest.java b/pulsar-client/src/test/java/org/apache/pulsar/client/impl/HttpClientTest.java index c7ef13761a76f..116a6591efee7 100644 --- a/pulsar-client/src/test/java/org/apache/pulsar/client/impl/HttpClientTest.java +++ b/pulsar-client/src/test/java/org/apache/pulsar/client/impl/HttpClientTest.java @@ -134,4 +134,26 @@ public void testCrossOriginRedirectCarriesAuthorizationHeader() throws Exception .withHeader("Authorization", equalTo("Bearer test-token"))); } } + + @Test + public void testHttpsLookup() throws Exception { + WireMockServer httpsServer = new WireMockServer(WireMockConfiguration.wireMockConfig() + .port(0).httpsPort(0)); + httpsServer.start(); + try { + httpsServer.stubFor(get(urlPathMatching(LOOKUP_PATH)) + .willReturn(aResponse().withHeader("Content-Type", "application/json").withBody(EXPECTED_BODY))); + ClientConfigurationData conf = new ClientConfigurationData(); + conf.setServiceUrl("https://localhost:" + httpsServer.httpsPort()); + // WireMock's built-in keystore is self-signed. Exercise Pulsar's custom SSL engine with AHC 3.x. + conf.setTlsAllowInsecureConnection(true); + try (HttpClient httpClient = new HttpClient(conf, eventLoopGroup)) { + LookupData result = httpClient.get(LOOKUP_PATH, LookupData.class).get(10, TimeUnit.SECONDS); + assertEquals(result.getBrokerUrl(), "pulsar://broker-b:6650"); + httpsServer.verify(getRequestedFor(urlPathMatching(LOOKUP_PATH))); + } + } finally { + httpsServer.stop(); + } + } } diff --git a/pulsar-common/pom.xml b/pulsar-common/pom.xml index 48866591148bc..d6f000f3782e7 100644 --- a/pulsar-common/pom.xml +++ b/pulsar-common/pom.xml @@ -107,6 +107,11 @@ netty-resolver-dns + + io.netty + netty-transport-classes-epoll + + io.netty netty-transport-native-epoll @@ -161,19 +166,19 @@ - io.netty.incubator - netty-incubator-transport-classes-io_uring + io.netty + netty-transport-classes-io_uring - io.netty.incubator - netty-incubator-transport-native-io_uring + io.netty + netty-transport-native-io_uring linux-x86_64 - io.netty.incubator - netty-incubator-transport-native-io_uring + io.netty + netty-transport-native-io_uring linux-aarch_64 diff --git a/pulsar-common/src/main/java/org/apache/pulsar/common/util/netty/EventLoopUtil.java b/pulsar-common/src/main/java/org/apache/pulsar/common/util/netty/EventLoopUtil.java index cfc766890f02d..47a4cce2ee0b7 100644 --- a/pulsar-common/src/main/java/org/apache/pulsar/common/util/netty/EventLoopUtil.java +++ b/pulsar-common/src/main/java/org/apache/pulsar/common/util/netty/EventLoopUtil.java @@ -20,6 +20,8 @@ import io.netty.bootstrap.ServerBootstrap; import io.netty.channel.EventLoopGroup; +import io.netty.channel.IoEventLoopGroup; +import io.netty.channel.MultiThreadIoEventLoopGroup; import io.netty.channel.SelectStrategy; import io.netty.channel.epoll.Epoll; import io.netty.channel.epoll.EpollChannelOption; @@ -35,11 +37,11 @@ import io.netty.channel.socket.nio.NioDatagramChannel; import io.netty.channel.socket.nio.NioServerSocketChannel; import io.netty.channel.socket.nio.NioSocketChannel; -import io.netty.incubator.channel.uring.IOUring; -import io.netty.incubator.channel.uring.IOUringDatagramChannel; -import io.netty.incubator.channel.uring.IOUringEventLoopGroup; -import io.netty.incubator.channel.uring.IOUringServerSocketChannel; -import io.netty.incubator.channel.uring.IOUringSocketChannel; +import io.netty.channel.uring.IoUring; +import io.netty.channel.uring.IoUringDatagramChannel; +import io.netty.channel.uring.IoUringIoHandler; +import io.netty.channel.uring.IoUringServerSocketChannel; +import io.netty.channel.uring.IoUringSocketChannel; import java.util.concurrent.CompletableFuture; import java.util.concurrent.ThreadFactory; import lombok.extern.slf4j.Slf4j; @@ -62,9 +64,9 @@ public static EventLoopGroup newEventLoopGroup(int nThreads, boolean enableBusyW // By default, io_uring will not be enabled, even if available. The environment variable will be used: // enable.io_uring=1 if (StringUtils.equalsAnyIgnoreCase(enableIoUring, "1", "true")) { - // Throw exception if IOUring cannot be used - IOUring.ensureAvailability(); - return new IOUringEventLoopGroup(nThreads, threadFactory); + // Throw exception if io_uring cannot be used + IoUring.ensureAvailability(); + return new MultiThreadIoEventLoopGroup(nThreads, threadFactory, IoUringIoHandler.newFactory()); } else { if (!enableBusyWait) { // Regular Epoll based event loop @@ -96,6 +98,11 @@ public static EventLoopGroup newEventLoopGroup(int nThreads, boolean enableBusyW } } + private static boolean isIoUring(EventLoopGroup eventLoopGroup) { + return eventLoopGroup instanceof IoEventLoopGroup + && ((IoEventLoopGroup) eventLoopGroup).isIoType(IoUringIoHandler.class); + } + /** * Return a SocketChannel class suitable for the given EventLoopGroup implementation. * @@ -103,8 +110,8 @@ public static EventLoopGroup newEventLoopGroup(int nThreads, boolean enableBusyW * @return */ public static Class getClientSocketChannelClass(EventLoopGroup eventLoopGroup) { - if (eventLoopGroup instanceof IOUringEventLoopGroup) { - return IOUringSocketChannel.class; + if (isIoUring(eventLoopGroup)) { + return IoUringSocketChannel.class; } else if (eventLoopGroup instanceof EpollEventLoopGroup) { return EpollSocketChannel.class; } else { @@ -113,8 +120,8 @@ public static Class getClientSocketChannelClass(EventLo } public static Class getServerSocketChannelClass(EventLoopGroup eventLoopGroup) { - if (eventLoopGroup instanceof IOUringEventLoopGroup) { - return IOUringServerSocketChannel.class; + if (isIoUring(eventLoopGroup)) { + return IoUringServerSocketChannel.class; } else if (eventLoopGroup instanceof EpollEventLoopGroup) { return EpollServerSocketChannel.class; } else { @@ -123,8 +130,8 @@ public static Class getServerSocketChannelClass(E } public static Class getDatagramChannelClass(EventLoopGroup eventLoopGroup) { - if (eventLoopGroup instanceof IOUringEventLoopGroup) { - return IOUringDatagramChannel.class; + if (isIoUring(eventLoopGroup)) { + return IoUringDatagramChannel.class; } else if (eventLoopGroup instanceof EpollEventLoopGroup) { return EpollDatagramChannel.class; } else { diff --git a/pulsar-common/src/main/java/org/apache/pulsar/common/util/netty/PulsarFlowControlHandler.java b/pulsar-common/src/main/java/org/apache/pulsar/common/util/netty/PulsarFlowControlHandler.java new file mode 100644 index 0000000000000..e6637a4b2d0c3 --- /dev/null +++ b/pulsar-common/src/main/java/org/apache/pulsar/common/util/netty/PulsarFlowControlHandler.java @@ -0,0 +1,245 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +package org.apache.pulsar.common.util.netty; + +import io.netty.channel.ChannelConfig; +import io.netty.channel.ChannelDuplexHandler; +import io.netty.channel.ChannelHandler; +import io.netty.channel.ChannelHandlerContext; +import io.netty.handler.codec.ByteToMessageDecoder; +import io.netty.util.Recycler; +import io.netty.util.ReferenceCountUtil; +import io.netty.util.internal.ObjectPool.Handle; +import io.netty.util.internal.logging.InternalLogger; +import io.netty.util.internal.logging.InternalLoggerFactory; +import java.util.ArrayDeque; +import java.util.Queue; + +/** + * The {@link PulsarFlowControlHandler} ensures that only one message per {@code read()} is sent downstream. + * + *

Classes such as {@link ByteToMessageDecoder} are free to emit as many events as they like for any given input. + * A channel's auto reading configuration doesn't usually apply in these scenarios. This is causing problems in + * downstream {@link ChannelHandler}s that would like to hold subsequent events while they're processing one event. + * + *

This class is derived from {@code io.netty.handler.flow.FlowControlHandler} of Netty 4.2.14.Final + * (Copyright 2016 The Netty Project, licensed under the Apache License, version 2.0), which is behaviorally + * identical to the implementation in Netty 4.1.135.Final. Up to Netty + * 4.1.135/4.2.14, the dequeue loop of Netty's handler re-checked {@link ChannelConfig#isAutoRead()} before + * releasing each queued message, so a downstream handler calling {@code setAutoRead(false)} from inside + * {@code channelRead} stopped the delivery of queued messages immediately. Netty 4.2.15 (netty/netty#16837, + * backport of netty/netty#15053; the equivalent 4.1 change is netty/netty#16912) rewrote the handler to decide + * once up front: with auto-read enabled it dequeues the entire queue and ignores auto-read changes made mid-drain + * by downstream handlers. Pulsar's reactive request throttling (see {@code ServerCnxThrottleTracker} in the broker) + * pauses a connection by disabling auto-read while processing a message and relies on the delivery of queued + * messages stopping immediately; with Netty's rewritten handler, the whole queued backlog would be delivered + * regardless, defeating throttling (and rate limiting) of already-buffered traffic. This copy preserves the + * pre-4.2.15 per-message auto-read behavior. + * + * @see ChannelConfig#setAutoRead(boolean) + */ +public class PulsarFlowControlHandler extends ChannelDuplexHandler { + private static final InternalLogger logger = InternalLoggerFactory.getInstance(PulsarFlowControlHandler.class); + + private final boolean releaseMessages; + + private RecyclableArrayDeque queue; + + private ChannelConfig config; + + private boolean shouldConsume; + + public PulsarFlowControlHandler() { + this(true); + } + + public PulsarFlowControlHandler(boolean releaseMessages) { + this.releaseMessages = releaseMessages; + } + + /** + * Determine if the underlying {@link Queue} is empty. This method exists for + * testing, debugging and inspection purposes and it is not Thread safe! + */ + boolean isQueueEmpty() { + return queue == null || queue.isEmpty(); + } + + /** + * Releases all messages and destroys the {@link Queue}. + */ + private void destroy() { + if (queue != null) { + + if (!queue.isEmpty()) { + logger.trace("Non-empty queue: {}", queue); + + if (releaseMessages) { + Object msg; + while ((msg = queue.poll()) != null) { + ReferenceCountUtil.safeRelease(msg); + } + } + } + + queue.recycle(); + queue = null; + } + } + + @Override + public void handlerAdded(ChannelHandlerContext ctx) throws Exception { + config = ctx.channel().config(); + } + + @Override + public void handlerRemoved(ChannelHandlerContext ctx) throws Exception { + super.handlerRemoved(ctx); + if (!isQueueEmpty()) { + dequeue(ctx, queue.size()); + } + destroy(); + } + + @Override + public void channelInactive(ChannelHandlerContext ctx) throws Exception { + destroy(); + ctx.fireChannelInactive(); + } + + @Override + public void read(ChannelHandlerContext ctx) throws Exception { + if (dequeue(ctx, 1) == 0) { + // It seems no messages were consumed. We need to read() some + // messages from upstream and once one arrives it need to be + // relayed to downstream to keep the flow going. + shouldConsume = true; + ctx.read(); + } else if (config.isAutoRead()) { + ctx.read(); + } + } + + @Override + public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception { + if (queue == null) { + queue = RecyclableArrayDeque.newInstance(); + } + + queue.offer(msg); + + // We just received one message. Do we need to relay it regardless + // of the auto reading configuration? The answer is yes if this + // method was called as a result of a prior read() call. + int minConsume = shouldConsume ? 1 : 0; + shouldConsume = false; + + dequeue(ctx, minConsume); + } + + @Override + public void channelReadComplete(ChannelHandlerContext ctx) throws Exception { + if (isQueueEmpty()) { + ctx.fireChannelReadComplete(); + } else { + // Don't relay completion events from upstream as they + // make no sense in this context. See dequeue() where + // a new set of completion events is being produced. + } + } + + /** + * Dequeues one or many (or none) messages depending on the channel's auto + * reading state and returns the number of messages that were consumed from + * the internal queue. + * + *

The {@code minConsume} argument is used to force {@code dequeue()} into + * consuming that number of messages regardless of the channel's auto + * reading configuration. + * + * @see #read(ChannelHandlerContext) + * @see #channelRead(ChannelHandlerContext, Object) + */ + private int dequeue(ChannelHandlerContext ctx, int minConsume) { + int consumed = 0; + + // fireChannelRead(...) may call ctx.read() and so this method may reentrance. Because of this we need to + // check if queue was set to null in the meantime and if so break the loop. + while (queue != null && (consumed < minConsume || config.isAutoRead())) { + Object msg = queue.poll(); + if (msg == null) { + break; + } + + ++consumed; + ctx.fireChannelRead(msg); + } + + // We're firing a completion event every time one (or more) + // messages were consumed and the queue ended up being drained + // to an empty state. + if (queue != null && queue.isEmpty()) { + queue.recycle(); + queue = null; + + if (consumed > 0) { + ctx.fireChannelReadComplete(); + } + } + + return consumed; + } + + /** + * A recyclable {@link ArrayDeque}. + */ + private static final class RecyclableArrayDeque extends ArrayDeque { + + private static final long serialVersionUID = 0L; + + /** + * A value of {@code 2} should be a good choice for most scenarios. + */ + private static final int DEFAULT_NUM_ELEMENTS = 2; + + private static final Recycler RECYCLER = + new Recycler() { + @Override + protected RecyclableArrayDeque newObject(Handle handle) { + return new RecyclableArrayDeque(DEFAULT_NUM_ELEMENTS, handle); + } + }; + + public static RecyclableArrayDeque newInstance() { + return RECYCLER.get(); + } + + private final transient Handle handle; + + private RecyclableArrayDeque(int numElements, Handle handle) { + super(numElements); + this.handle = handle; + } + + public void recycle() { + clear(); + handle.recycle(this); + } + } +} diff --git a/pulsar-common/src/test/java/org/apache/pulsar/common/util/netty/EventLoopUtilTest.java b/pulsar-common/src/test/java/org/apache/pulsar/common/util/netty/EventLoopUtilTest.java new file mode 100644 index 0000000000000..51e86cefdda2d --- /dev/null +++ b/pulsar-common/src/test/java/org/apache/pulsar/common/util/netty/EventLoopUtilTest.java @@ -0,0 +1,85 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +package org.apache.pulsar.common.util.netty; + +import static org.testng.Assert.assertTrue; +import io.netty.bootstrap.Bootstrap; +import io.netty.bootstrap.ServerBootstrap; +import io.netty.channel.Channel; +import io.netty.channel.ChannelInboundHandlerAdapter; +import io.netty.channel.EventLoopGroup; +import io.netty.channel.MultiThreadIoEventLoopGroup; +import io.netty.channel.uring.IoUring; +import io.netty.channel.uring.IoUringIoHandler; +import io.netty.util.concurrent.DefaultThreadFactory; +import java.net.InetSocketAddress; +import org.testng.SkipException; +import org.testng.annotations.Test; + +public class EventLoopUtilTest { + @Test + public void testDefaultTransport() throws Exception { + EventLoopGroup group = EventLoopUtil.newEventLoopGroup(1, false, + new DefaultThreadFactory("event-loop-test")); + verifyTransports(group); + } + + @Test + public void testIoUringTransport() throws Exception { + if (!IoUring.isAvailable()) { + throw new SkipException("io_uring is unavailable", IoUring.unavailabilityCause()); + } + EventLoopGroup group = new MultiThreadIoEventLoopGroup(1, new DefaultThreadFactory("io-uring-test"), + IoUringIoHandler.newFactory()); + verifyTransports(group); + } + + private void verifyTransports(EventLoopGroup group) throws Exception { + Channel server = null; + Channel client = null; + Channel datagram = null; + try { + server = new ServerBootstrap().group(group) + .channel(EventLoopUtil.getServerSocketChannelClass(group)) + .childHandler(new ChannelInboundHandlerAdapter()) + .bind(new InetSocketAddress("127.0.0.1", 0)).sync().channel(); + client = new Bootstrap().group(group) + .channel(EventLoopUtil.getClientSocketChannelClass(group)) + .handler(new ChannelInboundHandlerAdapter()) + .connect(server.localAddress()).sync().channel(); + datagram = new Bootstrap().group(group) + .channel(EventLoopUtil.getDatagramChannelClass(group)) + .handler(new ChannelInboundHandlerAdapter()) + .bind(new InetSocketAddress("127.0.0.1", 0)).sync().channel(); + assertTrue(client.isActive()); + assertTrue(datagram.isActive()); + } finally { + if (client != null) { + client.close().sync(); + } + if (server != null) { + server.close().sync(); + } + if (datagram != null) { + datagram.close().sync(); + } + group.shutdownGracefully().sync(); + } + } +} diff --git a/pulsar-common/src/test/java/org/apache/pulsar/common/util/netty/PulsarFlowControlHandlerTest.java b/pulsar-common/src/test/java/org/apache/pulsar/common/util/netty/PulsarFlowControlHandlerTest.java new file mode 100644 index 0000000000000..4f62fcbb64cbe --- /dev/null +++ b/pulsar-common/src/test/java/org/apache/pulsar/common/util/netty/PulsarFlowControlHandlerTest.java @@ -0,0 +1,115 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +package org.apache.pulsar.common.util.netty; + +import static org.testng.Assert.assertEquals; +import static org.testng.Assert.assertFalse; +import static org.testng.Assert.assertTrue; +import io.netty.channel.ChannelHandlerContext; +import io.netty.channel.ChannelInboundHandlerAdapter; +import io.netty.channel.embedded.EmbeddedChannel; +import java.util.ArrayList; +import java.util.List; +import org.testng.annotations.Test; + +/** + * Tests the auto-read behavior of {@link PulsarFlowControlHandler} that Pulsar's reactive request throttling + * depends on: a downstream handler disabling auto-read from inside {@code channelRead} must stop the delivery + * of queued messages immediately. Netty's {@code FlowControlHandler} lost this behavior in Netty 4.2.15 + * (netty/netty#16837), which is why Pulsar carries this copy of the previous implementation. + */ +public class PulsarFlowControlHandlerTest { + + /** + * Downstream handler that records received messages and disables auto-read whenever the number of received + * messages reaches the next configured threshold, mimicking how ServerCnxThrottleTracker pauses a connection + * while processing a message. + */ + private static class ThrottlingHandler extends ChannelInboundHandlerAdapter { + private final List received = new ArrayList<>(); + private int pauseAtCount; + + ThrottlingHandler(int pauseAtCount) { + this.pauseAtCount = pauseAtCount; + } + + void pauseAgainAtCount(int count) { + this.pauseAtCount = count; + } + + @Override + public void channelRead(ChannelHandlerContext ctx, Object msg) { + received.add(msg); + if (received.size() == pauseAtCount) { + ctx.channel().config().setAutoRead(false); + } + } + } + + @Test + public void shouldStopDeliveryWhenAutoReadIsDisabledDuringChannelRead() { + PulsarFlowControlHandler flowControlHandler = new PulsarFlowControlHandler(); + ThrottlingHandler throttlingHandler = new ThrottlingHandler(1); + EmbeddedChannel channel = new EmbeddedChannel(flowControlHandler, throttlingHandler); + + channel.writeInbound("1", "2", "3", "4", "5"); + + // the downstream handler disabled auto-read while processing the first message; the remaining messages + // must be held in the flow control handler's queue instead of being delivered + assertEquals(throttlingHandler.received.size(), 1); + assertFalse(flowControlHandler.isQueueEmpty()); + + // re-enabling auto-read resumes delivery; the downstream handler pauses again on the next message, + // so exactly one more message must be delivered (per-message auto-read granularity) + throttlingHandler.pauseAgainAtCount(2); + channel.config().setAutoRead(true); + assertEquals(throttlingHandler.received.size(), 2); + assertFalse(flowControlHandler.isQueueEmpty()); + + // with auto-read left enabled, the rest of the queue drains + throttlingHandler.pauseAgainAtCount(-1); + channel.config().setAutoRead(true); + assertEquals(throttlingHandler.received.size(), 5); + assertTrue(flowControlHandler.isQueueEmpty()); + + assertFalse(channel.finish()); + } + + @Test + public void shouldDeliverOneMessagePerReadWhenAutoReadIsDisabled() { + PulsarFlowControlHandler flowControlHandler = new PulsarFlowControlHandler(); + ThrottlingHandler throttlingHandler = new ThrottlingHandler(0); + EmbeddedChannel channel = new EmbeddedChannel(flowControlHandler, throttlingHandler); + channel.config().setAutoRead(false); + + // channel activation (with auto-read initially enabled) issued one read() through the pipeline, so there + // is one outstanding message of read demand that the first written message satisfies + channel.writeInbound("1", "2", "3"); + assertEquals(throttlingHandler.received.size(), 1); + + channel.read(); + assertEquals(throttlingHandler.received.size(), 2); + + channel.read(); + assertEquals(throttlingHandler.received.size(), 3); + assertTrue(flowControlHandler.isQueueEmpty()); + + assertFalse(channel.finish()); + } +}