You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by bo...@apache.org on 2022/11/10 14:28:05 UTC
[pulsar] branch branch-2.9 updated (a58894a859c -> 69ff4dbdf79)
This is an automated email from the ASF dual-hosted git repository.
bogong pushed a change to branch branch-2.9
in repository https://gitbox.apache.org/repos/asf/pulsar.git
from a58894a859c [improve][java-client] Support passing existing scheduled executor providers to the client (#16334)
new df3e2cea1da Disable stats recorder for built-in PulsarClient (#12217)
new a75a22ddd5f [Broker] Use shared executors for broker and geo-replication clients (#13839)
new 69ff4dbdf79 [improve][broker] Use shared broker client scheduled executor provider (#16338)
The 3 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:
.../org/apache/pulsar/broker/PulsarService.java | 40 +++++++++++++++++++++-
.../pulsar/broker/namespace/NamespaceService.java | 3 +-
.../pulsar/broker/service/BrokerService.java | 3 +-
3 files changed, 41 insertions(+), 5 deletions(-)
[pulsar] 01/03: Disable stats recorder for built-in PulsarClient (#12217)
Posted by bo...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
bogong pushed a commit to branch branch-2.9
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit df3e2cea1daefdba778e2d6a205b413af2766abb
Author: Yunze Xu <xy...@163.com>
AuthorDate: Tue Sep 28 22:14:08 2021 +0800
Disable stats recorder for built-in PulsarClient (#12217)
### Motivation
In `PulsarService`, there's a built-in client used in many places, like topic compaction, system topics read/write. It could also be used in protocol handlers. Different with the manually created Pulsar client, it can read the configured authentication params and TLS related configs for broker-to-broker communication.
However, it doesn't change the default `statsIntervalSeconds` config, which means each time a producer/consumer/reader is created, a stats recorder (`ProducerStatsRecorderImpl` or `ConsumerStatsRecorderImpl`) will be created, in which there's a Netty `TimerTask` that prints stats periodically (the default interval is 1 minute). If there're many producers/consumers/readers created by this built-in client, the unnecessary CPU usage will be high.
### Modifications
Configure the `statsIntervalSeconds` with zero value to disable stats recorder.
(cherry picked from commit 52232e6e78e2ebf7a2d04489fd116c7f3abe7901)
---
pulsar-broker/src/main/java/org/apache/pulsar/broker/PulsarService.java | 2 ++
1 file changed, 2 insertions(+)
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 a95ba68344a..6446da29071 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
@@ -1386,6 +1386,8 @@ public class PulsarService implements AutoCloseable, ShutdownService {
this.getConfiguration().getBrokerClientAuthenticationPlugin(),
this.getConfiguration().getBrokerClientAuthenticationParameters()));
}
+
+ conf.setStatsIntervalSeconds(0);
this.client = new PulsarClientImpl(conf, ioEventLoopGroup);
} catch (Exception e) {
throw new PulsarServerException(e);
[pulsar] 03/03: [improve][broker] Use shared broker client scheduled executor provider (#16338)
Posted by bo...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
bogong pushed a commit to branch branch-2.9
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 69ff4dbdf7929f3ffd19cf5d8ce35eba21159020
Author: lipenghui <pe...@apache.org>
AuthorDate: Mon Jul 4 09:32:23 2022 +0800
[improve][broker] Use shared broker client scheduled executor provider (#16338)
(cherry picked from commit 4dd4ba5ed6de06eb22431c74863cc9a878d8a791)
---
.../src/main/java/org/apache/pulsar/broker/PulsarService.java | 6 ++++++
1 file changed, 6 insertions(+)
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 9c799603df8..2e4fbe6cf07 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
@@ -130,6 +130,7 @@ import org.apache.pulsar.client.impl.conf.ClientConfigurationData;
import org.apache.pulsar.client.impl.conf.ConfigurationDataUtils;
import org.apache.pulsar.client.internal.PropertiesUtils;
import org.apache.pulsar.client.util.ExecutorProvider;
+import org.apache.pulsar.client.util.ScheduledExecutorProvider;
import org.apache.pulsar.common.conf.InternalConfigurationData;
import org.apache.pulsar.common.configuration.PulsarConfigurationLoader;
import org.apache.pulsar.common.configuration.VipStatus;
@@ -237,6 +238,7 @@ public class PulsarService implements AutoCloseable, ShutdownService {
protected final EventLoopGroup ioEventLoopGroup;
private final ExecutorProvider brokerClientSharedInternalExecutorProvider;
private final ExecutorProvider brokerClientSharedExternalExecutorProvider;
+ private final ScheduledExecutorProvider brokerClientSharedScheduledExecutorProvider;
private final Timer brokerClientSharedTimer;
private MetricsGenerator metricsGenerator;
@@ -340,6 +342,8 @@ public class PulsarService implements AutoCloseable, ShutdownService {
// 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.brokerClientSharedScheduledExecutorProvider =
+ new ScheduledExecutorProvider(1, "broker-client-shared-scheduled-executor");
this.brokerClientSharedTimer =
new HashedWheelTimer(new DefaultThreadFactory("broker-client-shared-timer"), 1, TimeUnit.MILLISECONDS);
}
@@ -536,6 +540,7 @@ public class PulsarService implements AutoCloseable, ShutdownService {
brokerClientSharedExternalExecutorProvider.shutdownNow();
brokerClientSharedInternalExecutorProvider.shutdownNow();
+ brokerClientSharedScheduledExecutorProvider.shutdownNow();
brokerClientSharedTimer.stop();
ioEventLoopGroup.shutdownGracefully();
@@ -1371,6 +1376,7 @@ public class PulsarService implements AutoCloseable, ShutdownService {
.timer(brokerClientSharedTimer)
.internalExecutorProvider(brokerClientSharedInternalExecutorProvider)
.externalExecutorProvider(brokerClientSharedExternalExecutorProvider)
+ .scheduledExecutorProvider(brokerClientSharedScheduledExecutorProvider)
.build();
}
[pulsar] 02/03: [Broker] Use shared executors for broker and geo-replication clients (#13839)
Posted by bo...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
bogong pushed a commit to branch branch-2.9
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit a75a22ddd5fd80d447fc7ed50d07c4a291a057b5
Author: Lari Hotari <lh...@users.noreply.github.com>
AuthorDate: Thu Jan 20 20:16:19 2022 +0200
[Broker] Use shared executors for broker and geo-replication clients (#13839)
* [Broker] Use shared executors for broker clients and geo-replication clients
* Remove brokerClientNumIOThreads configuration key and default to 1
* Revisit the shared timer creation
- don't ever make it a daemon thread
(cherry picked from commit 4924e6d54a8fa4fb1a01f48644c23c625d0407f2)
---
.../org/apache/pulsar/broker/PulsarService.java | 32 +++++++++++++++++++++-
.../pulsar/broker/namespace/NamespaceService.java | 3 +-
.../pulsar/broker/service/BrokerService.java | 3 +-
3 files changed, 33 insertions(+), 5 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 6446da29071..9c799603df8 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.ChannelInitializer;
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;
@@ -234,6 +235,9 @@ public class PulsarService implements AutoCloseable, ShutdownService {
private final Consumer<Integer> processTerminator;
protected final EventLoopGroup ioEventLoopGroup;
+ private final ExecutorProvider brokerClientSharedInternalExecutorProvider;
+ private final ExecutorProvider brokerClientSharedExternalExecutorProvider;
+ private final Timer brokerClientSharedTimer;
private MetricsGenerator metricsGenerator;
@@ -326,6 +330,18 @@ public class PulsarService implements AutoCloseable, ShutdownService {
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(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");
+ this.brokerClientSharedTimer =
+ new HashedWheelTimer(new DefaultThreadFactory("broker-client-shared-timer"), 1, TimeUnit.MILLISECONDS);
}
public MetadataStore createConfigurationMetadataStore() throws MetadataStoreException {
@@ -518,6 +534,9 @@ public class PulsarService implements AutoCloseable, ShutdownService {
transactionExecutorProvider.shutdownNow();
}
+ brokerClientSharedExternalExecutorProvider.shutdownNow();
+ brokerClientSharedInternalExecutorProvider.shutdownNow();
+ brokerClientSharedTimer.stop();
ioEventLoopGroup.shutdownGracefully();
// add timeout handling for closing executors
@@ -1344,6 +1363,17 @@ public class PulsarService implements AutoCloseable, ShutdownService {
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 {
@@ -1388,7 +1418,7 @@ public class PulsarService implements AutoCloseable, ShutdownService {
}
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 fbef655d489..933e750242e 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 java.util.concurrent.TimeUnit.SECONDS;
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;
@@ -1306,7 +1305,7 @@ public class NamespaceService implements AutoCloseable {
// 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 1fbc19657f5..b2dd0785688 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
@@ -124,7 +124,6 @@ import org.apache.pulsar.client.admin.PulsarAdminBuilder;
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.client.internal.PropertiesUtils;
import org.apache.pulsar.common.allocator.PulsarByteBufAllocator;
@@ -1177,7 +1176,7 @@ public class BrokerService implements Closeable {
}
// 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);
}