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

@@ -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=
 

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
 

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;
 

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"));
     }
 
     /**