Allow to configure the number of BK client worker threads (#10649)

apache · May 21, 2021 · 16223ba · 16223ba
1 parent 7814117
commit 16223ba
Show file tree

Hide file tree

Showing 5 changed files with 18 additions and 0 deletions.
diff --git a/conf/broker.conf b/conf/broker.conf
@@ -715,6 +715,10 @@ bookkeeperClientAuthenticationParameters=
 # Timeout for BK add / read operations
 bookkeeperClientTimeoutInSeconds=30
 
+# Number of BookKeeper client worker threads
+# Default is Runtime.getRuntime().availableProcessors()
+bookkeeperClientNumWorkerThreads=
+
 # Speculative reads are initiated if a read request doesn't complete within a certain time
 # Using a value of 0, is disabling the speculative reads
 bookkeeperClientSpeculativeReadTimeoutInMillis=0

diff --git a/conf/standalone.conf b/conf/standalone.conf
@@ -477,6 +477,10 @@ bookkeeperClientAuthenticationParameters=
 # Timeout for BK add / read operations
 bookkeeperClientTimeoutInSeconds=30
 
+# Number of BookKeeper client worker threads
+# Default is Runtime.getRuntime().availableProcessors()
+bookkeeperClientNumWorkerThreads=
+
 # Speculative reads are initiated if a read request doesn't complete within a certain time
 # Using a value of 0, is disabling the speculative reads
 bookkeeperClientSpeculativeReadTimeoutInMillis=0

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
@@ -1322,6 +1322,13 @@ public class ServiceConfiguration implements PulsarConfiguration {
     )
     private int bookkeeperClientThrottleValue = 0;
 
+    @FieldContext(
+            category = CATEGORY_STORAGE_BK,
+            doc = "Number of BookKeeper client worker threads. Default is Runtime.getRuntime().availableProcessors()"
+    )
+    private int bookkeeperClientNumWorkerThreads = Runtime.getRuntime().availableProcessors();
+
+
     /**** --- Managed Ledger --- ****/
     @FieldContext(
         minValue = 1,

diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/BookKeeperClientFactoryImpl.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/BookKeeperClientFactoryImpl.java
@@ -111,6 +111,7 @@ ClientConfiguration createBkClientConfiguration(ServiceConfiguration conf) {
             bkConf.setTLSTrustStorePasswordPath(conf.getBookkeeperTLSTrustStorePasswordPath());
         }
 
+        bkConf.setNumWorkerThreads(conf.getBookkeeperClientNumWorkerThreads());
         bkConf.setThrottleValue(conf.getBookkeeperClientThrottleValue());
         bkConf.setAddEntryTimeout((int) conf.getBookkeeperClientTimeoutInSeconds());
         bkConf.setReadEntryTimeout((int) conf.getBookkeeperClientTimeoutInSeconds());

diff --git a/site2/docs/reference-configuration.md b/site2/docs/reference-configuration.md
@@ -246,6 +246,7 @@ brokerServiceCompactionThresholdInBytes|If the estimated backlog size is greater
 |bookkeeperClientAuthenticationPlugin|  Authentication plugin to use when connecting to bookies ||
 |bookkeeperClientAuthenticationParametersName|  BookKeeper auth plugin implementatation specifics parameters name and values  ||
 |bookkeeperClientAuthenticationParameters|||
+|bookkeeperClientNumWorkerThreads|  Number of BookKeeper client worker threads. Default is Runtime.getRuntime().availableProcessors()  ||
 |bookkeeperClientTimeoutInSeconds|  Timeout for BK add / read operations  |30|
 |bookkeeperClientSpeculativeReadTimeoutInMillis|  Speculative reads are initiated if a read request doesn’t complete within a certain time Using a value of 0, is disabling the speculative reads |0|
 |bookkeeperNumberOfChannelsPerBookie|  Number of channels per bookie  |16|
@@ -566,6 +567,7 @@ You can set the log level and configuration in the  [log4j2.yaml](https://github
 |bookkeeperClientAuthenticationPlugin|  Authentication plugin to be used when connecting to bookies (BookKeeper servers). ||
 |bookkeeperClientAuthenticationParametersName|  BookKeeper authentication plugin implementation parameters and values.  ||
 |bookkeeperClientAuthenticationParameters|  Parameters associated with the bookkeeperClientAuthenticationParametersName ||
+|bookkeeperClientNumWorkerThreads|  Number of BookKeeper client worker threads. Default is Runtime.getRuntime().availableProcessors()  ||
 |bookkeeperClientTimeoutInSeconds|  Timeout for BookKeeper add and read operations. |30|
 |bookkeeperClientSpeculativeReadTimeoutInMillis|  Speculative reads are initiated if a read request doesn’t complete within a certain time. A value of 0 disables speculative reads.  |0|
 |bookkeeperUseV2WireProtocol|Use older Bookkeeper wire protocol with bookie.|true|