[pulsar-broker] load-balancer support disabling max-session for bundl…

…e split (#13108)
apache · Dec 20, 2021 · f7abada · f7abada
1 parent c62ca80
commit f7abada
Show file tree

Hide file tree

Showing 5 changed files with 11 additions and 2 deletions.
diff --git a/conf/broker.conf b/conf/broker.conf
@@ -1074,6 +1074,7 @@ loadBalancerAutoUnloadSplitBundlesEnabled=true
 loadBalancerNamespaceBundleMaxTopics=1000
 
 # maximum sessions (producers + consumers) in a bundle, otherwise bundle split will be triggered
+# (disable threshold check with value -1)
 loadBalancerNamespaceBundleMaxSessions=1000
 
 # maximum msgRate (in + out) in a bundle, otherwise bundle split will be triggered

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
@@ -1953,6 +1953,7 @@ public class ServiceConfiguration implements PulsarConfiguration {
     @FieldContext(
         category = CATEGORY_LOAD_BALANCER,
         doc = "maximum sessions (producers + consumers) in a bundle, otherwise bundle split will be triggered"
+                + "(disable threshold check with value -1)"
     )
     private int loadBalancerNamespaceBundleMaxSessions = 1000;
     @FieldContext(

diff --git a/...ar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/impl/BundleSplitterTask.java b/...ar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/impl/BundleSplitterTask.java
@@ -84,7 +84,8 @@ public Set<String> findBundlesToSplit(final LoadData loadData, final PulsarServi
                     totalMessageRate = longTermData.totalMsgRate();
                     totalMessageThroughput = longTermData.totalMsgThroughput();
                 }
-                if (stats.topics > maxBundleTopics || stats.consumerCount + stats.producerCount > maxBundleSessions
+                if (stats.topics > maxBundleTopics || (maxBundleSessions > 0 && (stats.consumerCount
+                        + stats.producerCount > maxBundleSessions))
                         || totalMessageRate > maxBundleMsgRate || totalMessageThroughput > maxBundleBandwidth) {
                     final String namespace = LoadManagerShared.getNamespaceNameFromBundleName(bundle);
                     try {

diff --git a/...broker/src/main/java/org/apache/pulsar/broker/loadbalance/impl/SimpleLoadManagerImpl.java b/...broker/src/main/java/org/apache/pulsar/broker/loadbalance/impl/SimpleLoadManagerImpl.java
@@ -1372,7 +1372,8 @@ public void doNamespaceBundleSplit() throws Exception {
             double totalBandwidth = stats.msgThroughputIn + stats.msgThroughputOut;
 
             boolean needSplit = false;
-            if (stats.topics > maxBundleTopics || totalSessions > maxBundleSessions || totalMsgRate > maxBundleMsgRate
+            if (stats.topics > maxBundleTopics || (maxBundleSessions > 0
+                    && totalSessions > maxBundleSessions) || totalMsgRate > maxBundleMsgRate
                     || totalBandwidth > maxBundleBandwidth) {
                 if (stats.topics <= 1) {
                     log.info("Unable to split hot namespace bundle {} since there is only one topic.", bundleName);

diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/LoadBalancerTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/LoadBalancerTest.java
@@ -662,6 +662,11 @@ public void testNamespaceBundleAutoSplit() throws Exception {
                 isAutoUnooadSplitBundleEnabled, null);
         verify(namespaceAdmin, never()).splitNamespaceBundle("pulsar/use/primary-ns-10", "0x00000000_0x02000000",
                 isAutoUnooadSplitBundleEnabled, null);
+        // disable max session
+        bundleStats.put("pulsar/use/primary-ns-03/0x00000000_0x80000000",
+                newBundleStats(2, -1, 0, 0, 0, 0, 0));
+        verify(namespaceAdmin, times(0)).splitNamespaceBundle("pulsar/use/primary-ns-12", "0x00000000_0x80000000",
+                isAutoUnooadSplitBundleEnabled, null);
     }
 
     /*