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 2025/10/06 20:46:16 UTC
(pulsar) branch branch-4.1 updated (4574b55920f -> 29e2a73bf73)
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a change to branch branch-4.1
in repository https://gitbox.apache.org/repos/asf/pulsar.git
from 4574b55920f [fix][build] Remove invalid profile in settings.xml that caused gpg signing to fail (#24812)
new bd5135ec329 [fix][client] Add description method to ClientBuilder (#24728)
new eabfdbd7c9a [improve][client] Allow adding custom description to User-Agent header (#24729)
new 80e9fef353d [fix][client] Fix PulsarAdmin description check and add test (#24734)
new 29e2a73bf73 [feat][client] Implement PIP-234 for sharing thread pools and DNS resolver/cache across multiple Pulsar Client instances (#24790)
The 4 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
build/run_unit_group.sh | 5 +-
.../org/apache/pulsar/broker/PulsarService.java | 2 +-
.../broker/auth/MockedPulsarServiceBaseTest.java | 15 +-
.../broker/service/ExclusiveProducerTest.java | 2 +
.../broker/service/TopicTerminationTest.java | 2 +
.../pulsar/client/api/ConsumerCleanupTest.java | 2 +
.../pulsar/client/api/ProducerCleanupTest.java | 2 +
...MultiListenersWithInternalListenerNameTest.java | 8 +-
.../client/api/SimpleProducerConsumerTest.java | 61 ++++-
.../pulsar/client/impl/PulsarTestClient.java | 2 +-
.../pulsar/client/impl/TopicDoesNotExistsTest.java | 4 +-
.../pulsar/client/admin/PulsarAdminBuilder.java | 17 ++
.../admin/internal/PulsarAdminBuilderImpl.java | 9 +
.../admin/internal/http/AsyncHttpConnector.java | 5 +-
.../admin/internal/PulsarAdminBuilderImplTest.java | 16 ++
.../apache/pulsar/client/api/ClientBuilder.java | 31 +++
.../pulsar/client/api/DnsResolverConfig.java | 126 +++++++++
.../{Messages.java => EventLoopGroupConfig.java} | 18 +-
.../client/api/PulsarClientSharedResources.java | 127 +++++++++
.../api/PulsarClientSharedResourcesBuilder.java | 122 +++++++++
...oducerAccessMode.java => ThreadPoolConfig.java} | 31 ++-
.../api/{MessagePayload.java => TimerConfig.java} | 25 +-
.../PulsarClientImplementationBinding.java | 3 +
.../pulsar/client/impl/AutoClusterFailover.java | 11 +-
.../pulsar/client/impl/ClientBuilderImpl.java | 36 +--
.../apache/pulsar/client/impl/ConnectionPool.java | 2 +-
.../client/impl/ControlledClusterFailover.java | 26 +-
.../pulsar/client/impl/DnsResolverGroupImpl.java | 54 +++-
.../org/apache/pulsar/client/impl/HttpClient.java | 19 +-
.../pulsar/client/impl/HttpLookupService.java | 14 +-
.../pulsar/client/impl/PulsarClientImpl.java | 67 ++---
.../PulsarClientImplementationBindingImpl.java | 6 +
.../impl/PulsarClientResourcesConfigurer.java | 149 +++++++++++
.../PulsarClientSharedResourcesBuilderImpl.java | 283 +++++++++++++++++++++
.../impl/PulsarClientSharedResourcesImpl.java | 160 ++++++++++++
.../pulsar/client/util/ExecutorProvider.java | 6 +-
.../client/util/ScheduledExecutorProvider.java | 3 +
.../src/main/resources/findbugsExclude.xml | 9 +
.../pulsar/client/impl/ClientBuilderImplTest.java | 17 ++
.../client/impl/ControlledClusterFailoverTest.java | 6 +
...PulsarClientSharedResourcesBuilderImplTest.java | 136 ++++++++++
.../pulsar/common/util/netty/DnsResolverUtil.java | 118 ++++++++-
.../pulsar/common/util/netty/EventLoopUtil.java | 76 +++++-
.../pulsar/common/util/netty/DnsResolverTest.java | 1 +
.../common/util/netty/DnsResolverUtilTest.java | 68 +++++
45 files changed, 1769 insertions(+), 133 deletions(-)
create mode 100644 pulsar-client-api/src/main/java/org/apache/pulsar/client/api/DnsResolverConfig.java
copy pulsar-client-api/src/main/java/org/apache/pulsar/client/api/{Messages.java => EventLoopGroupConfig.java} (68%)
create mode 100644 pulsar-client-api/src/main/java/org/apache/pulsar/client/api/PulsarClientSharedResources.java
create mode 100644 pulsar-client-api/src/main/java/org/apache/pulsar/client/api/PulsarClientSharedResourcesBuilder.java
copy pulsar-client-api/src/main/java/org/apache/pulsar/client/api/{ProducerAccessMode.java => ThreadPoolConfig.java} (54%)
copy pulsar-client-api/src/main/java/org/apache/pulsar/client/api/{MessagePayload.java => TimerConfig.java} (61%)
create mode 100644 pulsar-client/src/main/java/org/apache/pulsar/client/impl/PulsarClientResourcesConfigurer.java
create mode 100644 pulsar-client/src/main/java/org/apache/pulsar/client/impl/PulsarClientSharedResourcesBuilderImpl.java
create mode 100644 pulsar-client/src/main/java/org/apache/pulsar/client/impl/PulsarClientSharedResourcesImpl.java
create mode 100644 pulsar-client/src/test/java/org/apache/pulsar/client/impl/PulsarClientSharedResourcesBuilderImplTest.java
create mode 100644 pulsar-common/src/test/java/org/apache/pulsar/common/util/netty/DnsResolverUtilTest.java
(pulsar) 01/04: [fix][client] Add description method to ClientBuilder (#24728)
Posted by lh...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a commit to branch branch-4.1
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit bd5135ec3291797ff0d3c1fbbe114769c9b757bb
Author: Zixuan Liu <no...@gmail.com>
AuthorDate: Thu Sep 11 17:56:48 2025 +0800
[fix][client] Add description method to ClientBuilder (#24728)
(cherry picked from commit 65b3541dcfbdb2487b485b5702692b3f0730b4fe)
---
.../pulsar/client/api/SimpleProducerConsumerTest.java | 3 +--
.../org/apache/pulsar/client/api/ClientBuilder.java | 17 +++++++++++++++++
.../apache/pulsar/client/impl/ClientBuilderImpl.java | 16 +---------------
3 files changed, 19 insertions(+), 17 deletions(-)
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/SimpleProducerConsumerTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/SimpleProducerConsumerTest.java
index cc43bf06ded..2b1d83be591 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/SimpleProducerConsumerTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/SimpleProducerConsumerTest.java
@@ -98,7 +98,6 @@ import org.apache.pulsar.broker.storage.ManagedLedgerStorageClass;
import org.apache.pulsar.client.admin.PulsarAdminException;
import org.apache.pulsar.client.api.schema.GenericRecord;
import org.apache.pulsar.client.impl.BatchMessageIdImpl;
-import org.apache.pulsar.client.impl.ClientBuilderImpl;
import org.apache.pulsar.client.impl.ConsumerBase;
import org.apache.pulsar.client.impl.ConsumerImpl;
import org.apache.pulsar.client.impl.MessageIdImpl;
@@ -4868,7 +4867,7 @@ public class SimpleProducerConsumerTest extends ProducerConsumerBase {
assertEquals(stats.getPublishers().size(), 1);
assertEquals(stats.getPublishers().get(0).getClientVersion(), defaultClientVersion);
- PulsarClient client = ((ClientBuilderImpl) PulsarClient.builder())
+ PulsarClient client = PulsarClient.builder()
.description("my-java-client")
.serviceUrl(lookupUrl.toString())
.build();
diff --git a/pulsar-client-api/src/main/java/org/apache/pulsar/client/api/ClientBuilder.java b/pulsar-client-api/src/main/java/org/apache/pulsar/client/api/ClientBuilder.java
index c1ef9ed819b..034c167a9b3 100644
--- a/pulsar-client-api/src/main/java/org/apache/pulsar/client/api/ClientBuilder.java
+++ b/pulsar-client-api/src/main/java/org/apache/pulsar/client/api/ClientBuilder.java
@@ -717,4 +717,21 @@ public interface ClientBuilder extends Serializable, Cloneable {
* - The `loadManagerClassName` config in broker is a class that implements the `ExtensibleLoadManager` interface
*/
ClientBuilder lookupProperties(Map<String, String> properties);
+
+ /**
+ * Set the description.
+ *
+ * <p> By default, when the client connects to the broker, a version string like "Pulsar-Java-v<x.y.z>" will be
+ * carried and saved by the broker. The client version string could be queried from the topic stats.
+ *
+ * <p> This method provides a way to add more description to a specific PulsarClient instance. If it's configured,
+ * the description will be appended to the original client version string, with '-' as the separator.
+ *
+ * <p>For example, if the client version is 3.0.0, and the description is "forked", the final client version string
+ * will be "Pulsar-Java-v3.0.0-forked".
+ *
+ * @param description the description of the current PulsarClient instance
+ * @throws IllegalArgumentException if the length of description exceeds 64
+ */
+ ClientBuilder description(String description);
}
diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ClientBuilderImpl.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ClientBuilderImpl.java
index 7d851361505..d031c8e1710 100644
--- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ClientBuilderImpl.java
+++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ClientBuilderImpl.java
@@ -477,21 +477,7 @@ public class ClientBuilderImpl implements ClientBuilder {
return this;
}
- /**
- * Set the description.
- *
- * <p> By default, when the client connects to the broker, a version string like "Pulsar-Java-v<x.y.z>" will be
- * carried and saved by the broker. The client version string could be queried from the topic stats.
- *
- * <p> This method provides a way to add more description to a specific PulsarClient instance. If it's configured,
- * the description will be appended to the original client version string, with '-' as the separator.
- *
- * <p>For example, if the client version is 3.0.0, and the description is "forked", the final client version string
- * will be "Pulsar-Java-v3.0.0-forked".
- *
- * @param description the description of the current PulsarClient instance
- * @throws IllegalArgumentException if the length of description exceeds 64
- */
+ @Override
public ClientBuilder description(String description) {
if (description != null && description.length() > 64) {
throw new IllegalArgumentException("description should be at most 64 characters");
(pulsar) 02/04: [improve][client] Allow adding custom description to User-Agent header (#24729)
Posted by lh...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a commit to branch branch-4.1
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit eabfdbd7c9a2318aa3204a1da994bf7fa78a5503
Author: Zixuan Liu <no...@gmail.com>
AuthorDate: Fri Sep 12 09:59:02 2025 +0800
[improve][client] Allow adding custom description to User-Agent header (#24729)
(cherry picked from commit e19856a1a1bf065d86c6c61c6ce96577e3310eb5)
---
.../apache/pulsar/client/admin/PulsarAdminBuilder.java | 17 +++++++++++++++++
.../client/admin/internal/PulsarAdminBuilderImpl.java | 6 ++++++
.../client/admin/internal/http/AsyncHttpConnector.java | 5 ++++-
.../client/admin/internal/PulsarAdminImplTest.java | 9 +++++++++
.../java/org/apache/pulsar/client/impl/HttpClient.java | 5 ++++-
5 files changed, 40 insertions(+), 2 deletions(-)
diff --git a/pulsar-client-admin-api/src/main/java/org/apache/pulsar/client/admin/PulsarAdminBuilder.java b/pulsar-client-admin-api/src/main/java/org/apache/pulsar/client/admin/PulsarAdminBuilder.java
index 5c41d98b89d..216376c7eb6 100644
--- a/pulsar-client-admin-api/src/main/java/org/apache/pulsar/client/admin/PulsarAdminBuilder.java
+++ b/pulsar-client-admin-api/src/main/java/org/apache/pulsar/client/admin/PulsarAdminBuilder.java
@@ -376,4 +376,21 @@ public interface PulsarAdminBuilder {
* @return the PulsarAdminBuilder instance
*/
PulsarAdminBuilder connectionMaxIdleSeconds(int connectionMaxIdleSeconds);
+
+ /**
+ * Set the description.
+ *
+ * <p> By default, PulsarAdmin sends an HTTP <i>User-Agent</i> header such as
+ * <code>Pulsar-Java-v<x.y.z></code> when making requests to the broker.
+ *
+ * <p> This method provides a way to add more description to a specific PulsarAdmin instance. If it's configured,
+ * the description will be appended to the original admin version string, with '-' as the separator.
+ *
+ * <p>For example, if the admin version is 4.0.0, and the description is "forked", the final client version string
+ * will be "Pulsar-Java-v4.0.0-forked".
+ *
+ * @param description the description of the current PulsarAdmin instance
+ * @throws IllegalArgumentException if the length of description exceeds 64
+ */
+ PulsarAdminBuilder description(String description);
}
\ No newline at end of file
diff --git a/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/PulsarAdminBuilderImpl.java b/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/PulsarAdminBuilderImpl.java
index 7f0b3ab9a42..8a4659c5b87 100644
--- a/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/PulsarAdminBuilderImpl.java
+++ b/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/PulsarAdminBuilderImpl.java
@@ -283,4 +283,10 @@ public class PulsarAdminBuilderImpl implements PulsarAdminBuilder {
this.conf.setConnectionMaxIdleSeconds(connectionMaxIdleSeconds);
return this;
}
+
+ @Override
+ public PulsarAdminBuilder description(String description) {
+ this.conf.setDescription(description);
+ return this;
+ }
}
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 fb11d9e46d3..b9ef758e3c6 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
@@ -165,7 +165,10 @@ public class AsyncHttpConnector implements Connector, AsyncHttpRequestExecutor {
confBuilder.setRequestTimeout(conf.getRequestTimeoutMs());
confBuilder.setConnectTimeout(connectTimeoutMs);
confBuilder.setReadTimeout(readTimeoutMs);
- confBuilder.setUserAgent(String.format("Pulsar-Java-v%s", PulsarVersion.getVersion()));
+ confBuilder.setUserAgent(String.format("Pulsar-Java-v%s%s",
+ PulsarVersion.getVersion(),
+ (conf.getDescription() == null ? "" : ("-" + conf.getDescription()))
+ ));
confBuilder.setRequestTimeout(requestTimeoutMs);
confBuilder.setIoThreadsCount(conf.getNumIoThreads());
confBuilder.setKeepAliveStrategy(new DefaultKeepAliveStrategy() {
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 27c6fd96079..8b406b8bc6d 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
@@ -20,8 +20,11 @@ package org.apache.pulsar.client.admin.internal;
import static org.assertj.core.api.Assertions.assertThat;
import static org.mockito.Mockito.mock;
+import lombok.Cleanup;
import lombok.SneakyThrows;
+import org.apache.pulsar.client.admin.PulsarAdmin;
import org.apache.pulsar.client.api.Authentication;
+import org.apache.pulsar.client.api.PulsarClientException;
import org.apache.pulsar.client.impl.auth.AuthenticationDisabled;
import org.apache.pulsar.client.impl.conf.ClientConfigurationData;
import org.testng.annotations.Test;
@@ -45,6 +48,12 @@ public class PulsarAdminImplTest {
assertThat(createAdminAndGetAuth(conf)).isSameAs(auth);
}
+ @Test
+ public void testClientDescription() throws PulsarClientException {
+ @Cleanup PulsarAdmin ignored =
+ PulsarAdmin.builder().serviceHttpUrl("http://localhost:8080").description("forked").build();
+ }
+
@SneakyThrows
private Authentication createAdminAndGetAuth(ClientConfigurationData conf) {
try (PulsarAdminImpl admin = new PulsarAdminImpl("http://localhost:8080", conf, null)) {
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 f9312be39aa..633ca606388 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
@@ -84,7 +84,10 @@ public class HttpClient implements Closeable {
confBuilder.setMaxRedirects(conf.getMaxLookupRedirects());
confBuilder.setConnectTimeout(DEFAULT_CONNECT_TIMEOUT_IN_SECONDS * 1000);
confBuilder.setReadTimeout(DEFAULT_READ_TIMEOUT_IN_SECONDS * 1000);
- confBuilder.setUserAgent(String.format("Pulsar-Java-v%s", PulsarVersion.getVersion()));
+ confBuilder.setUserAgent(String.format("Pulsar-Java-v%s%s",
+ PulsarVersion.getVersion(),
+ (conf.getDescription() == null ? "" : ("-" + conf.getDescription()))
+ ));
confBuilder.setKeepAliveStrategy(new DefaultKeepAliveStrategy() {
@Override
public boolean keepAlive(InetSocketAddress remoteAddress, Request ahcRequest,
(pulsar) 03/04: [fix][client] Fix PulsarAdmin description check and add test (#24734)
Posted by lh...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a commit to branch branch-4.1
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 80e9fef353d60e3805b5bf79a2eab22d6e7b7bda
Author: Zixuan Liu <no...@gmail.com>
AuthorDate: Fri Sep 12 19:36:48 2025 +0800
[fix][client] Fix PulsarAdmin description check and add test (#24734)
(cherry picked from commit 8d7a28c145485265942e6dbe6b0b59a8ce8cb1ec)
---
.../client/admin/internal/PulsarAdminBuilderImpl.java | 3 +++
.../admin/internal/PulsarAdminBuilderImplTest.java | 16 ++++++++++++++++
.../client/admin/internal/PulsarAdminImplTest.java | 9 ---------
.../pulsar/client/impl/ClientBuilderImplTest.java | 17 +++++++++++++++++
4 files changed, 36 insertions(+), 9 deletions(-)
diff --git a/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/PulsarAdminBuilderImpl.java b/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/PulsarAdminBuilderImpl.java
index 8a4659c5b87..a046f13f60e 100644
--- a/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/PulsarAdminBuilderImpl.java
+++ b/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/PulsarAdminBuilderImpl.java
@@ -286,6 +286,9 @@ public class PulsarAdminBuilderImpl implements PulsarAdminBuilder {
@Override
public PulsarAdminBuilder description(String description) {
+ if (description != null && description.length() > 64) {
+ throw new IllegalArgumentException("description should be at most 64 characters");
+ }
this.conf.setDescription(description);
return this;
}
diff --git a/pulsar-client-admin/src/test/java/org/apache/pulsar/client/admin/internal/PulsarAdminBuilderImplTest.java b/pulsar-client-admin/src/test/java/org/apache/pulsar/client/admin/internal/PulsarAdminBuilderImplTest.java
index 52ac4656580..9339c893492 100644
--- a/pulsar-client-admin/src/test/java/org/apache/pulsar/client/admin/internal/PulsarAdminBuilderImplTest.java
+++ b/pulsar-client-admin/src/test/java/org/apache/pulsar/client/admin/internal/PulsarAdminBuilderImplTest.java
@@ -20,6 +20,7 @@ package org.apache.pulsar.client.admin.internal;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatIllegalArgumentException;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
import com.google.gson.Gson;
import com.google.gson.JsonObject;
import java.io.IOException;
@@ -181,6 +182,21 @@ public class PulsarAdminBuilderImplTest {
}
}
+ @Test
+ public void testClientDescription() throws PulsarClientException {
+ @Cleanup PulsarAdmin ignored =
+ PulsarAdmin.builder().serviceHttpUrl("http://localhost:8080").description("forked").build();
+ }
+
+ @Test
+ public void testClientDescriptionLengthExceed64() {
+ String longDescription = "a".repeat(65);
+ assertThatThrownBy(() -> {
+ @Cleanup PulsarAdmin ignored =
+ PulsarAdmin.builder().serviceHttpUrl("http://localhost:8080").description(longDescription).build();
+ }).isInstanceOf(IllegalArgumentException.class);
+ }
+
private String secretAuthParams(String secret) {
return String.format("{\"secret\":\"%s\"}", secret);
}
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 8b406b8bc6d..27c6fd96079 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
@@ -20,11 +20,8 @@ package org.apache.pulsar.client.admin.internal;
import static org.assertj.core.api.Assertions.assertThat;
import static org.mockito.Mockito.mock;
-import lombok.Cleanup;
import lombok.SneakyThrows;
-import org.apache.pulsar.client.admin.PulsarAdmin;
import org.apache.pulsar.client.api.Authentication;
-import org.apache.pulsar.client.api.PulsarClientException;
import org.apache.pulsar.client.impl.auth.AuthenticationDisabled;
import org.apache.pulsar.client.impl.conf.ClientConfigurationData;
import org.testng.annotations.Test;
@@ -48,12 +45,6 @@ public class PulsarAdminImplTest {
assertThat(createAdminAndGetAuth(conf)).isSameAs(auth);
}
- @Test
- public void testClientDescription() throws PulsarClientException {
- @Cleanup PulsarAdmin ignored =
- PulsarAdmin.builder().serviceHttpUrl("http://localhost:8080").description("forked").build();
- }
-
@SneakyThrows
private Authentication createAdminAndGetAuth(ClientConfigurationData conf) {
try (PulsarAdminImpl admin = new PulsarAdminImpl("http://localhost:8080", conf, null)) {
diff --git a/pulsar-client/src/test/java/org/apache/pulsar/client/impl/ClientBuilderImplTest.java b/pulsar-client/src/test/java/org/apache/pulsar/client/impl/ClientBuilderImplTest.java
index 93a5b90aec4..2194abb5a23 100644
--- a/pulsar-client/src/test/java/org/apache/pulsar/client/impl/ClientBuilderImplTest.java
+++ b/pulsar-client/src/test/java/org/apache/pulsar/client/impl/ClientBuilderImplTest.java
@@ -19,6 +19,7 @@
package org.apache.pulsar.client.impl;
import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
import static org.testng.Assert.fail;
import com.google.gson.Gson;
import com.google.gson.JsonObject;
@@ -28,6 +29,7 @@ import java.util.Arrays;
import java.util.Collections;
import java.util.HashMap;
import java.util.Map;
+import lombok.Cleanup;
import lombok.SneakyThrows;
import org.apache.pulsar.client.api.Authentication;
import org.apache.pulsar.client.api.AuthenticationDataProvider;
@@ -221,6 +223,21 @@ public class ClientBuilderImplTest {
assertThatAuthIsNotSet(auth);
}
+ @Test
+ public void testClientDescription() throws PulsarClientException {
+ @Cleanup PulsarClient ignored =
+ PulsarClient.builder().serviceUrl("pulsar://localhost:6650").description("forked").build();
+ }
+
+ @Test
+ public void testClientDescriptionLengthExceed64() {
+ String longDescription = "a".repeat(65);
+ assertThatThrownBy(() -> {
+ @Cleanup PulsarClient ignored =
+ PulsarClient.builder().serviceUrl("pulsar://localhost:6650").description(longDescription).build();
+ }).isInstanceOf(IllegalArgumentException.class);
+ }
+
private void assertThatAuthIsNotSet(Authentication authentication) {
// getAuthentication() returns disabled when null
assertThat(authentication).isInstanceOf(AuthenticationDisabled.class);
(pulsar) 04/04: [feat][client] Implement PIP-234 for sharing thread pools and DNS resolver/cache across multiple Pulsar Client instances (#24790)
Posted by lh...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a commit to branch branch-4.1
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 29e2a73bf7322d68dd7d59822430d1a90dca5bbc
Author: Lari Hotari <lh...@users.noreply.github.com>
AuthorDate: Sun Sep 28 02:21:51 2025 +0300
[feat][client] Implement PIP-234 for sharing thread pools and DNS resolver/cache across multiple Pulsar Client instances (#24790)
(cherry picked from commit 8559b689fc6cdca2d6e13aeffb56996fc46a6f04)
---
build/run_unit_group.sh | 5 +-
.../org/apache/pulsar/broker/PulsarService.java | 2 +-
.../broker/auth/MockedPulsarServiceBaseTest.java | 15 +-
.../broker/service/ExclusiveProducerTest.java | 2 +
.../broker/service/TopicTerminationTest.java | 2 +
.../pulsar/client/api/ConsumerCleanupTest.java | 2 +
.../pulsar/client/api/ProducerCleanupTest.java | 2 +
...MultiListenersWithInternalListenerNameTest.java | 8 +-
.../client/api/SimpleProducerConsumerTest.java | 58 +++++
.../pulsar/client/impl/PulsarTestClient.java | 2 +-
.../pulsar/client/impl/TopicDoesNotExistsTest.java | 4 +-
.../apache/pulsar/client/api/ClientBuilder.java | 14 +
.../pulsar/client/api/DnsResolverConfig.java | 126 +++++++++
.../pulsar/client/api/EventLoopGroupConfig.java | 28 +-
.../client/api/PulsarClientSharedResources.java | 127 +++++++++
.../api/PulsarClientSharedResourcesBuilder.java | 122 +++++++++
.../apache/pulsar/client/api/ThreadPoolConfig.java | 48 ++++
.../org/apache/pulsar/client/api/TimerConfig.java | 35 ++-
.../PulsarClientImplementationBinding.java | 3 +
.../pulsar/client/impl/AutoClusterFailover.java | 11 +-
.../pulsar/client/impl/ClientBuilderImpl.java | 20 +-
.../apache/pulsar/client/impl/ConnectionPool.java | 2 +-
.../client/impl/ControlledClusterFailover.java | 26 +-
.../pulsar/client/impl/DnsResolverGroupImpl.java | 54 +++-
.../org/apache/pulsar/client/impl/HttpClient.java | 14 +-
.../pulsar/client/impl/HttpLookupService.java | 14 +-
.../pulsar/client/impl/PulsarClientImpl.java | 67 ++---
.../PulsarClientImplementationBindingImpl.java | 6 +
.../impl/PulsarClientResourcesConfigurer.java | 149 +++++++++++
.../PulsarClientSharedResourcesBuilderImpl.java | 283 +++++++++++++++++++++
.../impl/PulsarClientSharedResourcesImpl.java | 160 ++++++++++++
.../pulsar/client/util/ExecutorProvider.java | 6 +-
.../client/util/ScheduledExecutorProvider.java | 3 +
.../src/main/resources/findbugsExclude.xml | 9 +
.../client/impl/ControlledClusterFailoverTest.java | 6 +
...PulsarClientSharedResourcesBuilderImplTest.java | 136 ++++++++++
.../pulsar/common/util/netty/DnsResolverUtil.java | 118 ++++++++-
.../pulsar/common/util/netty/EventLoopUtil.java | 76 +++++-
.../pulsar/common/util/netty/DnsResolverTest.java | 1 +
.../common/util/netty/DnsResolverUtilTest.java | 68 +++++
40 files changed, 1727 insertions(+), 107 deletions(-)
diff --git a/build/run_unit_group.sh b/build/run_unit_group.sh
index 02347648b94..cfdf94d6436 100755
--- a/build/run_unit_group.sh
+++ b/build/run_unit_group.sh
@@ -177,10 +177,13 @@ function test_group_other() {
**/OffloadersCacheTest.java
**/PrimitiveSchemaTest.java,
**/BlobStoreManagedLedgerOffloaderTest.java,
- **/BlobStoreManagedLedgerOffloaderStreamingTest.java'
+ **/BlobStoreManagedLedgerOffloaderStreamingTest.java,
+ **/DnsResolverTest.java'
mvn_test -pl managed-ledger -Dinclude='**/ManagedLedgerTest.java,
**/OffloadersCacheTest.java'
+ # DnsResolverTest needs to be run separately since it relies on static field values
+ mvn_test -pl pulsar-common -Dinclude='**/DnsResolverTest.java'
mvn_test -pl tiered-storage/jcloud -Dinclude='**/BlobStoreManagedLedgerOffloaderTest.java'
mvn_test -pl tiered-storage/jcloud -Dinclude='**/BlobStoreManagedLedgerOffloaderStreamingTest.java'
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/PulsarService.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/PulsarService.java
index aa5c0b0ad2d..15fe1dcd610 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/PulsarService.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/PulsarService.java
@@ -401,7 +401,7 @@ public class PulsarService implements AutoCloseable, ShutdownService {
this.brokerClientSharedLookupExecutorProvider =
new ScheduledExecutorProvider(1, "broker-client-shared-lookup-executor");
this.brokerClientSharedDnsResolverGroup =
- new DnsResolverGroupImpl(this.ioEventLoopGroup,
+ new DnsResolverGroupImpl(
loadBrokerClientProperties(new ClientConfigurationData()));
// here in the constructor we don't have the offloader scheduler yet
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/auth/MockedPulsarServiceBaseTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/auth/MockedPulsarServiceBaseTest.java
index 6fdfee86b9f..b349ba71922 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/auth/MockedPulsarServiceBaseTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/auth/MockedPulsarServiceBaseTest.java
@@ -182,7 +182,15 @@ public abstract class MockedPulsarServiceBaseTest extends TestRetrySupport {
}
private URI resolveLookupUrl() {
- if (isTcpLookup) {
+ return resolveLookupUrl(isTcpLookup);
+ }
+
+ protected String brokerServiceUrl(boolean usePulsarBinaryProtocol) {
+ return resolveLookupUrl(usePulsarBinaryProtocol).toString();
+ }
+
+ private URI resolveLookupUrl(boolean usePulsarBinaryProtocol) {
+ if (usePulsarBinaryProtocol) {
return URI.create(pulsar.getBrokerServiceUrl());
} else {
return URI.create(brokerUrl != null
@@ -801,5 +809,10 @@ public abstract class MockedPulsarServiceBaseTest extends TestRetrySupport {
BrokerTestUtil.logTopicStats(log, admin, topic);
}
+ @DataProvider(name = "trueFalse")
+ public static Object[][] trueFalse() {
+ return new Object[][] { { Boolean.TRUE }, { Boolean.FALSE } };
+ }
+
private static final Logger log = LoggerFactory.getLogger(MockedPulsarServiceBaseTest.class);
}
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ExclusiveProducerTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ExclusiveProducerTest.java
index 9a070efa95d..c108d5b8ca1 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ExclusiveProducerTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ExclusiveProducerTest.java
@@ -49,6 +49,8 @@ public class ExclusiveProducerTest extends BrokerTestBase {
@BeforeClass
protected void setup() throws Exception {
+ // use Pulsar binary lookup since the HTTP client shares the Pulsar client timer
+ isTcpLookup = true;
baseSetup();
}
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/TopicTerminationTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/TopicTerminationTest.java
index a2436e4360e..e4e361eb0b1 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/TopicTerminationTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/TopicTerminationTest.java
@@ -56,6 +56,8 @@ public class TopicTerminationTest extends BrokerTestBase {
@BeforeMethod
@Override
protected void setup() throws Exception {
+ // use Pulsar binary lookup since the HTTP client shares the Pulsar client timer
+ isTcpLookup = true;
super.baseSetup();
}
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/ConsumerCleanupTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/ConsumerCleanupTest.java
index cc7c2cabb87..0cbfde06381 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/ConsumerCleanupTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/ConsumerCleanupTest.java
@@ -34,6 +34,8 @@ public class ConsumerCleanupTest extends ProducerConsumerBase {
@BeforeClass
@Override
protected void setup() throws Exception {
+ // use Pulsar binary lookup since the HTTP client shares the Pulsar client timer
+ isTcpLookup = true;
super.internalSetup();
super.producerBaseSetup();
}
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/ProducerCleanupTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/ProducerCleanupTest.java
index 5ad3c85441b..f4daba3bf05 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/ProducerCleanupTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/ProducerCleanupTest.java
@@ -33,6 +33,8 @@ public class ProducerCleanupTest extends ProducerConsumerBase {
@BeforeMethod
@Override
protected void setup() throws Exception {
+ // use Pulsar binary lookup since the HTTP client shares the Pulsar client timer
+ isTcpLookup = true;
super.internalSetup();
super.producerBaseSetup();
}
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/PulsarMultiListenersWithInternalListenerNameTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/PulsarMultiListenersWithInternalListenerNameTest.java
index 094c32c7ae6..2844d211a30 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/PulsarMultiListenersWithInternalListenerNameTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/PulsarMultiListenersWithInternalListenerNameTest.java
@@ -22,6 +22,8 @@ import static org.mockito.ArgumentMatchers.any;
import static org.mockito.Mockito.doReturn;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
+import io.netty.resolver.DefaultNameResolver;
+import io.netty.util.concurrent.ImmediateEventExecutor;
import java.io.IOException;
import java.io.UncheckedIOException;
import java.net.InetAddress;
@@ -137,7 +139,8 @@ public class PulsarMultiListenersWithInternalListenerNameTest extends MockedPuls
conf.setMaxLookupRedirects(10);
@Cleanup
- LookupService lookupService = useHttp ? new HttpLookupService(InstrumentProvider.NOOP, conf, eventExecutors) :
+ LookupService lookupService = useHttp ? new HttpLookupService(InstrumentProvider.NOOP, conf, eventExecutors,
+ null, new DefaultNameResolver(ImmediateEventExecutor.INSTANCE)) :
new BinaryProtoLookupService((PulsarClientImpl) this.pulsarClient,
lookupUrl.toString(), "internal", false, this.executorService);
TopicName topicName = TopicName.get("persistent://public/default/test");
@@ -173,7 +176,8 @@ public class PulsarMultiListenersWithInternalListenerNameTest extends MockedPuls
conf.setMaxLookupRedirects(10);
@Cleanup
- HttpLookupService lookupService = new HttpLookupService(InstrumentProvider.NOOP, conf, eventExecutors);
+ HttpLookupService lookupService = new HttpLookupService(InstrumentProvider.NOOP, conf, eventExecutors, null,
+ new DefaultNameResolver(ImmediateEventExecutor.INSTANCE));
NamespaceService namespaceService = pulsar.getNamespaceService();
LookupResult lookupResult = new LookupResult(pulsar.getWebServiceAddress(), null,
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/SimpleProducerConsumerTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/SimpleProducerConsumerTest.java
index 2b1d83be591..4e15062b1e2 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/SimpleProducerConsumerTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/SimpleProducerConsumerTest.java
@@ -42,6 +42,7 @@ import io.netty.channel.ChannelDuplexHandler;
import io.netty.channel.ChannelHandlerContext;
import io.netty.util.Timeout;
import java.io.ByteArrayInputStream;
+import java.io.Closeable;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.nio.file.Files;
@@ -5304,4 +5305,61 @@ public class SimpleProducerConsumerTest extends ProducerConsumerBase {
producer.close();
consumer.close();
}
+
+ @Test(dataProvider = "trueFalse")
+ public void testResourceSharingEndToEnd(boolean usePulsarBinaryProtocol) throws Exception {
+ log.info("-- Starting {} test --", methodName);
+ String topic = BrokerTestUtil.newUniqueName("persistent://my-property/my-ns/my-topic");
+ int numberOfClients = 50;
+ List<PulsarClient> clients = new ArrayList<>();
+ @Cleanup
+ Closeable closeClients = () -> {
+ for (PulsarClient client : clients) {
+ try {
+ client.close();
+ } catch (PulsarClientException e) {
+ log.error("Failed to close client {}", client, e);
+ }
+ }
+ };
+ @Cleanup
+ PulsarClientSharedResources sharedResources = PulsarClientSharedResources.builder().build();
+ List<Consumer<byte[]>> consumers = new ArrayList<>();
+
+ for (int i = 0; i < numberOfClients; i++) {
+ PulsarClient client = PulsarClient.builder()
+ .serviceUrl(brokerServiceUrl(usePulsarBinaryProtocol))
+ .sharedResources(sharedResources)
+ .build();
+ clients.add(client);
+ consumers.add(client.newConsumer().topic(topic)
+ .subscriptionName("my-subscriber-name" + i).subscribe());
+ }
+
+ ProducerBuilder<byte[]> producerBuilder = pulsarClient.newProducer()
+ .topic(topic);
+
+ Producer<byte[]> producer = producerBuilder.create();
+ for (int i = 0; i < 10; i++) {
+ String message = "my-message-" + i;
+ producer.send(message.getBytes());
+ }
+
+ for (Consumer<byte[]> consumer : consumers) {
+ Message<byte[]> msg = null;
+ Set<String> messageSet = new HashSet<>();
+ for (int i = 0; i < 10; i++) {
+ msg = consumer.receive(RECEIVE_TIMEOUT_SECONDS, TimeUnit.SECONDS);
+ String receivedMessage = new String(msg.getData());
+ log.debug("Received message: [{}]", receivedMessage);
+ String expectedMessage = "my-message-" + i;
+ testMessageOrderAndDuplicates(messageSet, receivedMessage, expectedMessage);
+ }
+ // Acknowledge the consumption of all messages at once
+ consumer.acknowledgeCumulative(msg);
+ consumer.close();
+ }
+
+ log.info("-- Exiting {} test --", methodName);
+ }
}
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/PulsarTestClient.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/PulsarTestClient.java
index 27301db9179..db330951f74 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/PulsarTestClient.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/PulsarTestClient.java
@@ -91,7 +91,7 @@ public class PulsarTestClient extends PulsarClientImpl {
private PulsarTestClient(ClientConfigurationData conf, EventLoopGroup eventLoopGroup, ConnectionPool cnxPool,
AtomicReference<Supplier<ClientCnx>> clientCnxSupplierReference)
throws PulsarClientException {
- super(conf, eventLoopGroup, cnxPool);
+ super(conf, eventLoopGroup, cnxPool, null, null, null, null, null, new DnsResolverGroupImpl(conf));
// workaround initialization order issue so that ClientCnx can be created in this class
clientCnxSupplierReference.set(this::createClientCnx);
}
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/TopicDoesNotExistsTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/TopicDoesNotExistsTest.java
index b48b97978b7..f528c827549 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/TopicDoesNotExistsTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/TopicDoesNotExistsTest.java
@@ -39,6 +39,8 @@ public class TopicDoesNotExistsTest extends ProducerConsumerBase {
@Override
@BeforeClass
public void setup() throws Exception {
+ // use Pulsar binary lookup since the HTTP client shares the Pulsar client timer
+ isTcpLookup = true;
conf.setAllowAutoTopicCreation(false);
super.internalSetup();
super.producerBaseSetup();
@@ -61,7 +63,7 @@ public class TopicDoesNotExistsTest extends ProducerConsumerBase {
.create();
Assert.fail("Create producer should failed while topic does not exists.");
} catch (PulsarClientException e) {
- Assert.assertTrue(e instanceof PulsarClientException.NotFoundException);
+ Assert.assertTrue(e instanceof PulsarClientException.TopicDoesNotExistException);
}
Thread.sleep(2000);
HashedWheelTimer timer = (HashedWheelTimer) ((PulsarClientImpl) pulsarClient).timer();
diff --git a/pulsar-client-api/src/main/java/org/apache/pulsar/client/api/ClientBuilder.java b/pulsar-client-api/src/main/java/org/apache/pulsar/client/api/ClientBuilder.java
index 034c167a9b3..d31d42bbe63 100644
--- a/pulsar-client-api/src/main/java/org/apache/pulsar/client/api/ClientBuilder.java
+++ b/pulsar-client-api/src/main/java/org/apache/pulsar/client/api/ClientBuilder.java
@@ -734,4 +734,18 @@ public interface ClientBuilder extends Serializable, Cloneable {
* @throws IllegalArgumentException if the length of description exceeds 64
*/
ClientBuilder description(String description);
+
+ /**
+ * Provide a set of shared client resources to be reused by this client.
+ * <p>
+ * Providing a shared resource instance allows PulsarClient instances to share resources
+ * (such as IO/event loops, timers, executors, DNS resolver/cache) with other PulsarClient
+ * instances, reducing memory footprint and thread usage when creating many clients in the same JVM.
+ *
+ * @param sharedResources the shared resources instance created with {@link PulsarClientSharedResources#builder()}
+ * @return the client builder instance
+ * @see PulsarClientSharedResources
+ * @see PulsarClientSharedResourcesBuilder
+ */
+ ClientBuilder sharedResources(PulsarClientSharedResources sharedResources);
}
diff --git a/pulsar-client-api/src/main/java/org/apache/pulsar/client/api/DnsResolverConfig.java b/pulsar-client-api/src/main/java/org/apache/pulsar/client/api/DnsResolverConfig.java
new file mode 100644
index 00000000000..09c8f5fccfc
--- /dev/null
+++ b/pulsar-client-api/src/main/java/org/apache/pulsar/client/api/DnsResolverConfig.java
@@ -0,0 +1,126 @@
+/*
+ * 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.client.api;
+
+import java.net.InetSocketAddress;
+
+/**
+ * Configuration interface for DNS resolver settings.
+ */
+public interface DnsResolverConfig {
+ /**
+ * Sets the minimum TTL for DNS records.
+ * Defaults to 0.
+ *
+ * @param minTtl the minimum TTL value
+ * @return the DNS resolver configuration instance for chained calls
+ */
+ DnsResolverConfig minTtl(int minTtl);
+
+ /**
+ * Sets the maximum TTL for DNS records.
+ * Defaults to JDK level setting.
+ *
+ * @param maxTtl the maximum TTL value
+ * @return the DNS resolver configuration instance for chained calls
+ */
+ DnsResolverConfig maxTtl(int maxTtl);
+
+ /**
+ * Sets the TTL for negative DNS responses.
+ * Defaults to JDK level setting.
+ *
+ * @param negativeTtl the negative TTL value
+ * @return the DNS resolver configuration instance for chained calls
+ */
+ DnsResolverConfig negativeTtl(int negativeTtl);
+
+ /**
+ * Sets the query timeout in milliseconds.
+ * Defaults to the OS level setting.
+ *
+ * @param queryTimeoutMillis the timeout value in milliseconds
+ * @return the DNS resolver configuration instance for chained calls
+ */
+ DnsResolverConfig queryTimeoutMillis(long queryTimeoutMillis);
+
+ /**
+ * Enables or disables TCP fallback for DNS queries.
+ * Defaults to true.
+ *
+ * @param tcpFallbackEnabled true to enable TCP fallback, false to disable
+ * @return the DNS resolver configuration instance for chained calls
+ */
+ DnsResolverConfig tcpFallbackEnabled(boolean tcpFallbackEnabled);
+
+ /**
+ * Enables or disables TCP fallback on timeout for DNS queries.
+ * Defaults to true.
+ *
+ * @param tcpFallbackOnTimeoutEnabled true to enable TCP fallback on timeout, false to disable
+ * @return the DNS resolver configuration instance for chained calls
+ */
+ DnsResolverConfig tcpFallbackOnTimeoutEnabled(boolean tcpFallbackOnTimeoutEnabled);
+
+ /**
+ * Sets the ndots value for DNS resolution. Set this to 1 to disable the use of DNS search domains.
+ * Defaults to the OS level configuration.
+ *
+ * @param ndots the ndots value
+ * @return the DNS resolver configuration instance for chained calls
+ */
+ DnsResolverConfig ndots(int ndots);
+
+ /**
+ * Sets the search domains for DNS resolution.
+ * Defaults to the OS level configuration.
+ *
+ * @param searchDomains collection of search domains
+ * @return the DNS resolver configuration instance for chained calls
+ */
+ DnsResolverConfig searchDomains(Iterable<String> searchDomains);
+
+ /**
+ * Sets the local bind address and port for DNS lookup client.
+ * By default any available port is used. This setting is mainly in cases where it's necessary to bind
+ * the DNS client to a specific address and optionally also the port.
+ *
+ * @param localAddress the socket address (address and port) to bind the DNS client to
+ * @return the DNS resolver configuration instance for chained calls
+ */
+ DnsResolverConfig localAddress(InetSocketAddress localAddress);
+
+ /**
+ * Sets the DNS server addresses to be used for DNS lookups.
+ * Defaults to the OS level configuration.
+ *
+ * @param addresses collection of DNS server socket addresses (address and port)
+ * @return the DNS resolver configuration instance for chained calls
+ */
+ DnsResolverConfig serverAddresses(Iterable<InetSocketAddress> addresses);
+
+ /**
+ * Enable detailed trace information for resolution failure exception messages.
+ * Defaults to true.
+ *
+ * @param traceEnabled true to enable tracing, false to disable
+ * @return the DNS resolver configuration instance for chained calls
+ */
+ DnsResolverConfig traceEnabled(boolean traceEnabled);
+}
diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/util/ScheduledExecutorProvider.java b/pulsar-client-api/src/main/java/org/apache/pulsar/client/api/EventLoopGroupConfig.java
similarity index 60%
copy from pulsar-client/src/main/java/org/apache/pulsar/client/util/ScheduledExecutorProvider.java
copy to pulsar-client-api/src/main/java/org/apache/pulsar/client/api/EventLoopGroupConfig.java
index 3ebd41ebe56..84b8b78ee41 100644
--- a/pulsar-client/src/main/java/org/apache/pulsar/client/util/ScheduledExecutorProvider.java
+++ b/pulsar-client-api/src/main/java/org/apache/pulsar/client/api/EventLoopGroupConfig.java
@@ -16,21 +16,17 @@
* specific language governing permissions and limitations
* under the License.
*/
-package org.apache.pulsar.client.util;
+package org.apache.pulsar.client.api;
-import java.util.concurrent.ExecutorService;
-import java.util.concurrent.Executors;
-import lombok.extern.slf4j.Slf4j;
-
-@Slf4j
-public class ScheduledExecutorProvider extends ExecutorProvider {
-
- public ScheduledExecutorProvider(int numThreads, String poolName) {
- super(numThreads, poolName);
- }
-
- @Override
- protected ExecutorService createExecutor(ExtendedThreadFactory threadFactory) {
- return Executors.newSingleThreadScheduledExecutor(threadFactory);
- }
+/**
+ * Configuration interface for event loop group settings.
+ */
+public interface EventLoopGroupConfig extends ThreadPoolConfig<EventLoopGroupConfig> {
+ /**
+ * Enables or disables busy-wait polling mode.
+ *
+ * @param enableBusyWait true to enable busy-wait polling, false to disable
+ * @return this config instance for method chaining
+ */
+ EventLoopGroupConfig enableBusyWait(boolean enableBusyWait);
}
diff --git a/pulsar-client-api/src/main/java/org/apache/pulsar/client/api/PulsarClientSharedResources.java b/pulsar-client-api/src/main/java/org/apache/pulsar/client/api/PulsarClientSharedResources.java
new file mode 100644
index 00000000000..34a56ebd608
--- /dev/null
+++ b/pulsar-client-api/src/main/java/org/apache/pulsar/client/api/PulsarClientSharedResources.java
@@ -0,0 +1,127 @@
+/*
+ * 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.client.api;
+
+import java.util.Set;
+import org.apache.pulsar.client.internal.DefaultImplementation;
+
+/**
+ * Manages shared resources across multiple PulsarClient instances to optimize resource utilization.
+ * This interface provides access to common resources such as thread pools and DNS resolver/cache
+ * that can be shared between multiple PulsarClient instances.
+ *
+ * This allows creating a large number of PulsarClient instances in a single JVM without wasting resources.
+ * Sharing the DNS resolver and cache will help reduce the number of DNS lookups performed by the clients
+ * and improve latency and reduce load on the DNS servers.
+ *
+ * <p>Key features:
+ * <ul>
+ * <li>Thread pool sharing for various Pulsar operations (IO, timer, lookup, etc.)</li>
+ * <li>Shared DNS resolver and cache</li>
+ * <li>Resource lifecycle management</li>
+ * </ul>
+ *
+ * <p>Usage example:
+ * <pre>{@code
+ * // To share all possible resources across multiple PulsarClient instances
+ * PulsarClientSharedResources sharedResources = PulsarClientSharedResources.builder()
+ * .build();
+ * // Use these resources with multiple PulsarClient instances
+ * PulsarClient client1 = PulsarClient.builder().sharedResources(sharedResources).build();
+ * PulsarClient client2 = PulsarClient.builder().sharedResources(sharedResources).build();
+ * client1.close();
+ * client2.close();
+ * sharedResources.close();
+ * }</pre>
+ *
+ * <p>The instance should be created using the {@link PulsarClientSharedResourcesBuilder},
+ * which can be obtained via the {@link #builder()} method.
+ *
+ * @see PulsarClientSharedResourcesBuilder
+ * @see SharedResource
+ */
+public interface PulsarClientSharedResources extends AutoCloseable {
+ enum SharedResource {
+ // pulsar-io threadpool
+ EventLoopGroup(SharedResourceType.EventLoopGroup),
+ // pulsar-external-listener threadpool
+ ListenerExecutor(SharedResourceType.ThreadPool),
+ // pulsar-timer threadpool
+ Timer(SharedResourceType.Timer),
+ // pulsar-client-internal threadpool
+ InternalExecutor(SharedResourceType.ThreadPool),
+ // pulsar-client-scheduled threadpool
+ ScheduledExecutor(SharedResourceType.ThreadPool),
+ // pulsar-lookup threadpool
+ LookupExecutor(SharedResourceType.ThreadPool),
+ // DNS resolver and cache that must be shared together with eventLoopGroup
+ DnsResolver(SharedResourceType.DnsResolver);
+
+ private final SharedResourceType type;
+
+ SharedResource(SharedResourceType type) {
+ this.type = type;
+ }
+
+ public SharedResourceType getType() {
+ return type;
+ }
+ }
+
+ enum SharedResourceType {
+ EventLoopGroup,
+ ThreadPool,
+ Timer,
+ DnsResolver;
+ }
+
+ /**
+ * Checks if a resource type is contained in the shared resources instance.
+ *
+ * @param sharedResource the type of resource to check
+ * @return true if the resource type is contained in this instance, false otherwise
+ */
+ boolean contains(SharedResource sharedResource);
+
+ /**
+ * Gets all resource types contained in this shared resources instance.
+ *
+ * @return set of resource types available in this instance
+ */
+ Set<SharedResource> getSharedResources();
+
+ /**
+ * Creates a new builder for constructing instances of {@link PulsarClientSharedResources}.
+ *
+ * @return a new {@link PulsarClientSharedResourcesBuilder} which can be used to configure
+ * and build shared resources for use across multiple PulsarClient instances.
+ */
+ static PulsarClientSharedResourcesBuilder builder() {
+ return DefaultImplementation.getDefaultImplementation().newSharedResourcesBuilder();
+ }
+
+ /**
+ * Closes this instance of PulsarClientSharedResources and releases all associated resources.
+ * All PulsarClient instances using these shared resources must be closed before calling this method.
+ *
+ * @throws PulsarClientException if there is an error while closing the shared resources
+ */
+ @Override
+ void close() throws PulsarClientException;
+}
diff --git a/pulsar-client-api/src/main/java/org/apache/pulsar/client/api/PulsarClientSharedResourcesBuilder.java b/pulsar-client-api/src/main/java/org/apache/pulsar/client/api/PulsarClientSharedResourcesBuilder.java
new file mode 100644
index 00000000000..3a94d5352b5
--- /dev/null
+++ b/pulsar-client-api/src/main/java/org/apache/pulsar/client/api/PulsarClientSharedResourcesBuilder.java
@@ -0,0 +1,122 @@
+/*
+ * 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.client.api;
+
+import static org.apache.pulsar.client.api.PulsarClientSharedResources.SharedResource;
+import java.util.Collection;
+import java.util.function.Consumer;
+
+/**
+ * Builder for configuring and creating {@link PulsarClientSharedResources}.
+ * <p>
+ * Shared resources allow multiple PulsarClient instances to reuse common components
+ * (for example executors, timers, DNS resolver, event loop, connection pool),
+ * reducing memory footprint and thread count when many clients are created in the same JVM.
+ * This allows creating a large number of PulsarClient instances in a single JVM without wasting resources.
+ * Sharing the DNS resolver and cache will help reduce the number of DNS lookups performed by the clients
+ * and improve latency and reduce load on the DNS servers.
+ * <p>
+ * Typical usage:
+ * <pre>{@code
+ * PulsarClientSharedResources shared = PulsarClientSharedResources
+ * .builder()
+ * // shared resources can be configured using a ClientBuilder
+ * .configureResources(PulsarClient.builder().ioThreads(4))
+ * .build();
+ *
+ * PulsarClient client = PulsarClient.builder()
+ * .sharedResources(shared)
+ * .serviceUrl("pulsar://localhost:6650")
+ * .build();
+ *
+ * client.close();
+ * // it's necessary to close the shared resources after usage
+ * shared.close();
+ * }</pre>
+ */
+public interface PulsarClientSharedResourcesBuilder {
+ /**
+ * Optionally limits the shared resource types to be created and exposed by the resulting
+ * {@link PulsarClientSharedResources}. If this method isn't called, resources for all resource types
+ * will be created and exposed.
+ * Calling this method is additive and the builder will accumulate the shared resource types from all method calls.
+ *
+ * @param sharedResource one or more resource types to include
+ * @return this builder
+ */
+ PulsarClientSharedResourcesBuilder resourceTypes(SharedResource... sharedResource);
+
+ /**
+ * @param sharedResource one or more resource types to include
+ * @return this builder
+ * @see {@link #resourceTypes(SharedResource...)}
+ */
+ PulsarClientSharedResourcesBuilder resourceTypes(Collection<SharedResource> sharedResource);
+
+ /**
+ * Share only the configured resources. It's not allowed to use {@link #resourceTypes(SharedResource...)} when
+ * this method is called.
+ * @return this builder instance for method chaining
+ */
+ PulsarClientSharedResourcesBuilder shareConfigured();
+
+ /**
+ * Builds the {@link PulsarClientSharedResources} instance. It is necessary to call
+ * {@link PulsarClientSharedResources#close()} on the instance to release resources. All clients must be closed
+ * before closing the shared resources.
+ *
+ * @return the shared resources to be used in {@link ClientBuilder#sharedResources(PulsarClientSharedResources)}
+ */
+ PulsarClientSharedResources build();
+
+ /**
+ * Configures a thread pool for the specified shared resource type.
+ *
+ * @param sharedResource the type of shared resource to configure
+ * @param configurer a consumer that configures the thread pool settings
+ * @return this builder instance for method chaining
+ */
+ PulsarClientSharedResourcesBuilder configureThreadPool(SharedResource sharedResource,
+ Consumer<ThreadPoolConfig<?>> configurer);
+
+ /**
+ * Configures the event loop group settings.
+ *
+ * @param configurer a consumer that configures the event loop group settings
+ * @return this builder instance for method chaining
+ */
+ PulsarClientSharedResourcesBuilder configureEventLoop(Consumer<EventLoopGroupConfig> configurer);
+
+ /**
+ * Configures the DNS resolver settings.
+ *
+ * @param configurer a consumer that configures the DNS resolver settings
+ * @return this builder instance for method chaining
+ */
+ PulsarClientSharedResourcesBuilder configureDnsResolver(Consumer<DnsResolverConfig> configurer);
+
+ /**
+ * Configures the timer settings.
+ *
+ * @param configurer a consumer that configures the timer settings
+ * @return this builder instance for method chaining
+ */
+ PulsarClientSharedResourcesBuilder configureTimer(Consumer<TimerConfig> configurer);
+
+}
\ No newline at end of file
diff --git a/pulsar-client-api/src/main/java/org/apache/pulsar/client/api/ThreadPoolConfig.java b/pulsar-client-api/src/main/java/org/apache/pulsar/client/api/ThreadPoolConfig.java
new file mode 100644
index 00000000000..dd57a6cebfe
--- /dev/null
+++ b/pulsar-client-api/src/main/java/org/apache/pulsar/client/api/ThreadPoolConfig.java
@@ -0,0 +1,48 @@
+/*
+ * 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.client.api;
+
+/**
+ * Configuration interface for thread pool settings.
+ */
+public interface ThreadPoolConfig<T extends ThreadPoolConfig<T>> {
+ /**
+ * Sets the name of the thread pool.
+ *
+ * @param name the name to set for the thread pool
+ * @return this config instance for method chaining
+ */
+ T name(String name);
+
+ /**
+ * Sets the number of threads in the thread pool.
+ *
+ * @param numberOfThreads the number of threads to use
+ * @return this config instance for method chaining
+ */
+ T numberOfThreads(int numberOfThreads);
+
+ /**
+ * Sets whether the threads should be daemon threads.
+ *
+ * @param daemon true if the threads should be daemon threads, false otherwise
+ * @return this config instance for method chaining
+ */
+ T daemon(boolean daemon);
+}
diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/util/ScheduledExecutorProvider.java b/pulsar-client-api/src/main/java/org/apache/pulsar/client/api/TimerConfig.java
similarity index 54%
copy from pulsar-client/src/main/java/org/apache/pulsar/client/util/ScheduledExecutorProvider.java
copy to pulsar-client-api/src/main/java/org/apache/pulsar/client/api/TimerConfig.java
index 3ebd41ebe56..1c1235704df 100644
--- a/pulsar-client/src/main/java/org/apache/pulsar/client/util/ScheduledExecutorProvider.java
+++ b/pulsar-client-api/src/main/java/org/apache/pulsar/client/api/TimerConfig.java
@@ -16,21 +16,28 @@
* specific language governing permissions and limitations
* under the License.
*/
-package org.apache.pulsar.client.util;
+package org.apache.pulsar.client.api;
-import java.util.concurrent.ExecutorService;
-import java.util.concurrent.Executors;
-import lombok.extern.slf4j.Slf4j;
+import java.util.concurrent.TimeUnit;
-@Slf4j
-public class ScheduledExecutorProvider extends ExecutorProvider {
-
- public ScheduledExecutorProvider(int numThreads, String poolName) {
- super(numThreads, poolName);
- }
+/**
+ * Configuration interface for timer settings.
+ */
+public interface TimerConfig {
+ /**
+ * Sets the name of the timer.
+ *
+ * @param name the name to set for the timer
+ * @return this config instance for method chaining
+ */
+ TimerConfig name(String name);
- @Override
- protected ExecutorService createExecutor(ExtendedThreadFactory threadFactory) {
- return Executors.newSingleThreadScheduledExecutor(threadFactory);
- }
+ /**
+ * Sets the tick duration for the timer.
+ *
+ * @param tickDuration the duration of each tick
+ * @param timeUnit the time unit for the tick duration
+ * @return this config instance for method chaining
+ */
+ TimerConfig tickDuration(long tickDuration, TimeUnit timeUnit);
}
diff --git a/pulsar-client-api/src/main/java/org/apache/pulsar/client/internal/PulsarClientImplementationBinding.java b/pulsar-client-api/src/main/java/org/apache/pulsar/client/internal/PulsarClientImplementationBinding.java
index 8fd05bff265..b5f2a3a468e 100644
--- a/pulsar-client-api/src/main/java/org/apache/pulsar/client/internal/PulsarClientImplementationBinding.java
+++ b/pulsar-client-api/src/main/java/org/apache/pulsar/client/internal/PulsarClientImplementationBinding.java
@@ -36,6 +36,7 @@ import org.apache.pulsar.client.api.ClientBuilder;
import org.apache.pulsar.client.api.MessageId;
import org.apache.pulsar.client.api.MessagePayloadFactory;
import org.apache.pulsar.client.api.PulsarClientException;
+import org.apache.pulsar.client.api.PulsarClientSharedResourcesBuilder;
import org.apache.pulsar.client.api.Schema;
import org.apache.pulsar.client.api.TopicMessageId;
import org.apache.pulsar.client.api.schema.GenericRecord;
@@ -255,4 +256,6 @@ public interface PulsarClientImplementationBinding {
Map<String, String> propertiesValue);
TopicMessageId newTopicMessageId(String topic, MessageId messageId);
+
+ PulsarClientSharedResourcesBuilder newSharedResourcesBuilder();
}
diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/AutoClusterFailover.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/AutoClusterFailover.java
index 844d1e2d253..7615ec9f810 100644
--- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/AutoClusterFailover.java
+++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/AutoClusterFailover.java
@@ -20,6 +20,7 @@ package org.apache.pulsar.client.impl;
import static org.apache.pulsar.common.util.Runnables.catchingAndLoggingThrowables;
import com.google.common.base.Strings;
+import io.netty.resolver.AddressResolver;
import java.io.IOException;
import java.net.InetSocketAddress;
import java.net.Socket;
@@ -63,6 +64,7 @@ public class AutoClusterFailover implements ServiceUrlProvider {
private final long intervalMs;
private static final int TIMEOUT = 30_000;
private final PulsarServiceNameResolver resolver;
+ private AddressResolver<InetSocketAddress> addressResolver;
private AutoClusterFailover(AutoClusterFailoverBuilderImpl builder) {
this.primary = builder.primary;
@@ -86,6 +88,7 @@ public class AutoClusterFailover implements ServiceUrlProvider {
@Override
public void initialize(PulsarClient client) {
this.pulsarClient = (PulsarClientImpl) client;
+ this.addressResolver = pulsarClient.getAddressResolver();
ClientConfigurationData config = pulsarClient.getConfiguration();
if (config != null) {
this.primaryAuthentication = config.getAuthentication();
@@ -128,8 +131,14 @@ public class AutoClusterFailover implements ServiceUrlProvider {
try {
resolver.updateServiceUrl(url);
InetSocketAddress endpoint = resolver.resolveHost();
+ if (!endpoint.isUnresolved()) {
+ // create an unresolved endpoint to ensure that DNS lookup is performed again
+ endpoint = InetSocketAddress.createUnresolved(endpoint.getHostString(), endpoint.getPort());
+ }
+ // Use Netty's DNS resolver and settings to resolve the host name
+ InetSocketAddress resolvedEndpoint = addressResolver.resolve(endpoint).get(TIMEOUT, TimeUnit.MILLISECONDS);
try (Socket socket = new Socket()) {
- socket.connect(new InetSocketAddress(endpoint.getHostName(), endpoint.getPort()), TIMEOUT);
+ socket.connect(resolvedEndpoint, TIMEOUT);
}
return true;
} catch (Exception e) {
diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ClientBuilderImpl.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ClientBuilderImpl.java
index d031c8e1710..5b3a52d5e42 100644
--- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ClientBuilderImpl.java
+++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ClientBuilderImpl.java
@@ -34,6 +34,7 @@ import org.apache.pulsar.client.api.ProxyProtocol;
import org.apache.pulsar.client.api.PulsarClient;
import org.apache.pulsar.client.api.PulsarClientException;
import org.apache.pulsar.client.api.PulsarClientException.UnsupportedAuthenticationException;
+import org.apache.pulsar.client.api.PulsarClientSharedResources;
import org.apache.pulsar.client.api.ServiceUrlProvider;
import org.apache.pulsar.client.api.SizeUnit;
import org.apache.pulsar.client.impl.auth.AuthenticationDisabled;
@@ -43,7 +44,9 @@ import org.apache.pulsar.common.tls.InetAddressUtils;
import org.apache.pulsar.common.util.DefaultPulsarSslFactory;
public class ClientBuilderImpl implements ClientBuilder {
+ private static final long serialVersionUID = 1L;
ClientConfigurationData conf;
+ private transient PulsarClientSharedResourcesImpl sharedResources;
public ClientBuilderImpl() {
this(new ClientConfigurationData());
@@ -68,12 +71,19 @@ public class ClientBuilderImpl implements ClientBuilder {
if (conf.getAuthentication() == null || conf.getAuthentication() == AuthenticationDisabled.INSTANCE) {
setAuthenticationFromPropsIfAvailable(conf);
}
- return new PulsarClientImpl(conf);
+ PulsarClientImpl.PulsarClientImplBuilder instanceBuilder = PulsarClientImpl.builder();
+ instanceBuilder.conf(conf);
+ if (sharedResources != null) {
+ sharedResources.applyTo(instanceBuilder);
+ }
+ return instanceBuilder.build();
}
@Override
public ClientBuilder clone() {
- return new ClientBuilderImpl(conf.clone());
+ ClientBuilderImpl clientBuilder = new ClientBuilderImpl(conf.clone());
+ clientBuilder.sharedResources = sharedResources;
+ return clientBuilder;
}
@Override
@@ -486,6 +496,12 @@ public class ClientBuilderImpl implements ClientBuilder {
return this;
}
+ @Override
+ public ClientBuilder sharedResources(PulsarClientSharedResources sharedResources) {
+ this.sharedResources = (PulsarClientSharedResourcesImpl) sharedResources;
+ return this;
+ }
+
@Override
public ClientBuilder lookupProperties(Map<String, String> properties) {
conf.setLookupProperties(properties);
diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConnectionPool.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConnectionPool.java
index f8412b5bf29..fea599ea408 100644
--- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConnectionPool.java
+++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConnectionPool.java
@@ -191,7 +191,7 @@ public class ConnectionPool implements AutoCloseable {
private Supplier<AddressResolver<InetSocketAddress>> createAddressResolver(ClientConfigurationData conf,
EventLoopGroup eventLoopGroup) {
if (dnsResolverGroup == null) {
- dnsResolverGroup = new DnsResolverGroupImpl(eventLoopGroup, conf);
+ dnsResolverGroup = new DnsResolverGroupImpl(conf);
}
return () -> dnsResolverGroup.createAddressResolver(eventLoopGroup);
}
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 c52a70c6cd7..032069786a4 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,8 @@ import io.netty.handler.codec.http.HttpResponse;
import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.charset.StandardCharsets;
+import java.util.Collections;
+import java.util.LinkedHashMap;
import java.util.Map;
import java.util.Objects;
import java.util.concurrent.ExecutionException;
@@ -59,27 +61,27 @@ public class ControlledClusterFailover implements ServiceUrlProvider {
private static final int DEFAULT_CONNECT_TIMEOUT_IN_SECONDS = 10;
private static final int DEFAULT_READ_TIMEOUT_IN_SECONDS = 30;
private static final int DEFAULT_MAX_REDIRECTS = 20;
+ private final Map<String, String> headers;
+ private final String urlProvider;
private PulsarClientImpl pulsarClient;
private volatile String currentPulsarServiceUrl;
private volatile ControlledConfiguration currentControlledConfiguration;
private final ScheduledExecutorService executor;
private final long interval;
- private final AsyncHttpClient httpClient;
- private final BoundRequestBuilder requestBuilder;
+ private AsyncHttpClient httpClient;
+ private BoundRequestBuilder requestBuilder;
private ControlledClusterFailover(ControlledClusterFailoverBuilderImpl builder) throws IOException {
this.currentPulsarServiceUrl = builder.defaultServiceUrl;
this.interval = builder.interval;
this.executor = Executors.newSingleThreadScheduledExecutor(
new ExecutorProvider.ExtendedThreadFactory("pulsar-service-provider"));
-
- this.httpClient = buildHttpClient();
- this.requestBuilder = httpClient.prepareGet(builder.urlProvider)
- .addHeader("Accept", "application/json");
-
+ urlProvider = builder.urlProvider;
if (builder.header != null && !builder.header.isEmpty()) {
- builder.header.forEach(requestBuilder::addHeader);
+ headers = new LinkedHashMap<>(builder.header);
+ } else {
+ headers = Collections.emptyMap();
}
}
@@ -101,6 +103,8 @@ public class ControlledClusterFailover implements ServiceUrlProvider {
&& super.keepAlive(remoteAddress, ahcRequest, request, response);
}
});
+ confBuilder.setNettyTimer(pulsarClient.timer());
+ confBuilder.setEventLoopGroup(pulsarClient.eventLoopGroup());
AsyncHttpClientConfig config = confBuilder.build();
return new DefaultAsyncHttpClient(config);
}
@@ -108,6 +112,12 @@ public class ControlledClusterFailover implements ServiceUrlProvider {
@Override
public void initialize(PulsarClient client) {
this.pulsarClient = (PulsarClientImpl) client;
+ this.httpClient = buildHttpClient();
+ this.requestBuilder = httpClient.prepareGet(urlProvider)
+ // share Pulsar client DNS resolver and cache
+ .setNameResolver(pulsarClient.getNameResolver())
+ .addHeader("Accept", "application/json");
+ headers.forEach(requestBuilder::addHeader);
// start to check service url every 30 seconds
this.executor.scheduleAtFixedRate(catchingAndLoggingThrowables(() -> {
diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/DnsResolverGroupImpl.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/DnsResolverGroupImpl.java
index 61af7968f81..0702e6324ba 100644
--- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/DnsResolverGroupImpl.java
+++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/DnsResolverGroupImpl.java
@@ -41,36 +41,68 @@ import org.apache.pulsar.common.util.netty.EventLoopUtil;
public class DnsResolverGroupImpl implements AutoCloseable {
private final DnsAddressResolverGroup dnsAddressResolverGroup;
- public DnsResolverGroupImpl(EventLoopGroup eventLoopGroup, ClientConfigurationData conf) {
+ public DnsResolverGroupImpl(ClientConfigurationData conf) {
Optional<InetSocketAddress> bindAddress = Optional.ofNullable(conf.getDnsLookupBindAddress())
.map(addr -> new InetSocketAddress(addr, conf.getDnsLookupBindPort()));
Optional<DnsServerAddressStreamProvider> dnsServerAddresses = Optional.ofNullable(conf.getDnsServerAddresses())
.filter(Predicate.not(List::isEmpty))
.map(SequentialDnsServerAddressStreamProvider::new);
- this.dnsAddressResolverGroup = createAddressResolverGroup(eventLoopGroup, bindAddress, dnsServerAddresses);
+ this.dnsAddressResolverGroup = createAddressResolverGroup(bindAddress, dnsServerAddresses);
}
- public DnsResolverGroupImpl(EventLoopGroup eventLoopGroup, Optional<InetSocketAddress> bindAddress,
- Optional<DnsServerAddressStreamProvider> dnsServerAddresses) {
- this.dnsAddressResolverGroup = createAddressResolverGroup(eventLoopGroup, bindAddress, dnsServerAddresses);
+ public DnsResolverGroupImpl(PulsarClientSharedResourcesBuilderImpl.DnsResolverResourceConfig dnsConfig) {
+ this.dnsAddressResolverGroup = createAddressResolverGroup(dnsConfig);
}
- private static DnsAddressResolverGroup createAddressResolverGroup(EventLoopGroup eventLoopGroup,
- Optional<InetSocketAddress> bindAddress,
+ private DnsAddressResolverGroup createAddressResolverGroup(
+ PulsarClientSharedResourcesBuilderImpl.DnsResolverResourceConfig dnsConfig) {
+ DnsNameResolverBuilder dnsNameResolverBuilder = new DnsNameResolverBuilder()
+ .traceEnabled(dnsConfig.traceEnabled)
+ .channelType(EventLoopUtil.getDatagramChannelClass());
+ if (dnsConfig.tcpFallbackEnabled || dnsConfig.tcpFallbackOnTimeoutEnabled) {
+ dnsNameResolverBuilder.socketChannelType(EventLoopUtil.getClientSocketChannelClass(),
+ dnsConfig.tcpFallbackOnTimeoutEnabled);
+ }
+ dnsNameResolverBuilder
+ .ttl(dnsConfig.minTtl, dnsConfig.maxTtl)
+ .negativeTtl(dnsConfig.negativeTtl);
+ if (dnsConfig.queryTimeoutMillis > -1L) {
+ dnsNameResolverBuilder.queryTimeoutMillis(dnsConfig.queryTimeoutMillis);
+ }
+ if (dnsConfig.ndots > -1) {
+ dnsNameResolverBuilder.ndots(dnsConfig.ndots);
+ }
+ if (dnsConfig.localAddress != null) {
+ dnsNameResolverBuilder.localAddress(dnsConfig.localAddress);
+ }
+ if (dnsConfig.serverAddresses != null) {
+ Optional.ofNullable(dnsConfig.serverAddresses)
+ .map(SequentialDnsServerAddressStreamProvider::new)
+ .ifPresent(dnsServerAddressStreamProvider -> {
+ dnsNameResolverBuilder.nameServerProvider(dnsServerAddressStreamProvider);
+ });
+ }
+ if (dnsConfig.searchDomains != null) {
+ dnsNameResolverBuilder.searchDomains(dnsConfig.searchDomains);
+ }
+ return new DnsAddressResolverGroup(dnsNameResolverBuilder);
+ }
+
+ private static DnsAddressResolverGroup createAddressResolverGroup(Optional<InetSocketAddress> bindAddress,
Optional<DnsServerAddressStreamProvider>
dnsServerAddresses) {
- DnsNameResolverBuilder dnsNameResolverBuilder = createDnsNameResolverBuilder(eventLoopGroup);
+ DnsNameResolverBuilder dnsNameResolverBuilder = createDnsNameResolverBuilder();
bindAddress.ifPresent(dnsNameResolverBuilder::localAddress);
dnsServerAddresses.ifPresent(dnsNameResolverBuilder::nameServerProvider);
return new DnsAddressResolverGroup(dnsNameResolverBuilder);
}
- private static DnsNameResolverBuilder createDnsNameResolverBuilder(EventLoopGroup eventLoopGroup) {
+ private static DnsNameResolverBuilder createDnsNameResolverBuilder() {
DnsNameResolverBuilder dnsNameResolverBuilder = new DnsNameResolverBuilder()
.traceEnabled(true)
- .channelType(EventLoopUtil.getDatagramChannelClass(eventLoopGroup))
- .socketChannelType(EventLoopUtil.getClientSocketChannelClass(eventLoopGroup), true);
+ .channelType(EventLoopUtil.getDatagramChannelClass())
+ .socketChannelType(EventLoopUtil.getClientSocketChannelClass(), true);
DnsResolverUtil.applyJdkDnsCacheSettings(dnsNameResolverBuilder);
return dnsNameResolverBuilder;
}
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 633ca606388..72fcf82b859 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
@@ -21,9 +21,12 @@ package org.apache.pulsar.client.impl;
import io.netty.channel.EventLoopGroup;
import io.netty.handler.codec.http.HttpRequest;
import io.netty.handler.codec.http.HttpResponse;
+import io.netty.resolver.NameResolver;
+import io.netty.util.Timer;
import java.io.Closeable;
import java.io.IOException;
import java.net.HttpURLConnection;
+import java.net.InetAddress;
import java.net.InetSocketAddress;
import java.net.URI;
import java.net.URL;
@@ -65,16 +68,20 @@ public class HttpClient implements Closeable {
protected final AsyncHttpClient httpClient;
protected final ServiceNameResolver serviceNameResolver;
+ private final NameResolver<InetAddress> nameResolver;
protected final Authentication authentication;
protected final ClientConfigurationData clientConf;
protected ScheduledExecutorService executorService;
protected PulsarSslFactory sslFactory;
- protected HttpClient(ClientConfigurationData conf, EventLoopGroup eventLoopGroup) throws PulsarClientException {
+ protected HttpClient(ClientConfigurationData conf, EventLoopGroup eventLoopGroup, Timer timer,
+ NameResolver<InetAddress> nameResolver)
+ throws PulsarClientException {
this.authentication = conf.getAuthentication();
this.clientConf = conf;
this.serviceNameResolver = new PulsarServiceNameResolver(conf.getServiceUrlQuarantineInitDurationMs(),
conf.getServiceUrlQuarantineMaxDurationMs());
+ this.nameResolver = nameResolver;
this.serviceNameResolver.updateServiceUrl(conf.getServiceUrl());
DefaultAsyncHttpClientConfig.Builder confBuilder = new DefaultAsyncHttpClientConfig.Builder();
@@ -128,6 +135,7 @@ public class HttpClient implements Closeable {
}
}
confBuilder.setEventLoopGroup(eventLoopGroup);
+ confBuilder.setNettyTimer(timer);
AsyncHttpClientConfig config = confBuilder.build();
httpClient = new DefaultAsyncHttpClient(config);
@@ -184,7 +192,9 @@ public class HttpClient implements Closeable {
// auth complete, use a new builder
BoundRequestBuilder builder = httpClient.prepareGet(requestUrl)
- .setHeader("Accept", "application/json");
+ // share the DNS resolver and cache with Pulsar client
+ .setNameResolver(nameResolver)
+ .setHeader("Accept", "application/json");
if (authData.hasDataForHttp()) {
Set<Entry<String, String>> headers;
diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/HttpLookupService.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/HttpLookupService.java
index 4a5557fa869..08c9956b5be 100644
--- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/HttpLookupService.java
+++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/HttpLookupService.java
@@ -19,7 +19,10 @@
package org.apache.pulsar.client.impl;
import io.netty.channel.EventLoopGroup;
+import io.netty.resolver.NameResolver;
+import io.netty.util.Timer;
import io.opentelemetry.api.common.Attributes;
+import java.net.InetAddress;
import java.net.InetSocketAddress;
import java.net.URI;
import java.nio.ByteBuffer;
@@ -65,10 +68,17 @@ public class HttpLookupService implements LookupService {
private final LatencyHistogram histoGetSchema;
private final LatencyHistogram histoListTopics;
+ @Deprecated
public HttpLookupService(InstrumentProvider instrumentProvider, ClientConfigurationData conf,
- EventLoopGroup eventLoopGroup)
+ EventLoopGroup eventLoopGroup) throws PulsarClientException {
+ this(instrumentProvider, conf, eventLoopGroup, null, null);
+ }
+
+ public HttpLookupService(InstrumentProvider instrumentProvider, ClientConfigurationData conf,
+ EventLoopGroup eventLoopGroup, Timer timer,
+ NameResolver<InetAddress> nameResolver)
throws PulsarClientException {
- this.httpClient = new HttpClient(conf, eventLoopGroup);
+ this.httpClient = new HttpClient(conf, eventLoopGroup, timer, nameResolver);
this.useTls = conf.isUseTls();
this.listenerName = conf.getListenerName();
diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/PulsarClientImpl.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/PulsarClientImpl.java
index d38d35926a4..6ffdfa55a9b 100644
--- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/PulsarClientImpl.java
+++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/PulsarClientImpl.java
@@ -25,9 +25,10 @@ import com.google.common.cache.CacheLoader;
import com.google.common.cache.LoadingCache;
import io.netty.channel.EventLoopGroup;
import io.netty.resolver.AddressResolver;
-import io.netty.util.HashedWheelTimer;
+import io.netty.resolver.NameResolver;
import io.netty.util.Timer;
import java.io.IOException;
+import java.net.InetAddress;
import java.net.InetSocketAddress;
import java.time.Clock;
import java.time.Duration;
@@ -44,7 +45,6 @@ import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
-import java.util.concurrent.ThreadFactory;
import java.util.concurrent.ThreadLocalRandom;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
@@ -97,7 +97,7 @@ import org.apache.pulsar.common.topics.TopicsPatternFactory;
import org.apache.pulsar.common.util.Backoff;
import org.apache.pulsar.common.util.BackoffBuilder;
import org.apache.pulsar.common.util.FutureUtil;
-import org.apache.pulsar.common.util.netty.EventLoopUtil;
+import org.apache.pulsar.common.util.netty.DnsResolverUtil;
import org.jspecify.annotations.Nullable;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -131,6 +131,7 @@ public class PulsarClientImpl implements PulsarClient {
private final boolean createdEventLoopGroup;
private final boolean createdCnxPool;
private final DnsResolverGroupImpl dnsResolverGroupLocalInstance;
+ @Getter
private final AddressResolver<InetSocketAddress> addressResolver;
public enum State {
@@ -199,12 +200,12 @@ public class PulsarClientImpl implements PulsarClient {
}
@Builder(builderClassName = "PulsarClientImplBuilder")
- private PulsarClientImpl(ClientConfigurationData conf, EventLoopGroup eventLoopGroup, ConnectionPool connectionPool,
- Timer timer, ExecutorProvider externalExecutorProvider,
- ExecutorProvider internalExecutorProvider,
- ScheduledExecutorProvider scheduledExecutorProvider,
- ExecutorProvider lookupExecutorProvider,
- DnsResolverGroupImpl dnsResolverGroup) throws PulsarClientException {
+ PulsarClientImpl(ClientConfigurationData conf, EventLoopGroup eventLoopGroup, ConnectionPool connectionPool,
+ Timer timer, ExecutorProvider externalExecutorProvider,
+ ExecutorProvider internalExecutorProvider,
+ ScheduledExecutorProvider scheduledExecutorProvider,
+ ExecutorProvider lookupExecutorProvider,
+ DnsResolverGroupImpl dnsResolverGroup) throws PulsarClientException {
EventLoopGroup eventLoopGroupReference = null;
ConnectionPool connectionPoolReference = null;
@@ -222,22 +223,26 @@ public class PulsarClientImpl implements PulsarClient {
this.createdExecutorProviders = externalExecutorProvider == null;
this.createdScheduledProviders = scheduledExecutorProvider == null;
this.createdLookupProviders = lookupExecutorProvider == null;
- eventLoopGroupReference = eventLoopGroup != null ? eventLoopGroup : getEventLoopGroup(conf);
+ eventLoopGroupReference = eventLoopGroup != null ? eventLoopGroup :
+ PulsarClientResourcesConfigurer.createEventLoopGroup(conf);
this.eventLoopGroup = eventLoopGroupReference;
this.instrumentProvider = new InstrumentProvider(conf.getOpenTelemetry());
clientClock = conf.getClock();
conf.getAuthentication().start();
this.scheduledExecutorProvider = scheduledExecutorProvider != null ? scheduledExecutorProvider :
- new ScheduledExecutorProvider(conf.getNumIoThreads(), "pulsar-client-scheduled");
+ PulsarClientResourcesConfigurer.createScheduledExecutorProvider(conf);
if (connectionPool != null) {
connectionPoolReference = connectionPool;
dnsResolverGroupLocalInstance = null;
- addressResolver = null;
+ addressResolver = dnsResolverGroup != null
+ ? dnsResolverGroup.createAddressResolver(eventLoopGroupReference) : null;
} else {
DnsResolverGroupImpl dnsResolverGroupReference;
if (dnsResolverGroup == null) {
dnsResolverGroupReference =
- dnsResolverGroupLocalInstance = new DnsResolverGroupImpl(eventLoopGroupReference, conf);
+ dnsResolverGroupLocalInstance =
+ PulsarClientResourcesConfigurer.createDnsResolverGroup(conf
+ );
} else {
dnsResolverGroupReference = dnsResolverGroup;
dnsResolverGroupLocalInstance = null;
@@ -254,24 +259,25 @@ public class PulsarClientImpl implements PulsarClient {
}
this.cnxPool = connectionPoolReference;
this.externalExecutorProvider = externalExecutorProvider != null ? externalExecutorProvider :
- new ExecutorProvider(conf.getNumListenerThreads(), "pulsar-external-listener");
+ PulsarClientResourcesConfigurer.createExternalExecutorProvider(conf);
this.internalExecutorProvider = internalExecutorProvider != null ? internalExecutorProvider :
- new ExecutorProvider(conf.getNumIoThreads(), "pulsar-client-internal");
+ PulsarClientResourcesConfigurer.createInternalExecutorProvider(conf);
this.lookupExecutorProvider = lookupExecutorProvider != null ? lookupExecutorProvider :
- new ExecutorProvider(1, "pulsar-client-lookup");
+ PulsarClientResourcesConfigurer.createLookupExecutorProvider();
+ if (timer == null) {
+ this.timer = PulsarClientResourcesConfigurer.createTimer();
+ needStopTimer = true;
+ } else {
+ this.timer = timer;
+ }
if (conf.getServiceUrl().startsWith("http")) {
- lookup = new HttpLookupService(instrumentProvider, conf, this.eventLoopGroup);
+ lookup = new HttpLookupService(instrumentProvider, conf, this.eventLoopGroup, this.timer,
+ getNameResolver());
} else {
lookup = new BinaryProtoLookupService(this, conf.getServiceUrl(), conf.getListenerName(),
conf.isUseTls(), this.scheduledExecutorProvider.getExecutor(),
this.lookupExecutorProvider.getExecutor());
}
- if (timer == null) {
- this.timer = new HashedWheelTimer(getThreadFactory("pulsar-timer"), 1, TimeUnit.MILLISECONDS);
- needStopTimer = true;
- } else {
- this.timer = timer;
- }
if (conf.getServiceUrlProvider() != null) {
conf.getServiceUrlProvider().initialize(this);
@@ -1211,7 +1217,7 @@ public class PulsarClientImpl implements PulsarClient {
public LookupService createLookup(String url) throws PulsarClientException {
if (url.startsWith("http")) {
- return new HttpLookupService(instrumentProvider, conf, eventLoopGroup);
+ return new HttpLookupService(instrumentProvider, conf, eventLoopGroup, timer, getNameResolver());
} else {
return new BinaryProtoLookupService(this, url, conf.getListenerName(), conf.isUseTls(),
externalExecutorProvider.getExecutor());
@@ -1299,15 +1305,6 @@ public class PulsarClientImpl implements PulsarClient {
});
}
- private static EventLoopGroup getEventLoopGroup(ClientConfigurationData conf) {
- ThreadFactory threadFactory = getThreadFactory("pulsar-client-io");
- return EventLoopUtil.newEventLoopGroup(conf.getNumIoThreads(), conf.isEnableBusyWait(), threadFactory);
- }
-
- private static ThreadFactory getThreadFactory(String poolName) {
- return new ExecutorProvider.ExtendedThreadFactory(poolName, Thread.currentThread().isDaemon());
- }
-
void cleanupProducer(ProducerBase<?> producer) {
producers.remove(producer);
}
@@ -1415,4 +1412,8 @@ public class PulsarClientImpl implements PulsarClient {
public TransactionBuilder newTransaction() {
return new TransactionBuilderImpl(this, tcClient);
}
+
+ NameResolver<InetAddress> getNameResolver() {
+ return DnsResolverUtil.adaptToNameResolver(addressResolver);
+ }
}
diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/PulsarClientImplementationBindingImpl.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/PulsarClientImplementationBindingImpl.java
index 6fe95140967..0351477985f 100644
--- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/PulsarClientImplementationBindingImpl.java
+++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/PulsarClientImplementationBindingImpl.java
@@ -37,6 +37,7 @@ import org.apache.pulsar.client.api.MessageId;
import org.apache.pulsar.client.api.MessageIdAdv;
import org.apache.pulsar.client.api.MessagePayloadFactory;
import org.apache.pulsar.client.api.PulsarClientException;
+import org.apache.pulsar.client.api.PulsarClientSharedResourcesBuilder;
import org.apache.pulsar.client.api.Schema;
import org.apache.pulsar.client.api.TopicMessageId;
import org.apache.pulsar.client.api.schema.GenericRecord;
@@ -403,4 +404,9 @@ public final class PulsarClientImplementationBindingImpl implements PulsarClient
}
return new TopicMessageIdImpl(topic, messageIdAdv);
}
+
+ @Override
+ public PulsarClientSharedResourcesBuilder newSharedResourcesBuilder() {
+ return new PulsarClientSharedResourcesBuilderImpl();
+ }
}
diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/PulsarClientResourcesConfigurer.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/PulsarClientResourcesConfigurer.java
new file mode 100644
index 00000000000..28f97e92159
--- /dev/null
+++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/PulsarClientResourcesConfigurer.java
@@ -0,0 +1,149 @@
+/*
+ * 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.client.impl;
+
+import io.netty.channel.EventLoopGroup;
+import io.netty.util.HashedWheelTimer;
+import java.util.concurrent.ThreadFactory;
+import java.util.concurrent.TimeUnit;
+import org.apache.commons.lang3.StringUtils;
+import org.apache.pulsar.client.impl.conf.ClientConfigurationData;
+import org.apache.pulsar.client.util.ExecutorProvider;
+import org.apache.pulsar.client.util.ScheduledExecutorProvider;
+import org.apache.pulsar.common.util.netty.EventLoopUtil;
+
+class PulsarClientResourcesConfigurer {
+ static final String NAME_TIMER = "pulsar-timer";
+ static final String POOL_NAME_LOOKUP_EXECUTOR = "pulsar-client-lookup";
+ static final String POOL_NAME_INTERNAL_EXECUTOR = "pulsar-client-internal";
+ static final String POOL_NAME_LISTENER_EXECUTOR = "pulsar-external-listener";
+ static final String POOL_NAME_SCHEDULED_EXECUTOR = "pulsar-client-scheduled";
+ static final String POOL_NAME_EVENT_LOOP_GROUP = "pulsar-client-io";
+ static final int LOOKUP_EXECUTOR_NUM_THREADS = 1;
+
+ static HashedWheelTimer createTimer() {
+ return createTimerInternal(NAME_TIMER, 1L, TimeUnit.MILLISECONDS);
+ }
+
+ private static HashedWheelTimer createTimerInternal(String poolName, long tickDuration, TimeUnit timeUnit) {
+ return new HashedWheelTimer(createThreadFactory(poolName), tickDuration, timeUnit);
+ }
+
+ static HashedWheelTimer createTimer(PulsarClientSharedResourcesBuilderImpl.TimerResourceConfig resourceConfig) {
+ if (resourceConfig == null) {
+ resourceConfig = new PulsarClientSharedResourcesBuilderImpl.TimerResourceConfig();
+ }
+ return createTimerInternal(resourceConfig.name, resourceConfig.tickDuration,
+ resourceConfig.tickDurationTimeUnit);
+ }
+
+ static ExecutorProvider createLookupExecutorProvider() {
+ return createExecutorProviderInternal(LOOKUP_EXECUTOR_NUM_THREADS, POOL_NAME_LOOKUP_EXECUTOR);
+ }
+
+ static ExecutorProvider createLookupExecutorProviderWithResourceConfig(
+ PulsarClientSharedResourcesBuilderImpl.ThreadPoolResourceConfig resourceConfig) {
+ return createExecutorProviderInternal(resourceConfig, POOL_NAME_LOOKUP_EXECUTOR);
+ }
+
+ private static ExecutorProvider createExecutorProviderInternal(int numThreads, String poolName) {
+ return createExecutorProviderInternal(numThreads, poolName, Thread.currentThread().isDaemon());
+ }
+
+ private static ExecutorProvider createExecutorProviderInternal(int numThreads, String poolName, boolean daemon) {
+ return new ExecutorProvider(numThreads, poolName, daemon);
+ }
+
+ static ExecutorProvider createInternalExecutorProvider(ClientConfigurationData conf) {
+ return createExecutorProviderInternal(conf.getNumIoThreads(), POOL_NAME_INTERNAL_EXECUTOR);
+ }
+
+ static ExecutorProvider createInternalExecutorProviderWithResourceConfig(
+ PulsarClientSharedResourcesBuilderImpl.ThreadPoolResourceConfig resourceConfig) {
+ return createExecutorProviderInternal(resourceConfig, POOL_NAME_INTERNAL_EXECUTOR);
+ }
+
+ static ExecutorProvider createExternalExecutorProvider(ClientConfigurationData conf) {
+ return createExecutorProviderInternal(conf.getNumListenerThreads(), POOL_NAME_LISTENER_EXECUTOR);
+ }
+
+ static ExecutorProvider createExternalExecutorProviderWithResourceConfig(
+ PulsarClientSharedResourcesBuilderImpl.ThreadPoolResourceConfig resourceConfig) {
+ return createExecutorProviderInternal(resourceConfig, POOL_NAME_LISTENER_EXECUTOR);
+ }
+
+ private static ExecutorProvider createExecutorProviderInternal(
+ PulsarClientSharedResourcesBuilderImpl.ThreadPoolResourceConfig resourceConfig, String defaultPoolName) {
+ if (resourceConfig == null) {
+ resourceConfig = new PulsarClientSharedResourcesBuilderImpl.ThreadPoolResourceConfig();
+ }
+ String poolName = StringUtils.isNotBlank(resourceConfig.name) ? resourceConfig.name : defaultPoolName;
+ return createExecutorProviderInternal(resourceConfig.numberOfThreads, poolName, resourceConfig.daemon);
+ }
+
+ static ScheduledExecutorProvider createScheduledExecutorProvider(ClientConfigurationData conf) {
+ return new ScheduledExecutorProvider(conf.getNumIoThreads(), POOL_NAME_SCHEDULED_EXECUTOR);
+ }
+
+ static ScheduledExecutorProvider createScheduledExecutorProviderWithResourceConfig(
+ PulsarClientSharedResourcesBuilderImpl.ThreadPoolResourceConfig resourceConfig) {
+ if (resourceConfig == null) {
+ resourceConfig = new PulsarClientSharedResourcesBuilderImpl.ThreadPoolResourceConfig();
+ }
+ String poolName =
+ StringUtils.isNotBlank(resourceConfig.name) ? resourceConfig.name : POOL_NAME_SCHEDULED_EXECUTOR;
+ return new ScheduledExecutorProvider(resourceConfig.numberOfThreads, poolName, resourceConfig.daemon);
+ }
+
+ static EventLoopGroup createEventLoopGroup(ClientConfigurationData conf) {
+ ThreadFactory threadFactory = createThreadFactory(POOL_NAME_EVENT_LOOP_GROUP);
+ return EventLoopUtil.newEventLoopGroup(conf.getNumIoThreads(), conf.isEnableBusyWait(), threadFactory);
+ }
+
+ static EventLoopGroup createEventLoopGroupWithResourceConfig(
+ PulsarClientSharedResourcesBuilderImpl.EventLoopResourceConfig
+ eventLoopResourceConfig) {
+ if (eventLoopResourceConfig == null) {
+ eventLoopResourceConfig = new PulsarClientSharedResourcesBuilderImpl.EventLoopResourceConfig();
+ }
+ ThreadFactory threadFactory = createThreadFactory(eventLoopResourceConfig.name, eventLoopResourceConfig.daemon);
+ return EventLoopUtil.newEventLoopGroup(eventLoopResourceConfig.numberOfThreads,
+ eventLoopResourceConfig.enableBusyWait, threadFactory);
+ }
+
+ static ThreadFactory createThreadFactory(String poolName) {
+ return createThreadFactory(poolName, Thread.currentThread().isDaemon());
+ }
+
+ static ThreadFactory createThreadFactory(String poolName, boolean daemon) {
+ return new ExecutorProvider.ExtendedThreadFactory(poolName, daemon);
+ }
+
+ static DnsResolverGroupImpl createDnsResolverGroup(ClientConfigurationData conf) {
+ return new DnsResolverGroupImpl(conf);
+ }
+
+ static DnsResolverGroupImpl createDnsResolverGroupWithResourceConfig(
+ PulsarClientSharedResourcesBuilderImpl.DnsResolverResourceConfig resourceConfig) {
+ if (resourceConfig == null) {
+ resourceConfig = new PulsarClientSharedResourcesBuilderImpl.DnsResolverResourceConfig();
+ }
+ return new DnsResolverGroupImpl(resourceConfig);
+ }
+}
diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/PulsarClientSharedResourcesBuilderImpl.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/PulsarClientSharedResourcesBuilderImpl.java
new file mode 100644
index 00000000000..fe9d43dc41e
--- /dev/null
+++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/PulsarClientSharedResourcesBuilderImpl.java
@@ -0,0 +1,283 @@
+/*
+ * 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.client.impl;
+
+import com.google.common.collect.Lists;
+import java.net.InetSocketAddress;
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.TimeUnit;
+import java.util.function.Consumer;
+import org.apache.pulsar.client.api.DnsResolverConfig;
+import org.apache.pulsar.client.api.EventLoopGroupConfig;
+import org.apache.pulsar.client.api.PulsarClientSharedResources;
+import org.apache.pulsar.client.api.PulsarClientSharedResourcesBuilder;
+import org.apache.pulsar.client.api.ThreadPoolConfig;
+import org.apache.pulsar.client.api.TimerConfig;
+import org.apache.pulsar.common.util.netty.DnsResolverUtil;
+
+public class PulsarClientSharedResourcesBuilderImpl implements PulsarClientSharedResourcesBuilder {
+ Set<PulsarClientSharedResources.SharedResource> sharedResources = new HashSet<>();
+ Map<PulsarClientSharedResources.SharedResource, ResourceConfig> resourceConfigs = new HashMap<>();
+ private boolean shareConfigured;
+
+ interface ResourceConfig {
+
+ }
+
+ abstract static class NamedResourceConfig<T> implements ResourceConfig {
+ String name;
+
+ public T name(String name) {
+ this.name = name;
+ return self();
+ }
+
+ @SuppressWarnings("unchecked")
+ T self() {
+ return (T) this;
+ }
+ }
+
+ static class ThreadPoolResourceConfig extends NamedResourceConfig<ThreadPoolConfig> implements ThreadPoolConfig {
+ int numberOfThreads = Runtime.getRuntime().availableProcessors();
+ boolean daemon = Thread.currentThread().isDaemon();
+
+ @Override
+ public ThreadPoolConfig numberOfThreads(int numberOfThreads) {
+ this.numberOfThreads = numberOfThreads;
+ return this;
+ }
+
+ @Override
+ public ThreadPoolConfig daemon(boolean daemon) {
+ this.daemon = daemon;
+ return this;
+ }
+ }
+
+ static class EventLoopResourceConfig extends NamedResourceConfig<EventLoopGroupConfig>
+ implements EventLoopGroupConfig {
+ int numberOfThreads = Runtime.getRuntime().availableProcessors();
+ boolean daemon = Thread.currentThread().isDaemon();
+ boolean enableBusyWait;
+
+ EventLoopResourceConfig() {
+ name = PulsarClientResourcesConfigurer.POOL_NAME_EVENT_LOOP_GROUP;
+ }
+
+ @Override
+ public EventLoopGroupConfig numberOfThreads(int numberOfThreads) {
+ this.numberOfThreads = numberOfThreads;
+ return this;
+ }
+
+ @Override
+ public EventLoopGroupConfig daemon(boolean daemon) {
+ this.daemon = daemon;
+ return this;
+ }
+
+ @Override
+ public EventLoopGroupConfig enableBusyWait(boolean enableBusyWait) {
+ this.enableBusyWait = enableBusyWait;
+ return this;
+ }
+ }
+
+ static class TimerResourceConfig extends NamedResourceConfig<TimerConfig> implements TimerConfig {
+ long tickDuration = 1L;
+ TimeUnit tickDurationTimeUnit = TimeUnit.MILLISECONDS;
+
+ TimerResourceConfig() {
+ name = PulsarClientResourcesConfigurer.NAME_TIMER;
+ }
+
+ @Override
+ public TimerConfig tickDuration(long tickDuration, TimeUnit timeUnit) {
+ this.tickDuration = tickDuration;
+ this.tickDurationTimeUnit = timeUnit;
+ return this;
+ }
+ }
+
+ static class DnsResolverResourceConfig implements ResourceConfig, DnsResolverConfig {
+ InetSocketAddress localAddress;
+ Collection<InetSocketAddress> serverAddresses;
+ int minTtl = DnsResolverUtil.getDefaultMinTTL();
+ int maxTtl = DnsResolverUtil.getDefaultTTL();
+ int negativeTtl = DnsResolverUtil.getDefaultNegativeTTL();
+ long queryTimeoutMillis = -1L;
+ boolean traceEnabled = true;
+ boolean tcpFallbackEnabled = true;
+ boolean tcpFallbackOnTimeoutEnabled = true;
+ int ndots = -1;
+ Collection<String> searchDomains;
+
+ @Override
+ public DnsResolverConfig localAddress(InetSocketAddress localAddress) {
+ this.localAddress = localAddress;
+ return this;
+ }
+
+ @Override
+ public DnsResolverConfig serverAddresses(Iterable<InetSocketAddress> addresses) {
+ this.serverAddresses = Lists.newArrayList(addresses);
+ return this;
+ }
+
+ @Override
+ public DnsResolverConfig minTtl(int minTtl) {
+ this.minTtl = minTtl;
+ return this;
+ }
+
+ @Override
+ public DnsResolverConfig maxTtl(int maxTtl) {
+ this.maxTtl = maxTtl;
+ return this;
+ }
+
+ @Override
+ public DnsResolverConfig negativeTtl(int negativeTtl) {
+ this.negativeTtl = negativeTtl;
+ return this;
+ }
+
+ @Override
+ public DnsResolverConfig queryTimeoutMillis(long queryTimeoutMillis) {
+ this.queryTimeoutMillis = queryTimeoutMillis;
+ return this;
+ }
+
+ @Override
+ public DnsResolverConfig traceEnabled(boolean traceEnabled) {
+ this.traceEnabled = traceEnabled;
+ return this;
+ }
+
+ @Override
+ public DnsResolverConfig tcpFallbackEnabled(boolean tcpFallbackEnabled) {
+ this.tcpFallbackEnabled = tcpFallbackEnabled;
+ return this;
+ }
+
+ @Override
+ public DnsResolverConfig tcpFallbackOnTimeoutEnabled(boolean tcpFallbackOnTimeoutEnabled) {
+ this.tcpFallbackOnTimeoutEnabled = tcpFallbackOnTimeoutEnabled;
+ return this;
+ }
+
+ @Override
+ public DnsResolverConfig ndots(int ndots) {
+ this.ndots = ndots;
+ return this;
+ }
+
+ @Override
+ public DnsResolverConfig searchDomains(Iterable<String> searchDomains) {
+ this.searchDomains = Lists.newArrayList(searchDomains);
+ return this;
+ }
+ }
+
+ @Override
+ public PulsarClientSharedResourcesBuilder resourceTypes(
+ PulsarClientSharedResources.SharedResource... sharedResource) {
+ if (shareConfigured) {
+ throw new IllegalStateException("Cannot set resourceTypes when shareConfigured() has already been called");
+ }
+ return resourceTypes(List.of(sharedResource));
+ }
+
+ @Override
+ public PulsarClientSharedResourcesBuilder resourceTypes(
+ Collection<PulsarClientSharedResources.SharedResource> sharedResource) {
+ if (shareConfigured) {
+ throw new IllegalStateException("Cannot set resourceTypes when shareConfigured() has already been called");
+ }
+ sharedResources.addAll(sharedResource);
+ return this;
+ }
+
+ @Override
+ public PulsarClientSharedResourcesBuilder shareConfigured() {
+ if (!sharedResources.isEmpty()) {
+ throw new IllegalStateException("Cannot use shareConfigured() when resourceTypes has already been set");
+ }
+ shareConfigured = true;
+ return this;
+ }
+
+ @SuppressWarnings("unchecked")
+ private <T extends ResourceConfig> T getOrCreateConfig(PulsarClientSharedResources.SharedResource sharedResource) {
+ return (T) resourceConfigs.computeIfAbsent(sharedResource, k -> {
+ switch (sharedResource.getType()) {
+ case EventLoopGroup:
+ return new EventLoopResourceConfig();
+ case DnsResolver:
+ return new DnsResolverResourceConfig();
+ case ThreadPool:
+ return new ThreadPoolResourceConfig();
+ case Timer:
+ return new TimerResourceConfig();
+ default:
+ throw new IllegalArgumentException("Unknown resource type: " + sharedResource.getType());
+ }
+ });
+ }
+
+ @Override
+ public PulsarClientSharedResourcesBuilder configureThreadPool(
+ PulsarClientSharedResources.SharedResource sharedResource, Consumer<ThreadPoolConfig<?>> configurer) {
+ if (sharedResource.getType() != PulsarClientSharedResources.SharedResourceType.ThreadPool) {
+ throw new IllegalArgumentException("The shared resource " + sharedResource + " doesn't support thread pool"
+ + " configuration");
+ }
+ configurer.accept(getOrCreateConfig(sharedResource));
+ return this;
+ }
+
+ @Override
+ public PulsarClientSharedResourcesBuilder configureEventLoop(Consumer<EventLoopGroupConfig> configurer) {
+ configurer.accept(getOrCreateConfig(PulsarClientSharedResources.SharedResource.EventLoopGroup));
+ return this;
+ }
+
+ @Override
+ public PulsarClientSharedResourcesBuilder configureDnsResolver(Consumer<DnsResolverConfig> configurer) {
+ configurer.accept(getOrCreateConfig(PulsarClientSharedResources.SharedResource.DnsResolver));
+ return this;
+ }
+
+ @Override
+ public PulsarClientSharedResourcesBuilder configureTimer(Consumer<TimerConfig> configurer) {
+ configurer.accept(getOrCreateConfig(PulsarClientSharedResources.SharedResource.Timer));
+ return this;
+ }
+
+ @Override
+ public PulsarClientSharedResources build() {
+ return new PulsarClientSharedResourcesImpl(sharedResources, resourceConfigs, shareConfigured);
+ }
+}
diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/PulsarClientSharedResourcesImpl.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/PulsarClientSharedResourcesImpl.java
new file mode 100644
index 00000000000..29953784681
--- /dev/null
+++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/PulsarClientSharedResourcesImpl.java
@@ -0,0 +1,160 @@
+/*
+ * 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.client.impl;
+
+import static org.apache.pulsar.client.impl.PulsarClientResourcesConfigurer.createDnsResolverGroupWithResourceConfig;
+import static org.apache.pulsar.client.impl.PulsarClientResourcesConfigurer.createEventLoopGroupWithResourceConfig;
+import static org.apache.pulsar.client.impl.PulsarClientResourcesConfigurer.createExternalExecutorProviderWithResourceConfig;
+import static org.apache.pulsar.client.impl.PulsarClientResourcesConfigurer.createInternalExecutorProviderWithResourceConfig;
+import static org.apache.pulsar.client.impl.PulsarClientResourcesConfigurer.createLookupExecutorProviderWithResourceConfig;
+import static org.apache.pulsar.client.impl.PulsarClientResourcesConfigurer.createScheduledExecutorProviderWithResourceConfig;
+import static org.apache.pulsar.client.impl.PulsarClientResourcesConfigurer.createTimer;
+import io.netty.channel.EventLoopGroup;
+import io.netty.util.Timer;
+import java.util.EnumSet;
+import java.util.Map;
+import java.util.Set;
+import lombok.Getter;
+import org.apache.pulsar.client.api.PulsarClientException;
+import org.apache.pulsar.client.api.PulsarClientSharedResources;
+import org.apache.pulsar.client.util.ExecutorProvider;
+import org.apache.pulsar.client.util.ScheduledExecutorProvider;
+import org.apache.pulsar.common.util.netty.EventLoopUtil;
+
+@Getter
+public class PulsarClientSharedResourcesImpl implements PulsarClientSharedResources {
+ Set<SharedResource> sharedResources;
+ protected final EventLoopGroup ioEventLoopGroup;
+ private final ExecutorProvider internalExecutorProvider;
+ private final ExecutorProvider externalExecutorProvider;
+ private final ScheduledExecutorProvider scheduledExecutorProvider;
+ private final Timer timer;
+ private final ExecutorProvider lookupExecutorProvider;
+ private final DnsResolverGroupImpl dnsResolverGroup;
+
+ public PulsarClientSharedResourcesImpl(Set<SharedResource> sharedResources,
+ Map<SharedResource, PulsarClientSharedResourcesBuilderImpl.ResourceConfig>
+ resourceConfigs, boolean shareConfigured) {
+ if (shareConfigured) {
+ if (resourceConfigs.isEmpty()) {
+ throw new IllegalArgumentException(
+ "No resources have been configured while using shareConfigured mode");
+ }
+ this.sharedResources = Set.copyOf(resourceConfigs.keySet());
+ } else {
+ if (sharedResources.isEmpty()) {
+ this.sharedResources = EnumSet.allOf(SharedResource.class);
+ } else {
+ this.sharedResources = Set.copyOf(sharedResources);
+ }
+ }
+ this.ioEventLoopGroup = this.sharedResources.contains(SharedResource.EventLoopGroup)
+ ? createEventLoopGroupWithResourceConfig(
+ getResourceConfig(resourceConfigs, SharedResource.EventLoopGroup))
+ : null;
+ this.externalExecutorProvider = this.sharedResources.contains(SharedResource.ListenerExecutor)
+ ? createExternalExecutorProviderWithResourceConfig(
+ getResourceConfig(resourceConfigs, SharedResource.ListenerExecutor))
+ : null;
+ this.internalExecutorProvider = this.sharedResources.contains(SharedResource.InternalExecutor)
+ ? createInternalExecutorProviderWithResourceConfig(
+ getResourceConfig(resourceConfigs, SharedResource.InternalExecutor))
+ : null;
+ this.scheduledExecutorProvider = this.sharedResources.contains(SharedResource.ScheduledExecutor)
+ ? createScheduledExecutorProviderWithResourceConfig(
+ getResourceConfig(resourceConfigs, SharedResource.ScheduledExecutor))
+ : null;
+ this.lookupExecutorProvider = this.sharedResources.contains(SharedResource.LookupExecutor)
+ ? createLookupExecutorProviderWithResourceConfig(
+ getResourceConfig(resourceConfigs, SharedResource.LookupExecutor))
+ : null;
+ this.timer = this.sharedResources.contains(SharedResource.Timer)
+ ? createTimer(getResourceConfig(resourceConfigs, SharedResource.Timer))
+ : null;
+ this.dnsResolverGroup = this.sharedResources.contains(SharedResource.DnsResolver)
+ ? createDnsResolverGroupWithResourceConfig(
+ getResourceConfig(resourceConfigs, SharedResource.DnsResolver))
+ : null;
+ }
+
+ @SuppressWarnings("unchecked")
+ private <T extends PulsarClientSharedResourcesBuilderImpl.ResourceConfig> T getResourceConfig(
+ Map<SharedResource, PulsarClientSharedResourcesBuilderImpl.ResourceConfig> resourceConfigs,
+ SharedResource resource) {
+ return (T) resourceConfigs.get(resource);
+ }
+
+ @Override
+ public boolean contains(SharedResource sharedResource) {
+ return sharedResources.contains(sharedResource);
+ }
+
+ public Set<SharedResource> getSharedResources() {
+ return sharedResources;
+ }
+
+ @Override
+ public void close() throws PulsarClientException {
+ if (externalExecutorProvider != null) {
+ externalExecutorProvider.shutdownNow();
+ }
+ if (internalExecutorProvider != null) {
+ internalExecutorProvider.shutdownNow();
+ }
+ if (scheduledExecutorProvider != null) {
+ scheduledExecutorProvider.shutdownNow();
+ }
+ if (lookupExecutorProvider != null) {
+ lookupExecutorProvider.shutdownNow();
+ }
+ if (dnsResolverGroup != null) {
+ dnsResolverGroup.close();
+ }
+ if (timer != null) {
+ timer.stop();
+ }
+ if (ioEventLoopGroup != null) {
+ EventLoopUtil.shutdownGracefully(ioEventLoopGroup);
+ }
+ }
+
+ public void applyTo(PulsarClientImpl.PulsarClientImplBuilder instanceBuilder) {
+ if (externalExecutorProvider != null) {
+ instanceBuilder.externalExecutorProvider(externalExecutorProvider);
+ }
+ if (internalExecutorProvider != null) {
+ instanceBuilder.internalExecutorProvider(internalExecutorProvider);
+ }
+ if (scheduledExecutorProvider != null) {
+ instanceBuilder.scheduledExecutorProvider(scheduledExecutorProvider);
+ }
+ if (lookupExecutorProvider != null) {
+ instanceBuilder.lookupExecutorProvider(lookupExecutorProvider);
+ }
+ if (dnsResolverGroup != null) {
+ instanceBuilder.dnsResolverGroup(dnsResolverGroup);
+ }
+ if (timer != null) {
+ instanceBuilder.timer(timer);
+ }
+ if (ioEventLoopGroup != null) {
+ instanceBuilder.eventLoopGroup(ioEventLoopGroup);
+ }
+ }
+}
diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/util/ExecutorProvider.java b/pulsar-client/src/main/java/org/apache/pulsar/client/util/ExecutorProvider.java
index 88654c51300..ae3b532a054 100644
--- a/pulsar-client/src/main/java/org/apache/pulsar/client/util/ExecutorProvider.java
+++ b/pulsar-client/src/main/java/org/apache/pulsar/client/util/ExecutorProvider.java
@@ -61,13 +61,17 @@ public class ExecutorProvider {
}
public ExecutorProvider(int numThreads, String poolName) {
+ this(numThreads, poolName, Thread.currentThread().isDaemon());
+ }
+
+ public ExecutorProvider(int numThreads, String poolName, boolean daemon) {
checkArgument(numThreads > 0);
this.numThreads = numThreads;
Objects.requireNonNull(poolName);
executors = new ArrayList<>(numThreads);
for (int i = 0; i < numThreads; i++) {
ExtendedThreadFactory threadFactory = new ExtendedThreadFactory(
- poolName, Thread.currentThread().isDaemon());
+ poolName, daemon);
ExecutorService executor = createExecutor(threadFactory);
executors.add(Pair.of(executor, threadFactory));
}
diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/util/ScheduledExecutorProvider.java b/pulsar-client/src/main/java/org/apache/pulsar/client/util/ScheduledExecutorProvider.java
index 3ebd41ebe56..b47659e514a 100644
--- a/pulsar-client/src/main/java/org/apache/pulsar/client/util/ScheduledExecutorProvider.java
+++ b/pulsar-client/src/main/java/org/apache/pulsar/client/util/ScheduledExecutorProvider.java
@@ -24,6 +24,9 @@ import lombok.extern.slf4j.Slf4j;
@Slf4j
public class ScheduledExecutorProvider extends ExecutorProvider {
+ public ScheduledExecutorProvider(int numThreads, String poolName, boolean daemon) {
+ super(numThreads, poolName, daemon);
+ }
public ScheduledExecutorProvider(int numThreads, String poolName) {
super(numThreads, poolName);
diff --git a/pulsar-client/src/main/resources/findbugsExclude.xml b/pulsar-client/src/main/resources/findbugsExclude.xml
index f7cf6b9cfd5..802c03622b0 100644
--- a/pulsar-client/src/main/resources/findbugsExclude.xml
+++ b/pulsar-client/src/main/resources/findbugsExclude.xml
@@ -257,6 +257,10 @@
<Method name="<init>"/>
<Bug pattern="EI_EXPOSE_REP2"/>
</Match>
+ <Match>
+ <Class name="org.apache.pulsar.client.impl.PulsarClientSharedResourcesImpl"/>
+ <Bug pattern="EI_EXPOSE_REP"/>
+ </Match>
<Match>
<Class name="org.apache.pulsar.client.impl.ClientCnx"/>
<Method name="getPendingRequests"/>
@@ -712,6 +716,11 @@
<Method name="<init>"/>
<Bug pattern="EI_EXPOSE_REP2"/>
</Match>
+ <Match>
+ <Class name="org.apache.pulsar.client.impl.PulsarClientImpl"/>
+ <Method name="<init>"/>
+ <Bug pattern="EI_EXPOSE_REP2"/>
+ </Match>
<Match>
<Class name="org.apache.pulsar.client.impl.PulsarClientImpl"/>
<Method name="getConfiguration"/>
diff --git a/pulsar-client/src/test/java/org/apache/pulsar/client/impl/ControlledClusterFailoverTest.java b/pulsar-client/src/test/java/org/apache/pulsar/client/impl/ControlledClusterFailoverTest.java
index 265af6dd23c..ca4aca6c329 100644
--- a/pulsar-client/src/test/java/org/apache/pulsar/client/impl/ControlledClusterFailoverTest.java
+++ b/pulsar-client/src/test/java/org/apache/pulsar/client/impl/ControlledClusterFailoverTest.java
@@ -54,6 +54,12 @@ public class ControlledClusterFailoverTest {
.build();
ControlledClusterFailover controlledClusterFailover = (ControlledClusterFailover) provider;
+
+ PulsarClientImpl pulsarClient = mock(PulsarClientImpl.class);
+ ConnectionPool connectionPool = mock(ConnectionPool.class);
+ when(pulsarClient.getCnxPool()).thenReturn(connectionPool);
+ controlledClusterFailover.initialize(pulsarClient);
+
Request request = controlledClusterFailover.getRequestBuilder().build();
Assert.assertTrue(provider instanceof ControlledClusterFailover);
diff --git a/pulsar-client/src/test/java/org/apache/pulsar/client/impl/PulsarClientSharedResourcesBuilderImplTest.java b/pulsar-client/src/test/java/org/apache/pulsar/client/impl/PulsarClientSharedResourcesBuilderImplTest.java
new file mode 100644
index 00000000000..f01906911a6
--- /dev/null
+++ b/pulsar-client/src/test/java/org/apache/pulsar/client/impl/PulsarClientSharedResourcesBuilderImplTest.java
@@ -0,0 +1,136 @@
+/*
+ * 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.client.impl;
+
+import java.net.InetSocketAddress;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.TimeUnit;
+import org.apache.pulsar.client.api.PulsarClient;
+import org.apache.pulsar.client.api.PulsarClientException;
+import org.apache.pulsar.client.api.PulsarClientSharedResources;
+import org.testng.annotations.Test;
+
+public class PulsarClientSharedResourcesBuilderImplTest {
+ @Test
+ public void testSharedResources() throws PulsarClientException {
+ runClientsWithSharedResources(PulsarClientSharedResources.builder().build(), 1000, false);
+ runClientsWithSharedResources(PulsarClientSharedResources.builder().build(), 1000, true);
+ }
+
+ private static void runClientsWithSharedResources(PulsarClientSharedResources sharedResources, int numberOfClients,
+ boolean useHttpLookup)
+ throws PulsarClientException {
+ List<PulsarClient> clients = new ArrayList<>();
+ for (int i = 0; i < numberOfClients; i++) {
+ clients.add(PulsarClient.builder()
+ .serviceUrl(useHttpLookup ? "http://localhost:8080" : "pulsar://localhost:6650")
+ .sharedResources(sharedResources)
+ .build());
+ }
+ for (PulsarClient client : clients) {
+ client.close();
+ }
+ sharedResources.close();
+ }
+
+ @Test
+ public void testPulsarClientSharedResourcesBuilderUsingPublicAPI() throws PulsarClientException {
+ PulsarClientSharedResources sharedResources = PulsarClientSharedResources.builder()
+ .configureEventLoop(eventLoopGroupConfig -> {
+ eventLoopGroupConfig
+ .name("testEventLoop")
+ .numberOfThreads(10);
+ })
+ .configureDnsResolver(dnsResolverConfig -> {
+ dnsResolverConfig.localAddress(new InetSocketAddress(0));
+ })
+ .configureThreadPool(PulsarClientSharedResources.SharedResource.ListenerExecutor,
+ threadPoolConfig -> {
+ threadPoolConfig.name("testListenerThreadPool")
+ .daemon(true)
+ .numberOfThreads(12);
+ }
+ )
+ .configureThreadPool(PulsarClientSharedResources.SharedResource.InternalExecutor,
+ threadPoolConfig -> {
+ threadPoolConfig.name("testInternalThreadPool")
+ .daemon(true)
+ .numberOfThreads(2);
+ }
+ )
+ .configureThreadPool(PulsarClientSharedResources.SharedResource.LookupExecutor,
+ threadPoolConfig -> {
+ threadPoolConfig.name("testLookupThreadPool")
+ .daemon(true)
+ .numberOfThreads(1);
+ }
+ )
+ .configureThreadPool(PulsarClientSharedResources.SharedResource.ScheduledExecutor,
+ threadPoolConfig -> {
+ threadPoolConfig.name("testSchedulerThreadPool")
+ .daemon(true)
+ .numberOfThreads(1);
+ }
+ )
+ .configureTimer(timerConfig -> {
+ timerConfig.name("testTimer").tickDuration(100, TimeUnit.MILLISECONDS);
+ })
+ .build();
+ runClientsWithSharedResources(sharedResources, 1000, false);
+ sharedResources.close();
+ }
+
+ @Test
+ public void testPartialSharing() throws PulsarClientException {
+ PulsarClientSharedResources sharedResources =
+ PulsarClientSharedResources.builder()
+ .shareConfigured()
+ .configureEventLoop(eventLoopGroupConfig -> {
+ eventLoopGroupConfig.name("testEventLoop").numberOfThreads(10);
+ }).configureDnsResolver(dnsResolverConfig -> {
+ dnsResolverConfig.localAddress(new InetSocketAddress(0));
+ }).build();
+ runClientsWithSharedResources(sharedResources, 2, false);
+ sharedResources.close();
+ }
+
+ @Test
+ public void testDnsResolverConfig() throws PulsarClientException {
+ PulsarClientSharedResources sharedResources =
+ PulsarClientSharedResources.builder()
+ .shareConfigured()
+ .configureDnsResolver(dnsResolverConfig -> {
+ dnsResolverConfig
+ .localAddress(new InetSocketAddress(0))
+ .serverAddresses(List.of(new InetSocketAddress("8.8.8.8", 53)))
+ .minTtl(0)
+ .maxTtl(45)
+ .negativeTtl(10)
+ .ndots(1)
+ .searchDomains(List.of("mycompany.com"))
+ .queryTimeoutMillis(5000L)
+ .tcpFallbackEnabled(true)
+ .tcpFallbackOnTimeoutEnabled(false)
+ .traceEnabled(false);
+ }).build();
+ runClientsWithSharedResources(sharedResources, 2, false);
+ sharedResources.close();
+ }
+}
\ No newline at end of file
diff --git a/pulsar-common/src/main/java/org/apache/pulsar/common/util/netty/DnsResolverUtil.java b/pulsar-common/src/main/java/org/apache/pulsar/common/util/netty/DnsResolverUtil.java
index bcff83acd94..92c025092d3 100644
--- a/pulsar-common/src/main/java/org/apache/pulsar/common/util/netty/DnsResolverUtil.java
+++ b/pulsar-common/src/main/java/org/apache/pulsar/common/util/netty/DnsResolverUtil.java
@@ -18,10 +18,24 @@
*/
package org.apache.pulsar.common.util.netty;
+import io.netty.resolver.AddressResolver;
+import io.netty.resolver.DefaultNameResolver;
+import io.netty.resolver.InetNameResolver;
+import io.netty.resolver.InetSocketAddressResolver;
+import io.netty.resolver.NameResolver;
import io.netty.resolver.dns.DnsNameResolverBuilder;
+import io.netty.util.concurrent.EventExecutor;
+import io.netty.util.concurrent.GenericFutureListener;
+import io.netty.util.concurrent.ImmediateEventExecutor;
+import io.netty.util.concurrent.Promise;
+import java.lang.reflect.Field;
+import java.net.InetAddress;
+import java.net.InetSocketAddress;
import java.security.Security;
+import java.util.List;
import java.util.Optional;
import lombok.extern.slf4j.Slf4j;
+import org.apache.commons.lang3.reflect.FieldUtils;
@Slf4j
public class DnsResolverUtil {
@@ -61,8 +75,15 @@ public class DnsResolverUtil {
.map(Integer::decode)
.filter(i -> i > 0)
.orElseGet(() -> {
- if (System.getSecurityManager() == null) {
- return JDK_DEFAULT_TTL;
+ try {
+ if (System.getSecurityManager() == null) {
+ return JDK_DEFAULT_TTL;
+ }
+ } catch (Throwable t) {
+ log.warn("Cannot use current logic to resolve JDK default DNS TTL settings. Use "
+ + "sun.net.inetaddr.ttl and sun.net.inetaddr.negative.ttl system "
+ + "properties for setting default values for DNS TTL settings. {}",
+ t.getMessage());
}
return DEFAULT_TTL;
});
@@ -92,4 +113,97 @@ public class DnsResolverUtil {
dnsNameResolverBuilder.ttl(MIN_TTL, TTL);
dnsNameResolverBuilder.negativeTtl(NEGATIVE_TTL);
}
+
+ public static int getDefaultMinTTL() {
+ return MIN_TTL;
+ }
+
+ public static int getDefaultTTL() {
+ return TTL;
+ }
+
+ public static int getDefaultNegativeTTL() {
+ return NEGATIVE_TTL;
+ }
+
+ /**
+ * Extract the underlying Netty NameResolver from an AddressResolver instance or creates an adapter as the
+ * fallback. If null is passed in a default Netty NameResolver will be returned which delegates to the
+ * blocking JDK DNS resolver.
+ *
+ * @param addressResolver Netty AddressResolver instance or null
+ * @return Netty NameResolver instance
+ */
+ @SuppressWarnings("unchecked")
+ public static NameResolver<InetAddress> adaptToNameResolver(AddressResolver<InetSocketAddress> addressResolver) {
+ if (addressResolver == null) {
+ return new DefaultNameResolver(ImmediateEventExecutor.INSTANCE);
+ }
+ // Use reflection to extract underlying Netty NameResolver instance.
+ if (InetSocketAddressResolver.class.isInstance(addressResolver)) {
+ try {
+ Field nameResolverField =
+ FieldUtils.getDeclaredField(InetSocketAddressResolver.class, "nameResolver", true);
+ if (nameResolverField != null) {
+ return (NameResolver<InetAddress>) FieldUtils.readField(nameResolverField, addressResolver);
+ } else {
+ log.warn("Could not find nameResolver Field in InetSocketAddressResolver instance.");
+ }
+ } catch (Throwable t) {
+ log.warn("Failed to extract NameResolver from InetSocketAddressResolver instance. {}", t.getMessage());
+ }
+ }
+ // fallback to use an adapter if reflection fails
+ log.info("Creating NameResolver adapter that wraps an AddressResolver instance.");
+ return createNameResolverAdapter(addressResolver, ImmediateEventExecutor.INSTANCE);
+ }
+
+ /**
+ * Creates a NameResolver adapter that wraps an AddressResolver instance.
+ * <p>
+ * This adapter is necessary because Netty doesn't provide a direct implementation for converting
+ * between AddressResolver and NameResolver, while AsyncHttpClient specifically requires a NameResolver.
+ * The adapter handles the resolution of hostnames to IP addresses by delegating to the underlying
+ * AddressResolver.
+ *
+ * @param addressResolver the AddressResolver instance to adapt, handling InetSocketAddress resolution
+ * @param executor the EventExecutor to be used for executing resolution tasks
+ * @return a NameResolver instance that wraps the provided AddressResolver
+ */
+ static NameResolver<InetAddress> createNameResolverAdapter(
+ AddressResolver<InetSocketAddress> addressResolver, EventExecutor executor) {
+ return new InetNameResolver(executor) {
+ @Override
+ protected void doResolve(String inetHost, Promise<InetAddress> promise) throws Exception {
+ Promise<InetSocketAddress> delegatedPromise = executor().newPromise();
+ addressResolver.resolve(InetSocketAddress.createUnresolved(inetHost, 1), delegatedPromise);
+ delegatedPromise.addListener(new GenericFutureListener<Promise<InetSocketAddress>>() {
+ @Override
+ public void operationComplete(Promise<InetSocketAddress> future) throws Exception {
+ if (future.isSuccess()) {
+ promise.setSuccess(future.get().getAddress());
+ } else {
+ promise.setFailure(future.cause());
+ }
+ }
+ });
+ }
+
+ @Override
+ protected void doResolveAll(String inetHost, Promise<List<InetAddress>> promise) throws Exception {
+ Promise<List<InetSocketAddress>> delegatedPromise = executor().newPromise();
+ addressResolver.resolveAll(InetSocketAddress.createUnresolved(inetHost, 1), delegatedPromise);
+ delegatedPromise.addListener(new GenericFutureListener<Promise<List<InetSocketAddress>>>() {
+ @Override
+ public void operationComplete(Promise<List<InetSocketAddress>> future) throws Exception {
+ if (future.isSuccess()) {
+ promise.setSuccess(future.get().stream().map(InetSocketAddress::getAddress).toList());
+ } else {
+ promise.setFailure(future.cause());
+ }
+ }
+ });
+ }
+ };
+ }
}
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 cfc766890f0..501fe53f3a3 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
@@ -57,13 +57,7 @@ public class EventLoopUtil {
*/
public static EventLoopGroup newEventLoopGroup(int nThreads, boolean enableBusyWait, ThreadFactory threadFactory) {
if (Epoll.isAvailable()) {
- String enableIoUring = System.getProperty(ENABLE_IO_URING);
-
- // 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();
+ if (isIoUringEnabledAndAvailable()) {
return new IOUringEventLoopGroup(nThreads, threadFactory);
} else {
if (!enableBusyWait) {
@@ -96,6 +90,17 @@ public class EventLoopUtil {
}
}
+ private static boolean isIoUringEnabledAndAvailable() {
+ // By default, io_uring will not be enabled, even if available. The environment variable will be used:
+ // enable.io_uring=1
+ boolean ioUringEnabled = StringUtils.equalsAnyIgnoreCase(System.getProperty(ENABLE_IO_URING), "1", "true");
+ if (ioUringEnabled) {
+ // Throw exception if IOUring cannot be used
+ IOUring.ensureAvailability();
+ }
+ return ioUringEnabled;
+ }
+
/**
* Return a SocketChannel class suitable for the given EventLoopGroup implementation.
*
@@ -112,6 +117,25 @@ public class EventLoopUtil {
}
}
+ /**
+ * Returns the most appropriate SocketChannel implementation based on the system's capabilities
+ * and configuration. Precedence:
+ * 1) IO_uring (if available and enabled via 'pulsar.enableUring')
+ * 2) Epoll (if available)
+ * 3) NIO (fallback)
+ */
+ public static Class<? extends SocketChannel> getClientSocketChannelClass() {
+ if (Epoll.isAvailable()) {
+ if (isIoUringEnabledAndAvailable()) {
+ return IOUringSocketChannel.class;
+ } else {
+ return EpollSocketChannel.class;
+ }
+ } else {
+ return NioSocketChannel.class;
+ }
+ }
+
public static Class<? extends ServerSocketChannel> getServerSocketChannelClass(EventLoopGroup eventLoopGroup) {
if (eventLoopGroup instanceof IOUringEventLoopGroup) {
return IOUringServerSocketChannel.class;
@@ -122,6 +146,25 @@ public class EventLoopUtil {
}
}
+ /**
+ * Returns the most appropriate ServerSocketChannel implementation based on the system's capabilities
+ * and configuration. Precedence:
+ * 1) IO_uring (if available and enabled via 'pulsar.enableUring')
+ * 2) Epoll (if available)
+ * 3) NIO (fallback)
+ */
+ public static Class<? extends ServerSocketChannel> getServerSocketChannelClass() {
+ if (Epoll.isAvailable()) {
+ if (isIoUringEnabledAndAvailable()) {
+ return IOUringServerSocketChannel.class;
+ } else {
+ return EpollServerSocketChannel.class;
+ }
+ } else {
+ return NioServerSocketChannel.class;
+ }
+ }
+
public static Class<? extends DatagramChannel> getDatagramChannelClass(EventLoopGroup eventLoopGroup) {
if (eventLoopGroup instanceof IOUringEventLoopGroup) {
return IOUringDatagramChannel.class;
@@ -132,6 +175,25 @@ public class EventLoopUtil {
}
}
+ /**
+ * Returns the most appropriate DatagramChannel implementation based on the system's capabilities
+ * and configuration. Precedence:
+ * 1) IO_uring (if available and enabled via 'pulsar.enableUring')
+ * 2) Epoll (if available)
+ * 3) NIO (fallback)
+ */
+ public static Class<? extends DatagramChannel> getDatagramChannelClass() {
+ if (Epoll.isAvailable()) {
+ if (isIoUringEnabledAndAvailable()) {
+ return IOUringDatagramChannel.class;
+ } else {
+ return EpollDatagramChannel.class;
+ }
+ } else {
+ return NioDatagramChannel.class;
+ }
+ }
+
public static void enableTriggeredMode(ServerBootstrap bootstrap) {
if (Epoll.isAvailable()) {
bootstrap.childOption(EpollChannelOption.EPOLL_MODE, EpollMode.LEVEL_TRIGGERED);
diff --git a/pulsar-common/src/test/java/org/apache/pulsar/common/util/netty/DnsResolverTest.java b/pulsar-common/src/test/java/org/apache/pulsar/common/util/netty/DnsResolverTest.java
index 46599cc45a0..62b59f89368 100644
--- a/pulsar-common/src/test/java/org/apache/pulsar/common/util/netty/DnsResolverTest.java
+++ b/pulsar-common/src/test/java/org/apache/pulsar/common/util/netty/DnsResolverTest.java
@@ -31,6 +31,7 @@ import org.testng.annotations.AfterClass;
import org.testng.annotations.BeforeClass;
import org.testng.annotations.Test;
+// This test needs to be run separately since it relies on static field values
public class DnsResolverTest {
private static final int MIN_TTL = 0;
private static final int TTL = 101;
diff --git a/pulsar-common/src/test/java/org/apache/pulsar/common/util/netty/DnsResolverUtilTest.java b/pulsar-common/src/test/java/org/apache/pulsar/common/util/netty/DnsResolverUtilTest.java
new file mode 100644
index 00000000000..beb68be6ed3
--- /dev/null
+++ b/pulsar-common/src/test/java/org/apache/pulsar/common/util/netty/DnsResolverUtilTest.java
@@ -0,0 +1,68 @@
+/*
+ * 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 com.google.common.net.InetAddresses;
+import io.netty.resolver.AddressResolver;
+import io.netty.resolver.DefaultNameResolver;
+import io.netty.resolver.NameResolver;
+import io.netty.util.concurrent.ImmediateEventExecutor;
+import java.net.InetAddress;
+import java.net.InetSocketAddress;
+import java.util.List;
+import java.util.concurrent.ExecutionException;
+import org.testng.annotations.Test;
+
+public class DnsResolverUtilTest {
+
+ @Test
+ public void testNameResolverAdapter() throws ExecutionException, InterruptedException {
+ ImmediateEventExecutor executor = ImmediateEventExecutor.INSTANCE;
+ AddressResolver<InetSocketAddress> addressResolver =
+ new DefaultNameResolver(executor).asAddressResolver();
+ NameResolver<InetAddress> nameResolverAdapter =
+ DnsResolverUtil.createNameResolverAdapter(addressResolver, executor);
+ testNameResolver(nameResolverAdapter);
+ }
+
+ @Test
+ public void testNameResolverExtraction() throws ExecutionException, InterruptedException {
+ ImmediateEventExecutor executor = ImmediateEventExecutor.INSTANCE;
+ AddressResolver<InetSocketAddress> addressResolver =
+ new DefaultNameResolver(executor).asAddressResolver();
+ NameResolver<InetAddress> nameResolverAdapter =
+ DnsResolverUtil.adaptToNameResolver(addressResolver);
+ testNameResolver(nameResolverAdapter);
+ }
+
+ @Test
+ public void testNameResolverNullFallback() throws ExecutionException, InterruptedException {
+ NameResolver<InetAddress> nameResolverAdapter = DnsResolverUtil.adaptToNameResolver(null);
+ testNameResolver(nameResolverAdapter);
+ }
+
+ private static void testNameResolver(NameResolver<InetAddress> nameResolver)
+ throws InterruptedException, ExecutionException {
+ InetAddress inetAddress = nameResolver.resolve("8.8.8.8").get();
+ assertEquals(inetAddress, InetAddresses.forString("8.8.8.8"));
+ List<InetAddress> inetAddresses = nameResolver.resolveAll("8.8.8.8").get();
+ assertEquals(inetAddresses.get(0), InetAddresses.forString("8.8.8.8"));
+ }
+}
\ No newline at end of file