[fix][broker] Remove useless load balancer items about MemoryResource…

…Weight (#19559)

Motivation
Even for a Broker with a very low load, its memory will grow slowly until GC is triggered. If memory is used as a load calculation item, the Bundle will be unloaded by mistake

Modifications
Remove memory as load calculation item

conf/broker.conf

@@ -1355,10 +1355,6 @@ loadBalancerBandwithOutResourceWeight=1.0
 # It only takes effect in the ThresholdShedder strategy.
 loadBalancerCPUResourceWeight=1.0
 
-# The heap memory usage weight when calculating new resource usage.
-# It only takes effect in the ThresholdShedder strategy.
-loadBalancerMemoryResourceWeight=1.0
-
 # The direct memory usage weight when calculating new resource usage.
 # It only takes effect in the ThresholdShedder strategy.
 loadBalancerDirectMemoryResourceWeight=1.0
@@ -1669,6 +1665,11 @@ strictBookieAffinityEnabled=false
 
 # These settings are left here for compatibility
 
+# The heap memory usage weight when calculating new resource usage.
+# It only takes effect in the ThresholdShedder strategy.
+# Deprecated: Memory is no longer used as a load balancing item
+loadBalancerMemoryResourceWeight=1.0
+
 # Zookeeper quorum connection string
 # Deprecated: use metadataStoreUrl instead
 zookeeperServers=

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

@@ -2341,10 +2341,12 @@ The delayed message index bucket time step(in seconds) in per bucket snapshot se
     )
     private double loadBalancerCPUResourceWeight = 1.0;
 
+    @Deprecated(since = "3.0.0")
     @FieldContext(
             dynamic = true,
             category = CATEGORY_LOAD_BALANCER,
-            doc = "Memory Resource Usage Weight"
+            doc = "Memory Resource Usage Weight. Deprecated: Memory is no longer used as a load balancing item.",
+            deprecated = true
     )
     private double loadBalancerMemoryResourceWeight = 1.0;
 

pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/impl/LeastResourceUsageWithWeight.java

@@ -98,7 +98,6 @@ private double updateAndGetMaxResourceUsageWithWeight(String broker, BrokerData
         }
         double resourceUsage = brokerData.getLocalData().getMaxResourceUsageWithWeight(
                 conf.getLoadBalancerCPUResourceWeight(),
-                conf.getLoadBalancerMemoryResourceWeight(),
                 conf.getLoadBalancerDirectMemoryResourceWeight(),
                 conf.getLoadBalancerBandwithInResourceWeight(),
                 conf.getLoadBalancerBandwithOutResourceWeight());

pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/impl/ThresholdShedder.java

@@ -168,7 +168,7 @@ private double updateAvgResourceUsage(String broker, LocalBrokerData localBroker
                 brokerAvgResourceUsage.get(broker);
         double resourceUsage = localBrokerData.getMaxResourceUsageWithWeight(
                 conf.getLoadBalancerCPUResourceWeight(),
-                conf.getLoadBalancerMemoryResourceWeight(), conf.getLoadBalancerDirectMemoryResourceWeight(),
+                conf.getLoadBalancerDirectMemoryResourceWeight(),
                 conf.getLoadBalancerBandwithInResourceWeight(),
                 conf.getLoadBalancerBandwithOutResourceWeight());
         historyUsage = historyUsage == null

pulsar-common/src/main/java/org/apache/pulsar/policies/data/loadbalancer/LocalBrokerData.java

@@ -249,14 +249,21 @@ public String printResourceUsage() {
                 cpu.percentUsage(), memory.percentUsage(), directMemory.percentUsage(), bandwidthIn.percentUsage(),
                 bandwidthOut.percentUsage());
     }
-
+    @Deprecated
     public double getMaxResourceUsageWithWeight(final double cpuWeight, final double memoryWeight,
                                                 final double directMemoryWeight, final double bandwidthInWeight,
                                                 final double bandwidthOutWeight) {
         return max(cpu.percentUsage() * cpuWeight, memory.percentUsage() * memoryWeight,
                 directMemory.percentUsage() * directMemoryWeight, bandwidthIn.percentUsage() * bandwidthInWeight,
                 bandwidthOut.percentUsage() * bandwidthOutWeight) / 100;
     }
+    public double getMaxResourceUsageWithWeight(final double cpuWeight,
+                                                final double directMemoryWeight, final double bandwidthInWeight,
+                                                final double bandwidthOutWeight) {
+        return max(cpu.percentUsage() * cpuWeight,
+                directMemory.percentUsage() * directMemoryWeight, bandwidthIn.percentUsage() * bandwidthInWeight,
+                bandwidthOut.percentUsage() * bandwidthOutWeight) / 100;
+    }
 
     public static double max(double... args) {
         double max = Double.NEGATIVE_INFINITY;

pulsar-common/src/test/java/org/apache/pulsar/policies/data/loadbalancer/LocalBrokerDataTest.java

@@ -43,19 +43,17 @@ public void testLocalBrokerDataDeserialization() {
     public void testMaxResourceUsage() {
         LocalBrokerData data = new LocalBrokerData();
         data.setCpu(new ResourceUsage(1.0, 100.0));
-        data.setMemory(new ResourceUsage(800.0, 200.0));
         data.setDirectMemory(new ResourceUsage(2.0, 100.0));
         data.setBandwidthIn(new ResourceUsage(3.0, 100.0));
         data.setBandwidthOut(new ResourceUsage(4.0, 100.0));
 
         double epsilon = 0.00001;
         double weight = 0.5;
         // skips memory usage
-        assertEquals(data.getMaxResourceUsage(), 0.04, epsilon);
+        assertEquals(data.getMaxResourceUsage(), data.getBandwidthOut().percentUsage() / 100, epsilon);
 
-        assertEquals(
-                data.getMaxResourceUsageWithWeight(
-                        weight, weight, weight, weight, weight), 2.0, epsilon);
+        assertEquals(data.getMaxResourceUsageWithWeight(weight, weight, weight, weight),
+                data.getBandwidthOut().percentUsage() * weight / 100, epsilon);
     }
 
     /*