From 4ef35256d32efb2ac591e62627452725d0361838 Mon Sep 17 00:00:00 2001 From: Lari Hotari Date: Wed, 19 Jan 2022 15:17:50 +0200 Subject: [PATCH 1/3] [Broker] Use shared executors for broker clients and geo-replication clients --- .../pulsar/broker/ServiceConfiguration.java | 7 +++++ .../apache/pulsar/broker/PulsarService.java | 29 ++++++++++++++++++- .../broker/namespace/NamespaceService.java | 3 +- .../pulsar/broker/service/BrokerService.java | 3 +- 4 files changed, 37 insertions(+), 5 deletions(-) diff --git a/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/ServiceConfiguration.java b/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/ServiceConfiguration.java index 1e93dc53d25b9..25cb4f43d942f 100644 --- a/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/ServiceConfiguration.java +++ b/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/ServiceConfiguration.java @@ -245,6 +245,13 @@ public class ServiceConfiguration implements PulsarConfiguration { ) private int numIOThreads = 2 * Runtime.getRuntime().availableProcessors(); + @FieldContext( + category = CATEGORY_SERVER, + doc = "Number of shared threads to use for broker clients." + + " Default is set to `2 * Runtime.getRuntime().availableProcessors()`" + ) + private int brokerClientNumIOThreads = 2 * Runtime.getRuntime().availableProcessors(); + @FieldContext( category = CATEGORY_SERVER, doc = "Number of threads to use for orderedExecutor." 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 eed476c72a9da..fe47983a339ae 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 @@ -31,6 +31,7 @@ import io.netty.channel.EventLoopGroup; import io.netty.channel.socket.SocketChannel; import io.netty.util.HashedWheelTimer; +import io.netty.util.Timer; import io.netty.util.concurrent.DefaultThreadFactory; import java.io.IOException; import java.lang.reflect.Constructor; @@ -126,6 +127,7 @@ import org.apache.pulsar.client.api.transaction.TransactionBufferClient; import org.apache.pulsar.client.impl.PulsarClientImpl; import org.apache.pulsar.client.impl.conf.ClientConfigurationData; +import org.apache.pulsar.client.util.ExecutorProvider; import org.apache.pulsar.common.conf.InternalConfigurationData; import org.apache.pulsar.common.configuration.PulsarConfigurationLoader; import org.apache.pulsar.common.configuration.VipStatus; @@ -226,6 +228,9 @@ public class PulsarService implements AutoCloseable, ShutdownService { private final Consumer processTerminator; protected final EventLoopGroup ioEventLoopGroup; + private final ExecutorProvider brokerClientSharedInternalExecutorProvider; + private final ExecutorProvider brokerClientSharedExternalExecutorProvider; + private final Timer brokerClientSharedTimer; private MetricsGenerator metricsGenerator; @@ -318,6 +323,14 @@ public PulsarService(ServiceConfiguration config, this.ioEventLoopGroup = EventLoopUtil.newEventLoopGroup(config.getNumIOThreads(), config.isEnableBusyWait(), new DefaultThreadFactory("pulsar-io")); + this.brokerClientSharedInternalExecutorProvider = + new ExecutorProvider(config.getBrokerClientNumIOThreads(), "broker-client-shared-internal-executor"); + // the external executor is not used in the Pulsar Proxy since this executor is used for consumer listeners + // since an instance is required, a single threaded shared instance is used for all broker client instances + this.brokerClientSharedExternalExecutorProvider = + new ExecutorProvider(1, "broker-client-shared-external-executor"); + this.brokerClientSharedTimer = new HashedWheelTimer(new DefaultThreadFactory("broker-client-shared-timer", + Thread.currentThread().isDaemon()), 1, TimeUnit.MILLISECONDS); } public MetadataStore createConfigurationMetadataStore() throws MetadataStoreException { @@ -496,6 +509,9 @@ public CompletableFuture closeAsync() { transactionReplayExecutor.shutdown(); } + brokerClientSharedExternalExecutorProvider.shutdownNow(); + brokerClientSharedInternalExecutorProvider.shutdownNow(); + brokerClientSharedTimer.stop(); ioEventLoopGroup.shutdownGracefully(); // add timeout handling for closing executors @@ -1296,6 +1312,17 @@ protected synchronized OrderedScheduler getOffloaderScheduler(OffloadPoliciesImp return this.offloaderScheduler; } + public PulsarClientImpl createClientImpl(ClientConfigurationData clientConf) + throws PulsarClientException { + return PulsarClientImpl.builder() + .conf(clientConf) + .eventLoopGroup(ioEventLoopGroup) + .timer(brokerClientSharedTimer) + .internalExecutorProvider(brokerClientSharedInternalExecutorProvider) + .externalExecutorProvider(brokerClientSharedExternalExecutorProvider) + .build(); + } + public synchronized PulsarClient getClient() throws PulsarServerException { if (this.client == null) { try { @@ -1329,7 +1356,7 @@ public synchronized PulsarClient getClient() throws PulsarServerException { } conf.setStatsIntervalSeconds(0); - this.client = new PulsarClientImpl(conf, ioEventLoopGroup); + this.client = createClientImpl(conf); } catch (Exception e) { throw new PulsarServerException(e); } diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/namespace/NamespaceService.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/namespace/NamespaceService.java index a8a808c1012dd..095770fad2a93 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/namespace/NamespaceService.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/namespace/NamespaceService.java @@ -25,7 +25,6 @@ import static org.apache.commons.lang3.StringUtils.isNotBlank; import com.google.common.collect.Lists; import com.google.common.hash.Hashing; -import io.netty.channel.EventLoopGroup; import io.prometheus.client.Counter; import java.net.URI; import java.net.URL; @@ -1239,7 +1238,7 @@ public PulsarClientImpl getNamespaceClient(ClusterDataImpl cluster) { // Share all the IO threads across broker and client connections ClientConfigurationData conf = ((ClientBuilderImpl) clientBuilder).getClientConfigurationData(); - return new PulsarClientImpl(conf, (EventLoopGroup) pulsar.getBrokerService().executor()); + return pulsar.createClientImpl(conf); } catch (Exception e) { throw new RuntimeException(e); } diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java index 613961d1db01b..3792ca343d52a 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java @@ -128,7 +128,6 @@ import org.apache.pulsar.client.api.ClientBuilder; import org.apache.pulsar.client.api.PulsarClient; import org.apache.pulsar.client.impl.ClientBuilderImpl; -import org.apache.pulsar.client.impl.PulsarClientImpl; import org.apache.pulsar.client.impl.conf.ClientConfigurationData; import org.apache.pulsar.common.allocator.PulsarByteBufAllocator; import org.apache.pulsar.common.configuration.BindAddress; @@ -1203,7 +1202,7 @@ public PulsarClient getReplicationClient(String cluster, Optional c } // Share all the IO threads across broker and client connections ClientConfigurationData conf = ((ClientBuilderImpl) clientBuilder).getClientConfigurationData(); - return new PulsarClientImpl(conf, workerGroup); + return pulsar.createClientImpl(conf); } catch (Exception e) { throw new RuntimeException(e); } From 8f9f2ed1b01e3de61d559375ef3102e921cb0d6d Mon Sep 17 00:00:00 2001 From: Lari Hotari Date: Thu, 20 Jan 2022 06:36:39 +0200 Subject: [PATCH 2/3] Remove brokerClientNumIOThreads configuration key and default to 1 --- .../org/apache/pulsar/broker/ServiceConfiguration.java | 7 ------- .../main/java/org/apache/pulsar/broker/PulsarService.java | 8 ++++++-- 2 files changed, 6 insertions(+), 9 deletions(-) diff --git a/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/ServiceConfiguration.java b/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/ServiceConfiguration.java index 25cb4f43d942f..1e93dc53d25b9 100644 --- a/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/ServiceConfiguration.java +++ b/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/ServiceConfiguration.java @@ -245,13 +245,6 @@ public class ServiceConfiguration implements PulsarConfiguration { ) private int numIOThreads = 2 * Runtime.getRuntime().availableProcessors(); - @FieldContext( - category = CATEGORY_SERVER, - doc = "Number of shared threads to use for broker clients." - + " Default is set to `2 * Runtime.getRuntime().availableProcessors()`" - ) - private int brokerClientNumIOThreads = 2 * Runtime.getRuntime().availableProcessors(); - @FieldContext( category = CATEGORY_SERVER, doc = "Number of threads to use for orderedExecutor." 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 fe47983a339ae..08f43c957988b 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 @@ -323,9 +323,13 @@ public PulsarService(ServiceConfiguration config, this.ioEventLoopGroup = EventLoopUtil.newEventLoopGroup(config.getNumIOThreads(), config.isEnableBusyWait(), new DefaultThreadFactory("pulsar-io")); + // the internal executor is not used in the broker client or replication clients since this executor is + // used for consumers and the transaction support in the client. + // since an instance is required, a single threaded shared instance is used for all broker client instances this.brokerClientSharedInternalExecutorProvider = - new ExecutorProvider(config.getBrokerClientNumIOThreads(), "broker-client-shared-internal-executor"); - // the external executor is not used in the Pulsar Proxy since this executor is used for consumer listeners + new ExecutorProvider(1, "broker-client-shared-internal-executor"); + // the external executor is not used in the broker client or replication clients since this executor is + // used for consumer listeners. // since an instance is required, a single threaded shared instance is used for all broker client instances this.brokerClientSharedExternalExecutorProvider = new ExecutorProvider(1, "broker-client-shared-external-executor"); From f1c8fec0d18f663a4e70c54722f7f62e1396cdde Mon Sep 17 00:00:00 2001 From: Lari Hotari Date: Thu, 20 Jan 2022 06:44:30 +0200 Subject: [PATCH 3/3] Revisit the shared timer creation - don't ever make it a daemon thread --- .../src/main/java/org/apache/pulsar/broker/PulsarService.java | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) 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 08f43c957988b..1de1cc0a73a63 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 @@ -333,8 +333,8 @@ public PulsarService(ServiceConfiguration config, // since an instance is required, a single threaded shared instance is used for all broker client instances this.brokerClientSharedExternalExecutorProvider = new ExecutorProvider(1, "broker-client-shared-external-executor"); - this.brokerClientSharedTimer = new HashedWheelTimer(new DefaultThreadFactory("broker-client-shared-timer", - Thread.currentThread().isDaemon()), 1, TimeUnit.MILLISECONDS); + this.brokerClientSharedTimer = + new HashedWheelTimer(new DefaultThreadFactory("broker-client-shared-timer"), 1, TimeUnit.MILLISECONDS); } public MetadataStore createConfigurationMetadataStore() throws MetadataStoreException {