[pulsar] branch master updated: make zk cache executor pool size configurable (#7794)

[sijie]

This is an automated email from the ASF dual-hosted git repository.

sijie pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/pulsar.git


The following commit(s) were added to refs/heads/master by this push:
     new 1e8f4b6  make zk cache executor pool size configurable (#7794)
1e8f4b6 is described below

commit 1e8f4b6b78ac2bbcac1f15f81a5a29565d7b7112
Author: hangc0276 <hangc0...@163.com>
AuthorDate: Wed Aug 12 06:48:35 2020 +0800

    make zk cache executor pool size configurable (#7794)
    
    ### Motivation
    The zk cache executor thread pool size is hard code to 10 when pulsar 
service start, it should be configurable in broker.conf.
    ```
    private final ScheduledExecutorService cacheExecutor = 
Executors.newScheduledThreadPool(10,
                new DefaultThreadFactory("zk-cache-callback"));
    ```
    #### Changes
    make the zk cache executor threads pool size configurable in broker.conf
---
 conf/broker.conf                                                  | 5 +++++
 conf/standalone.conf                                              | 5 +++++
 .../main/java/org/apache/pulsar/broker/ServiceConfiguration.java  | 8 ++++++++
 .../src/main/java/org/apache/pulsar/broker/PulsarService.java     | 5 +++--
 4 files changed, 21 insertions(+), 2 deletions(-)

diff --git a/conf/broker.conf b/conf/broker.conf
index d6b78fa..05a75e7 100644
--- a/conf/broker.conf
+++ b/conf/broker.conf
@@ -71,6 +71,11 @@ numHttpServerThreads=
 # Default is Runtime.getRuntime().availableProcessors()
 numExecutorThreadPoolSize=
 
+# Number of thread pool size to use for pulsar zookeeper callback service
+# The cache executor thread pool is used for restarting global zookeeper 
session.
+# Default is 10
+numCacheExecutorThreadPoolSize=10
+
 # Flag to control features that are meant to be used when running in 
standalone mode
 isRunningStandalone=
 
diff --git a/conf/standalone.conf b/conf/standalone.conf
index 7d6bfba..d1e0315 100644
--- a/conf/standalone.conf
+++ b/conf/standalone.conf
@@ -52,6 +52,11 @@ numHttpServerThreads=
 # Default is Runtime.getRuntime().availableProcessors()
 numExecutorThreadPoolSize=
 
+# Number of thread pool size to use for pulsar zookeeper callback service
+# The cache executor thread pool is used for restarting global zookeeper 
session.
+# Default is 10
+numCacheExecutorThreadPoolSize=10
+
 # Name of the cluster to which this broker belongs to
 clusterName=standalone
 
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 f6a0ba8..84e10a0 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
@@ -197,6 +197,14 @@ public class ServiceConfiguration implements 
PulsarConfiguration {
     )
     private int numExecutorThreadPoolSize = 
Runtime.getRuntime().availableProcessors();
 
+    @FieldContext(
+        category = CATEGORY_SERVER,
+        doc = "Number of thread pool size to use for pulsar zookeeper callback 
service."
+            + "The cache executor thread pool is used for restarting global 
zookeeper session. "
+            + "Default is 10"
+    )
+    private int numCacheExecutorThreadPoolSize = 10;
+
     @FieldContext(category = CATEGORY_SERVER, doc = "Whether to enable the 
delayed delivery for messages.")
     private boolean delayedDeliveryEnabled = true;
 
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 8a639c7..55c0127 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
@@ -176,9 +176,8 @@ public class PulsarService implements AutoCloseable {
     private Compactor compactor;
 
     private final ScheduledExecutorService executor;
+    private final ScheduledExecutorService cacheExecutor;
 
-    private final ScheduledExecutorService cacheExecutor = 
Executors.newScheduledThreadPool(10,
-            new DefaultThreadFactory("zk-cache-callback"));
     private OrderedExecutor orderedExecutor;
     private final ScheduledExecutorService loadManagerExecutor;
     private ScheduledExecutorService compactorExecutor;
@@ -260,6 +259,8 @@ public class PulsarService implements AutoCloseable {
         this.functionWorkerService = functionWorkerService;
         this.executor = 
Executors.newScheduledThreadPool(config.getNumExecutorThreadPoolSize(),
                 new DefaultThreadFactory("pulsar"));
+        this.cacheExecutor = 
Executors.newScheduledThreadPool(config.getNumCacheExecutorThreadPoolSize(),
+                new DefaultThreadFactory("zk-cache-callback"));
     }
 
     /**