[improve][broker] PIP-275:Introduce topicOrderedExecutorThreadNum to …

…deprecate numWorkerThreadsForNonPersistentTopic (#20504)

PIP: #20507

<!-- Details of when a PIP is required and how the PIP process work, please see: https://github.com/apache/pulsar/blob/master/pip/README.md -->

### Motivation

See #20507

### Modifications

Introduce `topicOrderedExecutorThreadNum ` to deprecate `numWorkerThreadsForNonPersistentTopic`

conf/broker.conf

@@ -474,8 +474,8 @@ maxConcurrentTopicLoadRequest=5000
 # Max concurrent non-persistent message can be processed per connection
 maxConcurrentNonPersistentMessagePerConnection=1000
 
-# Number of worker threads to serve non-persistent topic
-numWorkerThreadsForNonPersistentTopic=
+# Number of worker threads to serve topic ordered executor
+topicOrderedExecutorThreadNum=
 
 # Enable broker to load persistent topics
 enablePersistentTopics=true
@@ -1816,3 +1816,7 @@ persistentUnackedRangesWithMultipleEntriesEnabled=false
 
 # Deprecated - Use managedLedgerCacheEvictionIntervalMs instead
 managedLedgerCacheEvictionFrequency=0
+
+# Number of worker threads to serve non-persistent topic
+# Deprecated - use topicOrderedExecutorThreadNum instead.
+numWorkerThreadsForNonPersistentTopic=

conf/standalone.conf

@@ -281,8 +281,8 @@ maxConcurrentTopicLoadRequest=5000
 # Max concurrent non-persistent message can be processed per connection
 maxConcurrentNonPersistentMessagePerConnection=1000
 
-# Number of worker threads to serve non-persistent topic
-numWorkerThreadsForNonPersistentTopic=8
+# Number of worker threads to serve topic ordered executor
+topicOrderedExecutorThreadNum=8
 
 # Enable broker to load persistent topics
 enablePersistentTopics=true
@@ -1201,6 +1201,10 @@ functionsWorkerEnablePackageManagement=false
 
 # These settings are left here for compatibility
 
+# Number of worker threads to serve non-persistent topic
+# Deprecated - use topicOrderedExecutorThreadNum instead.
+numWorkerThreadsForNonPersistentTopic=8
+
 # Zookeeper session timeout in milliseconds
 # Deprecated: use metadataStoreSessionTimeoutMillis
 zooKeeperSessionTimeoutMillis=-1

deployment/terraform-ansible/templates/broker.conf

@@ -332,8 +332,8 @@ maxConcurrentTopicLoadRequest=5000
 # Max concurrent non-persistent message can be processed per connection
 maxConcurrentNonPersistentMessagePerConnection=1000
 
-# Number of worker threads to serve non-persistent topic
-numWorkerThreadsForNonPersistentTopic=8
+# Number of worker threads to serve topic ordered executor
+topicOrderedExecutorThreadNum=8
 
 # Enable broker to load persistent topics
 enablePersistentTopics=true
@@ -1127,6 +1127,10 @@ fileSystemURI=
 
 ### --- Deprecated config variables --- ###
 
+# Number of worker threads to serve non-persistent topic
+# Deprecated - use topicOrderedExecutorThreadNum instead.
+numWorkerThreadsForNonPersistentTopic=8
+
 # Deprecated. Use configurationStoreServers
 globalZookeeperServers={{ zookeeper_servers }}
 

pulsar-broker-common/src/main/java/org/apache/pulsar/broker/ServiceConfiguration.java

@@ -1190,10 +1190,18 @@ The delayed message index time step(in seconds) in per bucket snapshot segment,
         category = CATEGORY_SERVER,
         doc = "Max concurrent non-persistent message can be processed per connection")
     private int maxConcurrentNonPersistentMessagePerConnection = 1000;
+
+    @Deprecated
     @FieldContext(
         category = CATEGORY_SERVER,
-        doc = "Number of worker threads to serve non-persistent topic")
-    private int numWorkerThreadsForNonPersistentTopic = Runtime.getRuntime().availableProcessors();
+        deprecated = true,
+        doc = "Number of worker threads to serve non-persistent topic.\n"
+                + "@deprecated - use topicOrderedExecutorThreadNum instead.")
+    private int numWorkerThreadsForNonPersistentTopic = -1;
+    @FieldContext(
+            category = CATEGORY_SERVER,
+            doc = "Number of worker threads to serve topic ordered executor")
+    private int topicOrderedExecutorThreadNum = Runtime.getRuntime().availableProcessors();
 
     @FieldContext(
         category = CATEGORY_SERVER,
@@ -3474,4 +3482,9 @@ public long getManagedLedgerCacheEvictionIntervalMs() {
                                    MIN_ML_CACHE_EVICTION_FREQUENCY))
                 : Math.min(MAX_ML_CACHE_EVICTION_INTERVAL_MS, managedLedgerCacheEvictionIntervalMs);
     }
+
+    public int getTopicOrderedExecutorThreadNum() {
+        return numWorkerThreadsForNonPersistentTopic > 0
+                ? numWorkerThreadsForNonPersistentTopic : topicOrderedExecutorThreadNum;
+    }
 }

pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java

@@ -316,7 +316,7 @@ public BrokerService(PulsarService pulsar, EventLoopGroup eventLoopGroup) throws
                         PersistentOfflineTopicStats>newBuilder().build();
 
         this.topicOrderedExecutor = OrderedExecutor.newBuilder()
-                .numThreads(pulsar.getConfiguration().getNumWorkerThreadsForNonPersistentTopic())
+                .numThreads(pulsar.getConfiguration().getTopicOrderedExecutorThreadNum())
                 .name("broker-topic-workers").build();
         final DefaultThreadFactory acceptorThreadFactory =
                 new ExecutorProvider.ExtendedThreadFactory("pulsar-acceptor");

pulsar-broker/src/test/java/org/apache/pulsar/broker/EmbeddedPulsarCluster.java

@@ -112,7 +112,7 @@ private ServiceConfiguration getConf() {
         conf.setWebServicePort(Optional.of(0));
         conf.setNumExecutorThreadPoolSize(1);
         conf.setNumCacheExecutorThreadPoolSize(1);
-        conf.setNumWorkerThreadsForNonPersistentTopic(1);
+        conf.setTopicOrderedExecutorThreadNum(1);
         conf.setNumIOThreads(1);
         conf.setNumOrderedExecutorThreads(1);
         conf.setBookkeeperClientNumWorkerThreads(1);