[improve][misc] Sync commits from apache into 3.1_ds

conf/broker.conf

@@ -159,6 +159,14 @@ skipBrokerShutdownOnOOM=false
 # Factory class-name to create topic with custom workflow
 topicFactoryClassName=
 
+# Max capacity of the topic name cache. -1 means unlimited cache; 0 means broker will clear all cache
+# per "maxSecondsToClearTopicNameCache", it does not mean broker will not cache TopicName.
+topicNameCacheMaxCapacity=100000
+
+# A Specifies the minimum number of seconds that the topic name stays in memory, to avoid clear cache frequently when
+# there are too many topics are in use.
+maxSecondsToClearTopicNameCache=7200
+
 # Enable backlog quota check. Enforces action on topic when the quota is reached
 backlogQuotaCheckEnabled=true
 
@@ -1416,6 +1424,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

@@ -588,6 +588,21 @@ The delayed message index time step(in seconds) in per bucket snapshot segment,
     )
     private boolean backlogQuotaCheckEnabled = true;
 
+    @FieldContext(
+        dynamic = true,
+        category = CATEGORY_POLICIES,
+        doc = "Max capacity of the topic name cache. -1 means unlimited cache; 0 means broker will clear all cache"
+                + " per maxSecondsToClearTopicNameCache, it does not mean broker will not cache TopicName."
+    )
+    private int topicNameCacheMaxCapacity = 100_000;
+
+    @FieldContext(
+        category = CATEGORY_POLICIES,
+        doc = "A Specifies the minimum number of seconds that the topic name stays in memory, to avoid clear cache"
+                + " frequently when there are too many topics are in use."
+    )
+    private int maxSecondsToClearTopicNameCache = 3600 * 2;
+
     @FieldContext(
             category = CATEGORY_POLICIES,
             doc = "Whether to enable precise time based backlog quota check. "
@@ -2387,21 +2402,51 @@ The delayed message index time step(in seconds) in per bucket snapshot segment,
     @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;