[improve][broker] PIP-364: Introduce a new load balance algorithm AvgShedder

conf/broker.conf

@@ -1414,6 +1414,25 @@ loadBalancerBundleUnloadMinThroughputThreshold=10
 # Time to wait for the unloading of a namespace bundle
 namespaceBundleUnloadingTimeoutMs=60000
 
+# configuration for AvgShedder, a new shedding and placement strategy
+# The low threshold for the difference between the highest and lowest loaded brokers.
+loadBalancerAvgShedderLowThreshold = 15
+
+# The high threshold for the difference between the highest and lowest loaded brokers.
+loadBalancerAvgShedderHighThreshold = 40
+
+# The number of times the low threshold is triggered before the bundle is unloaded.
+loadBalancerAvgShedderHitCountLowThreshold = 8
+
+# The number of times the high threshold is triggered before the bundle is unloaded.
+loadBalancerAvgShedderHitCountHighThreshold = 2
+
+# In the UniformLoadShedder and AvgShedder strategy, the maximum unload ratio.
+# For AvgShedder, recommend to set to 0.5, so that it will distribute the load evenly
+#  between the highest and lowest brokers.
+maxUnloadPercentage = 0.2
+
+
 ### --- Load balancer extension --- ###
 
 # Option to enable the debug mode for the load balancer logics.

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

@@ -2395,21 +2395,51 @@ The max allowed delay for delayed delivery (in milliseconds). If the broker rece
     @FieldContext(
             dynamic = true,
             category = CATEGORY_LOAD_BALANCER,
-            doc = "In the UniformLoadShedder strategy, the minimum message that triggers unload."
+            doc = "The low threshold for the difference between the highest and lowest loaded brokers."
+    )
+    private int loadBalancerAvgShedderLowThreshold = 15;
+
+    @FieldContext(
+            dynamic = true,
+            category = CATEGORY_LOAD_BALANCER,
+            doc = "The high threshold for the difference between the highest and lowest loaded brokers."
+    )
+    private int loadBalancerAvgShedderHighThreshold = 40;
+
+    @FieldContext(
+            dynamic = true,
+            category = CATEGORY_LOAD_BALANCER,
+            doc = "The number of times the low threshold is triggered before the bundle is unloaded."
+    )
+    private int loadBalancerAvgShedderHitCountLowThreshold = 8;
+
+    @FieldContext(
+            dynamic = true,
+            category = CATEGORY_LOAD_BALANCER,
+            doc = "The number of times the high threshold is triggered before the bundle is unloaded."
+    )
+    private int loadBalancerAvgShedderHitCountHighThreshold = 2;
+
+    @FieldContext(
+            dynamic = true,
+            category = CATEGORY_LOAD_BALANCER,
+            doc = "In the UniformLoadShedder and AvgShedder strategy, the minimum message that triggers unload."
     )
     private int minUnloadMessage = 1000;
 
     @FieldContext(
             dynamic = true,
             category = CATEGORY_LOAD_BALANCER,
-            doc = "In the UniformLoadShedder strategy, the minimum throughput that triggers unload."
+            doc = "In the UniformLoadShedder and AvgShedder strategy, the minimum throughput that triggers unload."
     )
     private int minUnloadMessageThroughput = 1 * 1024 * 1024;
 
     @FieldContext(
             dynamic = true,
             category = CATEGORY_LOAD_BALANCER,
-            doc = "In the UniformLoadShedder strategy, the maximum unload ratio."
+            doc = "In the UniformLoadShedder and AvgShedder strategy, the maximum unload ratio."
+                    + "For AvgShedder, recommend to set to 0.5, so that it will distribute the load "
+                    + "evenly between the highest and lowest brokers."
     )
     private double maxUnloadPercentage = 0.2;