From fa33461a9b9bda1d2ce258f1332a35c4f02e23bb Mon Sep 17 00:00:00 2001 From: feynmanlin <315157973@qq.com> Date: Sun, 4 Sep 2022 15:22:54 +0800 Subject: [PATCH 01/11] Add a new RangeThresholdShedder --- .../impl/RangeThresholdShedder.java | 107 ++++++++++++++++ .../loadbalance/impl/ThresholdShedder.java | 60 +++++---- .../RangeThresholdShedderTest.java | 116 ++++++++++++++++++ .../impl/ThresholdShedderTest.java | 2 +- 4 files changed, 258 insertions(+), 27 deletions(-) create mode 100644 pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/impl/RangeThresholdShedder.java create mode 100644 pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/RangeThresholdShedderTest.java diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/impl/RangeThresholdShedder.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/impl/RangeThresholdShedder.java new file mode 100644 index 0000000000000..30ed6bce992de --- /dev/null +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/impl/RangeThresholdShedder.java @@ -0,0 +1,107 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +package org.apache.pulsar.broker.loadbalance.impl; + +import com.google.common.collect.Multimap; +import java.util.Map; +import lombok.extern.slf4j.Slf4j; +import org.apache.commons.lang3.tuple.Pair; +import org.apache.pulsar.broker.ServiceConfiguration; +import org.apache.pulsar.broker.loadbalance.LoadData; +import org.apache.pulsar.policies.data.loadbalancer.BrokerData; +import org.apache.pulsar.policies.data.loadbalancer.LocalBrokerData; + +/** + * On the basis of ThresholdShedder, RangeThresholdShedder adds the lower boundary judgment of the load. + * When 【current usage < average usage - threshold】, the broker with the highest load will be triggered to unload, + * avoiding the following scenarios: + * There are 11 Brokers, of which 10 are loaded at 80% and 1 is loaded at 0%. + * The average load is 80 * 10 / 11 = 72.73, and the threshold to unload is 72.73 + 10 = 82.73. + * Since 80 < 82.73, unload will not be trigger, and there is one idle Broker with load of 0%. + */ +@Slf4j +public class RangeThresholdShedder extends ThresholdShedder { + + @Override + public Multimap findBundlesForUnloading(LoadData loadData, ServiceConfiguration conf) { + super.findBundlesForUnloading(loadData, conf); + // Return if the bundle to unload has already been selected. + if (!selectedBundlesCache.isEmpty()) { + return selectedBundlesCache; + } + // Select the broker with the most resource usage. + final double threshold = conf.getLoadBalancerBrokerThresholdShedderPercentage() / 100.0; + final double avgUsage = getBrokerAvgUsage(loadData, conf); + Pair result = getMaxUsageBroker(loadData, threshold, avgUsage); + boolean hasBrokerBelowLowerBound = result.getLeft(); + String maxUsageBroker = result.getRight(); + BrokerData brokerData = loadData.getBrokerData().get(maxUsageBroker); + if (brokerData == null || brokerData.getLocalData() == null || + brokerData.getLocalData().getBundles().size() <= 1) { + log.info("Load data is null or bundle <=1, broker name is {}, skipping bundle unload.", maxUsageBroker); + return selectedBundlesCache; + } + if (!hasBrokerBelowLowerBound) { + log.info("No broker is below the lower bound, threshold is {}, " + + "avgUsage usage is {}, max usage of Broker {} is {}", + threshold, avgUsage, maxUsageBroker, + brokerAvgResourceUsage.getOrDefault(maxUsageBroker, 0.0)); + return selectedBundlesCache; + } + LocalBrokerData localData = brokerData.getLocalData(); + double minimumThroughputToOffload = getMinimumThroughputToOffload(threshold, localData); + final double minThroughputThreshold = conf.getLoadBalancerBundleUnloadMinThroughputThreshold() * MB; + if (minThroughputThreshold > minimumThroughputToOffload) { + log.info("broker {} in RangeThresholdShedder is planning to shed throughput {} MByte/s less than " + + "minimumThroughputThreshold {} MByte/s, skipping bundle unload.", + maxUsageBroker, minimumThroughputToOffload / MB, minThroughputThreshold / MB); + return selectedBundlesCache; + } + super.filterAndSelectBundle(loadData, loadData.getRecentlyUnloadedBundles(), maxUsageBroker, localData, + minimumThroughputToOffload); + return selectedBundlesCache; + } + + private Pair getMaxUsageBroker( + LoadData loadData, double threshold, double avgUsage) { + String maxUsageBrokerName = ""; + double maxUsage = -1; + boolean hasBrokerBelowLowerBound = false; + for (Map.Entry entry : loadData.getBrokerData().entrySet()) { + String broker = entry.getKey(); + double currentUsage = brokerAvgResourceUsage.getOrDefault(broker, 0.0); + // Select the broker with the most resource usage. + if (currentUsage > maxUsage) { + maxUsage = currentUsage; + maxUsageBrokerName = broker; + } + // Whether any brokers with low usage in the cluster. + if (currentUsage < avgUsage - threshold) { + hasBrokerBelowLowerBound = true; + } + } + return Pair.of(hasBrokerBelowLowerBound, maxUsageBrokerName); + } + + private double getMinimumThroughputToOffload(double threshold, LocalBrokerData localData) { + double brokerCurrentThroughput = localData.getMsgThroughputIn() + localData.getMsgThroughputOut(); + return brokerCurrentThroughput * threshold; + } + +} diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/impl/ThresholdShedder.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/impl/ThresholdShedder.java index 586a2fe101269..fd5b8c7ca70a4 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/impl/ThresholdShedder.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/impl/ThresholdShedder.java @@ -50,10 +50,10 @@ */ public class ThresholdShedder implements LoadSheddingStrategy { private static final Logger log = LoggerFactory.getLogger(ThresholdShedder.class); - private final Multimap selectedBundlesCache = ArrayListMultimap.create(); - private static final double ADDITIONAL_THRESHOLD_PERCENT_MARGIN = 0.05; - private static final double MB = 1024 * 1024; - private final Map brokerAvgResourceUsage = new HashMap<>(); + protected final Multimap selectedBundlesCache = ArrayListMultimap.create(); + public static final double ADDITIONAL_THRESHOLD_PERCENT_MARGIN = 0.05; + public static final double MB = 1024 * 1024; + protected final Map brokerAvgResourceUsage = new HashMap<>(); @Override public Multimap findBundlesForUnloading(final LoadData loadData, final ServiceConfiguration conf) { @@ -62,7 +62,7 @@ public Multimap findBundlesForUnloading(final LoadData loadData, final Map recentlyUnloadedBundles = loadData.getRecentlyUnloadedBundles(); final double minThroughputThreshold = conf.getLoadBalancerBundleUnloadMinThroughputThreshold() * MB; - final double avgUsage = getBrokerAvgUsage(loadData, conf.getLoadBalancerHistoryResourcePercentage(), conf); + final double avgUsage = getBrokerAvgUsage(loadData, conf); if (avgUsage == 0) { log.warn("average max resource usage is 0"); @@ -100,17 +100,32 @@ public Multimap findBundlesForUnloading(final LoadData loadData, broker, 100 * currentUsage, 100 * avgUsage, 100 * threshold, minimumThroughputToOffload / MB, (brokerCurrentThroughput - minimumThroughputToOffload) / MB); - MutableDouble trafficMarkedToOffload = new MutableDouble(0); - MutableBoolean atLeastOneBundleSelected = new MutableBoolean(false); - if (localData.getBundles().size() > 1) { - loadData.getBundleDataForLoadShedding().entrySet().stream() - .map((e) -> { - String bundle = e.getKey(); - BundleData bundleData = e.getValue(); - TimeAverageMessageData shortTermData = bundleData.getShortTermData(); - double throughput = shortTermData.getMsgThroughputIn() + shortTermData.getMsgThroughputOut(); - return Pair.of(bundle, throughput); + filterAndSelectBundle(loadData, recentlyUnloadedBundles, broker, localData, minimumThroughputToOffload); + } else if (localData.getBundles().size() == 1) { + log.warn( + "HIGH USAGE WARNING : Sole namespace bundle {} is overloading broker {}. " + + "No Load Shedding will be done on this broker", + localData.getBundles().iterator().next(), broker); + } else { + log.warn("Broker {} is overloaded despite having no bundles", broker); + } + }); + + return selectedBundlesCache; + } + + protected void filterAndSelectBundle(LoadData loadData, Map recentlyUnloadedBundles, String broker, + LocalBrokerData localData, double minimumThroughputToOffload) { + MutableDouble trafficMarkedToOffload = new MutableDouble(0); + MutableBoolean atLeastOneBundleSelected = new MutableBoolean(false); + loadData.getBundleDataForLoadShedding().entrySet().stream() + .map((e) -> { + String bundle = e.getKey(); + BundleData bundleData = e.getValue(); + TimeAverageMessageData shortTermData = bundleData.getShortTermData(); + double throughput = shortTermData.getMsgThroughputIn() + shortTermData.getMsgThroughputOut(); + return Pair.of(bundle, throughput); }).filter(e -> !recentlyUnloadedBundles.containsKey(e.getLeft()) ).filter(e -> @@ -125,20 +140,13 @@ public Multimap findBundlesForUnloading(final LoadData loadData, atLeastOneBundleSelected.setTrue(); } }); - } else if (localData.getBundles().size() == 1) { - log.warn( - "HIGH USAGE WARNING : Sole namespace bundle {} is overloading broker {}. " - + "No Load Shedding will be done on this broker", - localData.getBundles().iterator().next(), broker); - } else { - log.warn("Broker {} is overloaded despite having no bundles", broker); - } - }); + } - return selectedBundlesCache; + protected double getBrokerAvgUsage(LoadData loadData, ServiceConfiguration conf) { + return getBrokerAvgUsage(loadData, conf.getLoadBalancerHistoryResourcePercentage(), conf); } - private double getBrokerAvgUsage(final LoadData loadData, final double historyPercentage, + protected double getBrokerAvgUsage(final LoadData loadData, final double historyPercentage, final ServiceConfiguration conf) { double totalUsage = 0.0; int totalBrokers = 0; diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/RangeThresholdShedderTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/RangeThresholdShedderTest.java new file mode 100644 index 0000000000000..7fc93ccf62360 --- /dev/null +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/RangeThresholdShedderTest.java @@ -0,0 +1,116 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +package org.apache.pulsar.broker.loadbalance; + +import static org.testng.Assert.assertFalse; +import static org.testng.Assert.assertTrue; +import com.google.common.collect.Multimap; +import lombok.extern.slf4j.Slf4j; +import org.apache.pulsar.broker.ServiceConfiguration; +import org.apache.pulsar.broker.loadbalance.impl.RangeThresholdShedder; +import org.apache.pulsar.broker.loadbalance.impl.ThresholdShedder; +import org.apache.pulsar.broker.loadbalance.impl.ThresholdShedderTest; +import org.apache.pulsar.policies.data.loadbalancer.BrokerData; +import org.apache.pulsar.policies.data.loadbalancer.BundleData; +import org.apache.pulsar.policies.data.loadbalancer.LocalBrokerData; +import org.apache.pulsar.policies.data.loadbalancer.ResourceUsage; +import org.apache.pulsar.policies.data.loadbalancer.TimeAverageMessageData; +import org.testng.annotations.BeforeMethod; +import org.testng.annotations.Test; + +@Test(groups = "broker") +@Slf4j +public class RangeThresholdShedderTest extends ThresholdShedderTest { + private final ServiceConfiguration conf = new ServiceConfiguration(); + + @BeforeMethod + public void setup() { + super.thresholdShedder = new RangeThresholdShedder(); + } + + @Test + public void testRangeThroughput() { + int numBundles = 10; + int brokerNum = 11; + int lowLoadNode = 10; + LoadData loadData = new LoadData(); + double throughput = 50 * ThresholdShedder.MB; + //There are 11 Brokers, of which 10 are loaded at 80% and 1 is loaded at 0%. + //At this time, the average load is 80*10/11 = 72.73, and the threshold for rebalancing is 72.73 + 10 = 82.73. + //Since 80 < 82.73, rebalancing will not be trigger, and there is one Broker with load of 0. + for (int i = 0; i < brokerNum; i++) { + LocalBrokerData broker = new LocalBrokerData(); + for (int j = 0; j < numBundles; j++) { + broker.getBundles().add("bundle-" + j); + BundleData bundle = new BundleData(); + TimeAverageMessageData timeAverageMessageData = new TimeAverageMessageData(); + timeAverageMessageData.setMsgThroughputIn(i == lowLoadNode ? 0 : throughput); + timeAverageMessageData.setMsgThroughputOut(i == lowLoadNode ? 0 : throughput); + bundle.setShortTermData(timeAverageMessageData); + String broker2BundleName = "broker-" + i + "-bundle-" + (numBundles + i); + loadData.getBundleData().put(broker2BundleName, bundle); + broker.getBundles().add(broker2BundleName); + } + broker.setBandwidthIn(new ResourceUsage(i == lowLoadNode ? 0 : 80, 100)); + broker.setBandwidthOut(new ResourceUsage(i == lowLoadNode ? 0 : 80, 100)); + broker.setMsgThroughputIn(i == lowLoadNode ? 0 : throughput); + broker.setMsgThroughputOut(i == lowLoadNode ? 0 : throughput); + loadData.getBrokerData().put("broker-" + i, new BrokerData(broker)); + } + ThresholdShedder shedder = new ThresholdShedder(); + Multimap bundlesToUnload = shedder.findBundlesForUnloading(loadData, conf); + assertTrue(bundlesToUnload.isEmpty()); + bundlesToUnload = thresholdShedder.findBundlesForUnloading(loadData, conf); + assertFalse(bundlesToUnload.isEmpty()); + } + + @Test + public void testNoBrokerToOffload() { + int numBundles = 10; + int brokerNum = 11; + LoadData loadData = new LoadData(); + double throughput = 80 * ThresholdShedder.MB; + //Load of all Brokers are 80%, and no Broker needs to offload. + for (int i = 0; i < brokerNum; i++) { + LocalBrokerData broker = new LocalBrokerData(); + for (int j = 0; j < numBundles; j++) { + broker.getBundles().add("bundle-" + j); + BundleData bundle = new BundleData(); + TimeAverageMessageData timeAverageMessageData = new TimeAverageMessageData(); + timeAverageMessageData.setMsgThroughputIn(throughput); + timeAverageMessageData.setMsgThroughputOut(throughput); + bundle.setShortTermData(timeAverageMessageData); + String broker2BundleName = "broker-" + i + "-bundle-" + (numBundles + i); + loadData.getBundleData().put(broker2BundleName, bundle); + broker.getBundles().add(broker2BundleName); + } + broker.setBandwidthIn(new ResourceUsage(80, 100)); + broker.setBandwidthOut(new ResourceUsage(80, 100)); + broker.setMsgThroughputIn(throughput); + broker.setMsgThroughputOut(throughput); + loadData.getBrokerData().put("broker-" + i, new BrokerData(broker)); + } + ThresholdShedder shedder = new ThresholdShedder(); + Multimap bundlesToUnload = shedder.findBundlesForUnloading(loadData, conf); + assertTrue(bundlesToUnload.isEmpty()); + bundlesToUnload = thresholdShedder.findBundlesForUnloading(loadData, conf); + assertTrue(bundlesToUnload.isEmpty()); + } + +} diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/impl/ThresholdShedderTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/impl/ThresholdShedderTest.java index 4af7d784909b2..60f4d3cec6475 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/impl/ThresholdShedderTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/impl/ThresholdShedderTest.java @@ -38,7 +38,7 @@ @Test(groups = "broker") @Slf4j public class ThresholdShedderTest { - private ThresholdShedder thresholdShedder; + protected ThresholdShedder thresholdShedder; private final ServiceConfiguration conf; public ThresholdShedderTest() { From 336454cedb3eff35d2d88b0c0b5191225a16ac0d Mon Sep 17 00:00:00 2001 From: feynmanlin <315157973@qq.com> Date: Sun, 4 Sep 2022 16:38:29 +0800 Subject: [PATCH 02/11] check style --- .../broker/loadbalance/impl/RangeThresholdShedder.java | 8 ++++---- .../pulsar/broker/loadbalance/impl/ThresholdShedder.java | 8 ++------ .../broker/loadbalance/RangeThresholdShedderTest.java | 2 +- 3 files changed, 7 insertions(+), 11 deletions(-) diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/impl/RangeThresholdShedder.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/impl/RangeThresholdShedder.java index 30ed6bce992de..3abb15f86ca27 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/impl/RangeThresholdShedder.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/impl/RangeThresholdShedder.java @@ -7,7 +7,7 @@ * "License"); you may not use this file except in compliance * with the License. You may obtain a copy of the License at * - * http://www.apache.org/licenses/LICENSE-2.0 + * http://www.apache.org/licenses/LICENSE-2.0 * * Unless required by applicable law or agreed to in writing, * software distributed under the License is distributed on an @@ -47,13 +47,13 @@ public Multimap findBundlesForUnloading(LoadData loadData, Servi } // Select the broker with the most resource usage. final double threshold = conf.getLoadBalancerBrokerThresholdShedderPercentage() / 100.0; - final double avgUsage = getBrokerAvgUsage(loadData, conf); + final double avgUsage = getBrokerAvgUsage(loadData, conf, super.canSampleLog()); Pair result = getMaxUsageBroker(loadData, threshold, avgUsage); boolean hasBrokerBelowLowerBound = result.getLeft(); String maxUsageBroker = result.getRight(); BrokerData brokerData = loadData.getBrokerData().get(maxUsageBroker); - if (brokerData == null || brokerData.getLocalData() == null || - brokerData.getLocalData().getBundles().size() <= 1) { + if (brokerData == null || brokerData.getLocalData() == null + || brokerData.getLocalData().getBundles().size() <= 1) { log.info("Load data is null or bundle <=1, broker name is {}, skipping bundle unload.", maxUsageBroker); return selectedBundlesCache; } diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/impl/ThresholdShedder.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/impl/ThresholdShedder.java index d04df3cf4df1d..c1ab26d4483a7 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/impl/ThresholdShedder.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/impl/ThresholdShedder.java @@ -63,7 +63,7 @@ private static int toPercentage(double usage) { return (int) (usage * 100); } - private boolean canSampleLog() { + protected boolean canSampleLog() { long now = System.currentTimeMillis() / 1000; boolean sampleLog = now - lastSampledLoadLogTS >= LOAD_LOG_SAMPLE_DELAY_IN_SEC; if (sampleLog) { @@ -163,11 +163,7 @@ protected void filterAndSelectBundle(LoadData loadData, Map recent }); } - protected double getBrokerAvgUsage(LoadData loadData, ServiceConfiguration conf) { - return getBrokerAvgUsage(loadData, conf, canSampleLog()); - } - - private double getBrokerAvgUsage(final LoadData loadData, + protected double getBrokerAvgUsage(final LoadData loadData, final ServiceConfiguration conf, boolean sampleLog) { double historyPercentage = conf.getLoadBalancerHistoryResourcePercentage(); double totalUsage = 0.0; diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/RangeThresholdShedderTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/RangeThresholdShedderTest.java index 7fc93ccf62360..dcea19417ad5e 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/RangeThresholdShedderTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/RangeThresholdShedderTest.java @@ -7,7 +7,7 @@ * "License"); you may not use this file except in compliance * with the License. You may obtain a copy of the License at * - * http://www.apache.org/licenses/LICENSE-2.0 + * http://www.apache.org/licenses/LICENSE-2.0 * * Unless required by applicable law or agreed to in writing, * software distributed under the License is distributed on an From 712d7f4418c4a11aa89503aaae50a019dc574514 Mon Sep 17 00:00:00 2001 From: feynmanlin <315157973@qq.com> Date: Mon, 5 Sep 2022 22:53:04 +0800 Subject: [PATCH 03/11] Merge two shedder and address comment --- .../pulsar/broker/ServiceConfiguration.java | 7 ++ .../impl/RangeThresholdShedder.java | 107 ---------------- .../loadbalance/impl/ThresholdShedder.java | 72 ++++++++++- .../RangeThresholdShedderTest.java | 116 ------------------ .../impl/ThresholdShedderTest.java | 74 ++++++++++- 5 files changed, 146 insertions(+), 230 deletions(-) delete mode 100644 pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/impl/RangeThresholdShedder.java delete mode 100644 pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/RangeThresholdShedderTest.java 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 index fffa57ff4ab38..af3acabc3c1f5 100644 --- 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 @@ -2069,6 +2069,13 @@ public class ServiceConfiguration implements PulsarConfiguration { ) private String loadBalancerLoadSheddingStrategy = "org.apache.pulsar.broker.loadbalance.impl.ThresholdShedder"; + @FieldContext( + category = CATEGORY_LOAD_BALANCER, + doc = "When 【current usage < average usage - threshold】, " + + "the broker with the highest load will be triggered to unload" + ) + private boolean enableLowerBoundaryShedding = false; + @FieldContext( category = CATEGORY_LOAD_BALANCER, doc = "load balance placement strategy" diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/impl/RangeThresholdShedder.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/impl/RangeThresholdShedder.java deleted file mode 100644 index 3abb15f86ca27..0000000000000 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/impl/RangeThresholdShedder.java +++ /dev/null @@ -1,107 +0,0 @@ -/** - * Licensed to the Apache Software Foundation (ASF) under one - * or more contributor license agreements. See the NOTICE file - * distributed with this work for additional information - * regarding copyright ownership. The ASF licenses this file - * to you under the Apache License, Version 2.0 (the - * "License"); you may not use this file except in compliance - * with the License. You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, - * software distributed under the License is distributed on an - * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY - * KIND, either express or implied. See the License for the - * specific language governing permissions and limitations - * under the License. - */ -package org.apache.pulsar.broker.loadbalance.impl; - -import com.google.common.collect.Multimap; -import java.util.Map; -import lombok.extern.slf4j.Slf4j; -import org.apache.commons.lang3.tuple.Pair; -import org.apache.pulsar.broker.ServiceConfiguration; -import org.apache.pulsar.broker.loadbalance.LoadData; -import org.apache.pulsar.policies.data.loadbalancer.BrokerData; -import org.apache.pulsar.policies.data.loadbalancer.LocalBrokerData; - -/** - * On the basis of ThresholdShedder, RangeThresholdShedder adds the lower boundary judgment of the load. - * When 【current usage < average usage - threshold】, the broker with the highest load will be triggered to unload, - * avoiding the following scenarios: - * There are 11 Brokers, of which 10 are loaded at 80% and 1 is loaded at 0%. - * The average load is 80 * 10 / 11 = 72.73, and the threshold to unload is 72.73 + 10 = 82.73. - * Since 80 < 82.73, unload will not be trigger, and there is one idle Broker with load of 0%. - */ -@Slf4j -public class RangeThresholdShedder extends ThresholdShedder { - - @Override - public Multimap findBundlesForUnloading(LoadData loadData, ServiceConfiguration conf) { - super.findBundlesForUnloading(loadData, conf); - // Return if the bundle to unload has already been selected. - if (!selectedBundlesCache.isEmpty()) { - return selectedBundlesCache; - } - // Select the broker with the most resource usage. - final double threshold = conf.getLoadBalancerBrokerThresholdShedderPercentage() / 100.0; - final double avgUsage = getBrokerAvgUsage(loadData, conf, super.canSampleLog()); - Pair result = getMaxUsageBroker(loadData, threshold, avgUsage); - boolean hasBrokerBelowLowerBound = result.getLeft(); - String maxUsageBroker = result.getRight(); - BrokerData brokerData = loadData.getBrokerData().get(maxUsageBroker); - if (brokerData == null || brokerData.getLocalData() == null - || brokerData.getLocalData().getBundles().size() <= 1) { - log.info("Load data is null or bundle <=1, broker name is {}, skipping bundle unload.", maxUsageBroker); - return selectedBundlesCache; - } - if (!hasBrokerBelowLowerBound) { - log.info("No broker is below the lower bound, threshold is {}, " - + "avgUsage usage is {}, max usage of Broker {} is {}", - threshold, avgUsage, maxUsageBroker, - brokerAvgResourceUsage.getOrDefault(maxUsageBroker, 0.0)); - return selectedBundlesCache; - } - LocalBrokerData localData = brokerData.getLocalData(); - double minimumThroughputToOffload = getMinimumThroughputToOffload(threshold, localData); - final double minThroughputThreshold = conf.getLoadBalancerBundleUnloadMinThroughputThreshold() * MB; - if (minThroughputThreshold > minimumThroughputToOffload) { - log.info("broker {} in RangeThresholdShedder is planning to shed throughput {} MByte/s less than " - + "minimumThroughputThreshold {} MByte/s, skipping bundle unload.", - maxUsageBroker, minimumThroughputToOffload / MB, minThroughputThreshold / MB); - return selectedBundlesCache; - } - super.filterAndSelectBundle(loadData, loadData.getRecentlyUnloadedBundles(), maxUsageBroker, localData, - minimumThroughputToOffload); - return selectedBundlesCache; - } - - private Pair getMaxUsageBroker( - LoadData loadData, double threshold, double avgUsage) { - String maxUsageBrokerName = ""; - double maxUsage = -1; - boolean hasBrokerBelowLowerBound = false; - for (Map.Entry entry : loadData.getBrokerData().entrySet()) { - String broker = entry.getKey(); - double currentUsage = brokerAvgResourceUsage.getOrDefault(broker, 0.0); - // Select the broker with the most resource usage. - if (currentUsage > maxUsage) { - maxUsage = currentUsage; - maxUsageBrokerName = broker; - } - // Whether any brokers with low usage in the cluster. - if (currentUsage < avgUsage - threshold) { - hasBrokerBelowLowerBound = true; - } - } - return Pair.of(hasBrokerBelowLowerBound, maxUsageBrokerName); - } - - private double getMinimumThroughputToOffload(double threshold, LocalBrokerData localData) { - double brokerCurrentThroughput = localData.getMsgThroughputIn() + localData.getMsgThroughputOut(); - return brokerCurrentThroughput * threshold; - } - -} diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/impl/ThresholdShedder.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/impl/ThresholdShedder.java index c1ab26d4483a7..3d7be7b5b6103 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/impl/ThresholdShedder.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/impl/ThresholdShedder.java @@ -50,12 +50,15 @@ */ public class ThresholdShedder implements LoadSheddingStrategy { private static final Logger log = LoggerFactory.getLogger(ThresholdShedder.class); - protected final Multimap selectedBundlesCache = ArrayListMultimap.create(); - public static final double ADDITIONAL_THRESHOLD_PERCENT_MARGIN = 0.05; - public static final double MB = 1024 * 1024; + private final Multimap selectedBundlesCache = ArrayListMultimap.create(); + private static final double ADDITIONAL_THRESHOLD_PERCENT_MARGIN = 0.05; + + private static final double LOWER_THRESHOLD_MARGIN = 0.5; + + private static final double MB = 1024 * 1024; private static final long LOAD_LOG_SAMPLE_DELAY_IN_SEC = 5 * 60; // 5 mins - protected final Map brokerAvgResourceUsage = new HashMap<>(); + private final Map brokerAvgResourceUsage = new HashMap<>(); private long lastSampledLoadLogTS = 0; @@ -63,7 +66,7 @@ private static int toPercentage(double usage) { return (int) (usage * 100); } - protected boolean canSampleLog() { + private boolean canSampleLog() { long now = System.currentTimeMillis() / 1000; boolean sampleLog = now - lastSampledLoadLogTS >= LOAD_LOG_SAMPLE_DELAY_IN_SEC; if (sampleLog) { @@ -132,7 +135,9 @@ public Multimap findBundlesForUnloading(final LoadData loadData, log.warn("Broker {} is overloaded despite having no bundles", broker); } }); - + if (selectedBundlesCache.isEmpty() && conf.isEnableLowerBoundaryShedding()) { + tryLowerBoundaryShedding(loadData, conf); + } return selectedBundlesCache; } @@ -231,4 +236,59 @@ private double updateAvgResourceUsage(String broker, LocalBrokerData localBroker return historyUsage; } + private void tryLowerBoundaryShedding(LoadData loadData, ServiceConfiguration conf) { + // Select the broker with the most resource usage. + final double threshold = conf.getLoadBalancerBrokerThresholdShedderPercentage() / 100.0; + final double avgUsage = getBrokerAvgUsage(loadData, conf, canSampleLog()); + Pair result = getMaxUsageBroker(loadData, threshold, avgUsage); + boolean hasBrokerBelowLowerBound = result.getLeft(); + String maxUsageBroker = result.getRight(); + BrokerData brokerData = loadData.getBrokerData().get(maxUsageBroker); + if (brokerData == null || brokerData.getLocalData() == null + || brokerData.getLocalData().getBundles().size() <= 1) { + log.info("Load data is null or bundle <=1, broker name is {}, skipping bundle unload.", maxUsageBroker); + return; + } + if (!hasBrokerBelowLowerBound) { + log.info("No broker is below the lower bound, threshold is {}, " + + "avgUsage usage is {}, max usage of Broker {} is {}", + threshold, avgUsage, maxUsageBroker, + brokerAvgResourceUsage.getOrDefault(maxUsageBroker, 0.0)); + return; + } + LocalBrokerData localData = brokerData.getLocalData(); + double brokerCurrentThroughput = localData.getMsgThroughputIn() + localData.getMsgThroughputOut(); + double minimumThroughputToOffload = brokerCurrentThroughput * threshold * LOWER_THRESHOLD_MARGIN; + double minThroughputThreshold = conf.getLoadBalancerBundleUnloadMinThroughputThreshold() * MB; + if (minThroughputThreshold > minimumThroughputToOffload) { + log.info("broker {} in RangeThresholdShedder is planning to shed throughput {} MByte/s less than " + + "minimumThroughputThreshold {} MByte/s, skipping bundle unload.", + maxUsageBroker, minimumThroughputToOffload / MB, minThroughputThreshold / MB); + return; + } + filterAndSelectBundle(loadData, loadData.getRecentlyUnloadedBundles(), maxUsageBroker, localData, + minimumThroughputToOffload); + } + + private Pair getMaxUsageBroker( + LoadData loadData, double threshold, double avgUsage) { + String maxUsageBrokerName = ""; + double maxUsage = -1; + boolean hasBrokerBelowLowerBound = false; + for (Map.Entry entry : loadData.getBrokerData().entrySet()) { + String broker = entry.getKey(); + double currentUsage = brokerAvgResourceUsage.getOrDefault(broker, 0.0); + // Select the broker with the most resource usage. + if (currentUsage > maxUsage) { + maxUsage = currentUsage; + maxUsageBrokerName = broker; + } + // Whether any brokers with low usage in the cluster. + if (currentUsage < avgUsage - threshold) { + hasBrokerBelowLowerBound = true; + } + } + return Pair.of(hasBrokerBelowLowerBound, maxUsageBrokerName); + } + } diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/RangeThresholdShedderTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/RangeThresholdShedderTest.java deleted file mode 100644 index dcea19417ad5e..0000000000000 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/RangeThresholdShedderTest.java +++ /dev/null @@ -1,116 +0,0 @@ -/** - * Licensed to the Apache Software Foundation (ASF) under one - * or more contributor license agreements. See the NOTICE file - * distributed with this work for additional information - * regarding copyright ownership. The ASF licenses this file - * to you under the Apache License, Version 2.0 (the - * "License"); you may not use this file except in compliance - * with the License. You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, - * software distributed under the License is distributed on an - * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY - * KIND, either express or implied. See the License for the - * specific language governing permissions and limitations - * under the License. - */ -package org.apache.pulsar.broker.loadbalance; - -import static org.testng.Assert.assertFalse; -import static org.testng.Assert.assertTrue; -import com.google.common.collect.Multimap; -import lombok.extern.slf4j.Slf4j; -import org.apache.pulsar.broker.ServiceConfiguration; -import org.apache.pulsar.broker.loadbalance.impl.RangeThresholdShedder; -import org.apache.pulsar.broker.loadbalance.impl.ThresholdShedder; -import org.apache.pulsar.broker.loadbalance.impl.ThresholdShedderTest; -import org.apache.pulsar.policies.data.loadbalancer.BrokerData; -import org.apache.pulsar.policies.data.loadbalancer.BundleData; -import org.apache.pulsar.policies.data.loadbalancer.LocalBrokerData; -import org.apache.pulsar.policies.data.loadbalancer.ResourceUsage; -import org.apache.pulsar.policies.data.loadbalancer.TimeAverageMessageData; -import org.testng.annotations.BeforeMethod; -import org.testng.annotations.Test; - -@Test(groups = "broker") -@Slf4j -public class RangeThresholdShedderTest extends ThresholdShedderTest { - private final ServiceConfiguration conf = new ServiceConfiguration(); - - @BeforeMethod - public void setup() { - super.thresholdShedder = new RangeThresholdShedder(); - } - - @Test - public void testRangeThroughput() { - int numBundles = 10; - int brokerNum = 11; - int lowLoadNode = 10; - LoadData loadData = new LoadData(); - double throughput = 50 * ThresholdShedder.MB; - //There are 11 Brokers, of which 10 are loaded at 80% and 1 is loaded at 0%. - //At this time, the average load is 80*10/11 = 72.73, and the threshold for rebalancing is 72.73 + 10 = 82.73. - //Since 80 < 82.73, rebalancing will not be trigger, and there is one Broker with load of 0. - for (int i = 0; i < brokerNum; i++) { - LocalBrokerData broker = new LocalBrokerData(); - for (int j = 0; j < numBundles; j++) { - broker.getBundles().add("bundle-" + j); - BundleData bundle = new BundleData(); - TimeAverageMessageData timeAverageMessageData = new TimeAverageMessageData(); - timeAverageMessageData.setMsgThroughputIn(i == lowLoadNode ? 0 : throughput); - timeAverageMessageData.setMsgThroughputOut(i == lowLoadNode ? 0 : throughput); - bundle.setShortTermData(timeAverageMessageData); - String broker2BundleName = "broker-" + i + "-bundle-" + (numBundles + i); - loadData.getBundleData().put(broker2BundleName, bundle); - broker.getBundles().add(broker2BundleName); - } - broker.setBandwidthIn(new ResourceUsage(i == lowLoadNode ? 0 : 80, 100)); - broker.setBandwidthOut(new ResourceUsage(i == lowLoadNode ? 0 : 80, 100)); - broker.setMsgThroughputIn(i == lowLoadNode ? 0 : throughput); - broker.setMsgThroughputOut(i == lowLoadNode ? 0 : throughput); - loadData.getBrokerData().put("broker-" + i, new BrokerData(broker)); - } - ThresholdShedder shedder = new ThresholdShedder(); - Multimap bundlesToUnload = shedder.findBundlesForUnloading(loadData, conf); - assertTrue(bundlesToUnload.isEmpty()); - bundlesToUnload = thresholdShedder.findBundlesForUnloading(loadData, conf); - assertFalse(bundlesToUnload.isEmpty()); - } - - @Test - public void testNoBrokerToOffload() { - int numBundles = 10; - int brokerNum = 11; - LoadData loadData = new LoadData(); - double throughput = 80 * ThresholdShedder.MB; - //Load of all Brokers are 80%, and no Broker needs to offload. - for (int i = 0; i < brokerNum; i++) { - LocalBrokerData broker = new LocalBrokerData(); - for (int j = 0; j < numBundles; j++) { - broker.getBundles().add("bundle-" + j); - BundleData bundle = new BundleData(); - TimeAverageMessageData timeAverageMessageData = new TimeAverageMessageData(); - timeAverageMessageData.setMsgThroughputIn(throughput); - timeAverageMessageData.setMsgThroughputOut(throughput); - bundle.setShortTermData(timeAverageMessageData); - String broker2BundleName = "broker-" + i + "-bundle-" + (numBundles + i); - loadData.getBundleData().put(broker2BundleName, bundle); - broker.getBundles().add(broker2BundleName); - } - broker.setBandwidthIn(new ResourceUsage(80, 100)); - broker.setBandwidthOut(new ResourceUsage(80, 100)); - broker.setMsgThroughputIn(throughput); - broker.setMsgThroughputOut(throughput); - loadData.getBrokerData().put("broker-" + i, new BrokerData(broker)); - } - ThresholdShedder shedder = new ThresholdShedder(); - Multimap bundlesToUnload = shedder.findBundlesForUnloading(loadData, conf); - assertTrue(bundlesToUnload.isEmpty()); - bundlesToUnload = thresholdShedder.findBundlesForUnloading(loadData, conf); - assertTrue(bundlesToUnload.isEmpty()); - } - -} diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/impl/ThresholdShedderTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/impl/ThresholdShedderTest.java index b5ff6bf94418d..605722e3d563b 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/impl/ThresholdShedderTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/impl/ThresholdShedderTest.java @@ -38,7 +38,7 @@ @Test(groups = "broker") @Slf4j public class ThresholdShedderTest { - protected ThresholdShedder thresholdShedder; + private ThresholdShedder thresholdShedder; private final ServiceConfiguration conf; public ThresholdShedderTest() { @@ -47,6 +47,7 @@ public ThresholdShedderTest() { @BeforeMethod public void setup() { + conf.setEnableLowerBoundaryShedding(false); thresholdShedder = new ThresholdShedder(); } @@ -224,4 +225,75 @@ public void testPrintResourceUsage() { assertEquals(data.printResourceUsage(), "cpu: 10.00%, memory: 50.00%, directMemory: 90.00%, bandwidthIn: 30.00%, bandwidthOut: 20.00%"); } + + @Test + public void testRangeThroughput() { + int numBundles = 10; + int brokerNum = 11; + int lowLoadNode = 10; + LoadData loadData = new LoadData(); + double throughput = 100 * 1024 * 1024; + //There are 11 Brokers, of which 10 are loaded at 80% and 1 is loaded at 0%. + //At this time, the average load is 80*10/11 = 72.73, and the threshold for rebalancing is 72.73 + 10 = 82.73. + //Since 80 < 82.73, rebalancing will not be trigger, and there is one Broker with load of 0. + for (int i = 0; i < brokerNum; i++) { + LocalBrokerData broker = new LocalBrokerData(); + for (int j = 0; j < numBundles; j++) { + broker.getBundles().add("bundle-" + j); + BundleData bundle = new BundleData(); + TimeAverageMessageData timeAverageMessageData = new TimeAverageMessageData(); + timeAverageMessageData.setMsgThroughputIn(i == lowLoadNode ? 0 : throughput); + timeAverageMessageData.setMsgThroughputOut(i == lowLoadNode ? 0 : throughput); + bundle.setShortTermData(timeAverageMessageData); + String broker2BundleName = "broker-" + i + "-bundle-" + (numBundles + i); + loadData.getBundleData().put(broker2BundleName, bundle); + broker.getBundles().add(broker2BundleName); + } + broker.setBandwidthIn(new ResourceUsage(i == lowLoadNode ? 0 : 80, 100)); + broker.setBandwidthOut(new ResourceUsage(i == lowLoadNode ? 0 : 80, 100)); + broker.setMsgThroughputIn(i == lowLoadNode ? 0 : throughput); + broker.setMsgThroughputOut(i == lowLoadNode ? 0 : throughput); + loadData.getBrokerData().put("broker-" + i, new BrokerData(broker)); + } + ThresholdShedder shedder = new ThresholdShedder(); + Multimap bundlesToUnload = shedder.findBundlesForUnloading(loadData, conf); + assertTrue(bundlesToUnload.isEmpty()); + conf.setEnableLowerBoundaryShedding(true); + bundlesToUnload = thresholdShedder.findBundlesForUnloading(loadData, conf); + assertFalse(bundlesToUnload.isEmpty()); + } + + @Test + public void testNoBrokerToOffload() { + int numBundles = 10; + int brokerNum = 11; + LoadData loadData = new LoadData(); + double throughput = 80 * 1024 * 1024; + //Load of all Brokers are 80%, and no Broker needs to offload. + for (int i = 0; i < brokerNum; i++) { + LocalBrokerData broker = new LocalBrokerData(); + for (int j = 0; j < numBundles; j++) { + broker.getBundles().add("bundle-" + j); + BundleData bundle = new BundleData(); + TimeAverageMessageData timeAverageMessageData = new TimeAverageMessageData(); + timeAverageMessageData.setMsgThroughputIn(throughput); + timeAverageMessageData.setMsgThroughputOut(throughput); + bundle.setShortTermData(timeAverageMessageData); + String broker2BundleName = "broker-" + i + "-bundle-" + (numBundles + i); + loadData.getBundleData().put(broker2BundleName, bundle); + broker.getBundles().add(broker2BundleName); + } + broker.setBandwidthIn(new ResourceUsage(80, 100)); + broker.setBandwidthOut(new ResourceUsage(80, 100)); + broker.setMsgThroughputIn(throughput); + broker.setMsgThroughputOut(throughput); + loadData.getBrokerData().put("broker-" + i, new BrokerData(broker)); + } + ThresholdShedder shedder = new ThresholdShedder(); + Multimap bundlesToUnload = shedder.findBundlesForUnloading(loadData, conf); + assertTrue(bundlesToUnload.isEmpty()); + conf.setEnableLowerBoundaryShedding(true); + bundlesToUnload = thresholdShedder.findBundlesForUnloading(loadData, conf); + assertTrue(bundlesToUnload.isEmpty()); + } } From b3febe92def1dec30f86ad08eed16b1778677917 Mon Sep 17 00:00:00 2001 From: feynmanlin <315157973@qq.com> Date: Mon, 5 Sep 2022 22:56:40 +0800 Subject: [PATCH 04/11] Restore protect to private --- .../pulsar/broker/loadbalance/impl/ThresholdShedder.java | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/impl/ThresholdShedder.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/impl/ThresholdShedder.java index 3d7be7b5b6103..a0c23a8f79ac5 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/impl/ThresholdShedder.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/impl/ThresholdShedder.java @@ -53,7 +53,7 @@ public class ThresholdShedder implements LoadSheddingStrategy { private final Multimap selectedBundlesCache = ArrayListMultimap.create(); private static final double ADDITIONAL_THRESHOLD_PERCENT_MARGIN = 0.05; - private static final double LOWER_THRESHOLD_MARGIN = 0.5; + private static final double LOWER_BOUNDARY_THRESHOLD_MARGIN = 0.5; private static final double MB = 1024 * 1024; @@ -141,7 +141,7 @@ public Multimap findBundlesForUnloading(final LoadData loadData, return selectedBundlesCache; } - protected void filterAndSelectBundle(LoadData loadData, Map recentlyUnloadedBundles, String broker, + private void filterAndSelectBundle(LoadData loadData, Map recentlyUnloadedBundles, String broker, LocalBrokerData localData, double minimumThroughputToOffload) { MutableDouble trafficMarkedToOffload = new MutableDouble(0); MutableBoolean atLeastOneBundleSelected = new MutableBoolean(false); @@ -168,7 +168,7 @@ protected void filterAndSelectBundle(LoadData loadData, Map recent }); } - protected double getBrokerAvgUsage(final LoadData loadData, + private double getBrokerAvgUsage(final LoadData loadData, final ServiceConfiguration conf, boolean sampleLog) { double historyPercentage = conf.getLoadBalancerHistoryResourcePercentage(); double totalUsage = 0.0; @@ -258,7 +258,7 @@ private void tryLowerBoundaryShedding(LoadData loadData, ServiceConfiguration co } LocalBrokerData localData = brokerData.getLocalData(); double brokerCurrentThroughput = localData.getMsgThroughputIn() + localData.getMsgThroughputOut(); - double minimumThroughputToOffload = brokerCurrentThroughput * threshold * LOWER_THRESHOLD_MARGIN; + double minimumThroughputToOffload = brokerCurrentThroughput * threshold * LOWER_BOUNDARY_THRESHOLD_MARGIN; double minThroughputThreshold = conf.getLoadBalancerBundleUnloadMinThroughputThreshold() * MB; if (minThroughputThreshold > minimumThroughputToOffload) { log.info("broker {} in RangeThresholdShedder is planning to shed throughput {} MByte/s less than " From 2371def91a52c506f14dd8c3dadf74877f7a1106 Mon Sep 17 00:00:00 2001 From: feynmanlin <315157973@qq.com> Date: Wed, 7 Sep 2022 23:25:53 +0800 Subject: [PATCH 05/11] Update pulsar-broker-common/src/main/java/org/apache/pulsar/broker/ServiceConfiguration.java Co-authored-by: Michael Marshall --- .../java/org/apache/pulsar/broker/ServiceConfiguration.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) 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 index e9dd565800a6c..d9a40342cb55a 100644 --- 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 @@ -2071,7 +2071,7 @@ public class ServiceConfiguration implements PulsarConfiguration { @FieldContext( category = CATEGORY_LOAD_BALANCER, - doc = "When 【current usage < average usage - threshold】, " + doc = "When [current usage < average usage - threshold], " + "the broker with the highest load will be triggered to unload" ) private boolean enableLowerBoundaryShedding = false; From 1f5d4e4e14a168fc7c95a6978266e50548a65f85 Mon Sep 17 00:00:00 2001 From: feynmanlin <315157973@qq.com> Date: Wed, 14 Sep 2022 00:24:24 +0800 Subject: [PATCH 06/11] Address comment --- .../pulsar/broker/ServiceConfiguration.java | 2 +- .../broker/loadbalance/impl/ThresholdShedder.java | 15 ++++++++------- .../loadbalance/impl/ThresholdShedderTest.java | 6 +++--- 3 files changed, 12 insertions(+), 11 deletions(-) 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 index d9a40342cb55a..512e456dcf3f1 100644 --- 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 @@ -2074,7 +2074,7 @@ public class ServiceConfiguration implements PulsarConfiguration { doc = "When [current usage < average usage - threshold], " + "the broker with the highest load will be triggered to unload" ) - private boolean enableLowerBoundaryShedding = false; + private boolean lowerBoundarySheddingEnabled = false; @FieldContext( category = CATEGORY_LOAD_BALANCER, diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/impl/ThresholdShedder.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/impl/ThresholdShedder.java index a0c23a8f79ac5..3ffc4978f050d 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/impl/ThresholdShedder.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/impl/ThresholdShedder.java @@ -135,7 +135,7 @@ public Multimap findBundlesForUnloading(final LoadData loadData, log.warn("Broker {} is overloaded despite having no bundles", broker); } }); - if (selectedBundlesCache.isEmpty() && conf.isEnableLowerBoundaryShedding()) { + if (selectedBundlesCache.isEmpty() && conf.isLowerBoundarySheddingEnabled()) { tryLowerBoundaryShedding(loadData, conf); } return selectedBundlesCache; @@ -244,9 +244,8 @@ private void tryLowerBoundaryShedding(LoadData loadData, ServiceConfiguration co boolean hasBrokerBelowLowerBound = result.getLeft(); String maxUsageBroker = result.getRight(); BrokerData brokerData = loadData.getBrokerData().get(maxUsageBroker); - if (brokerData == null || brokerData.getLocalData() == null - || brokerData.getLocalData().getBundles().size() <= 1) { - log.info("Load data is null or bundle <=1, broker name is {}, skipping bundle unload.", maxUsageBroker); + if (brokerData == null) { + log.info("Load data is null or bundle <=1, skipping bundle unload."); return; } if (!hasBrokerBelowLowerBound) { @@ -261,7 +260,7 @@ private void tryLowerBoundaryShedding(LoadData loadData, ServiceConfiguration co double minimumThroughputToOffload = brokerCurrentThroughput * threshold * LOWER_BOUNDARY_THRESHOLD_MARGIN; double minThroughputThreshold = conf.getLoadBalancerBundleUnloadMinThroughputThreshold() * MB; if (minThroughputThreshold > minimumThroughputToOffload) { - log.info("broker {} in RangeThresholdShedder is planning to shed throughput {} MByte/s less than " + log.info("broker {} in lower boundary shedding is planning to shed throughput {} MByte/s less than " + "minimumThroughputThreshold {} MByte/s, skipping bundle unload.", maxUsageBroker, minimumThroughputToOffload / MB, minThroughputThreshold / MB); return; @@ -273,13 +272,15 @@ private void tryLowerBoundaryShedding(LoadData loadData, ServiceConfiguration co private Pair getMaxUsageBroker( LoadData loadData, double threshold, double avgUsage) { String maxUsageBrokerName = ""; - double maxUsage = -1; + double maxUsage = avgUsage + threshold; boolean hasBrokerBelowLowerBound = false; for (Map.Entry entry : loadData.getBrokerData().entrySet()) { String broker = entry.getKey(); + BrokerData brokerData = entry.getValue(); double currentUsage = brokerAvgResourceUsage.getOrDefault(broker, 0.0); // Select the broker with the most resource usage. - if (currentUsage > maxUsage) { + if (currentUsage > maxUsage && brokerData.getLocalData() != null + && brokerData.getLocalData().getBundles().size() > 1) { maxUsage = currentUsage; maxUsageBrokerName = broker; } diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/impl/ThresholdShedderTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/impl/ThresholdShedderTest.java index 605722e3d563b..e58cc9ea6c104 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/impl/ThresholdShedderTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/impl/ThresholdShedderTest.java @@ -47,7 +47,7 @@ public ThresholdShedderTest() { @BeforeMethod public void setup() { - conf.setEnableLowerBoundaryShedding(false); + conf.setLowerBoundarySheddingEnabled(false); thresholdShedder = new ThresholdShedder(); } @@ -258,7 +258,7 @@ public void testRangeThroughput() { ThresholdShedder shedder = new ThresholdShedder(); Multimap bundlesToUnload = shedder.findBundlesForUnloading(loadData, conf); assertTrue(bundlesToUnload.isEmpty()); - conf.setEnableLowerBoundaryShedding(true); + conf.setLowerBoundarySheddingEnabled(true); bundlesToUnload = thresholdShedder.findBundlesForUnloading(loadData, conf); assertFalse(bundlesToUnload.isEmpty()); } @@ -292,7 +292,7 @@ public void testNoBrokerToOffload() { ThresholdShedder shedder = new ThresholdShedder(); Multimap bundlesToUnload = shedder.findBundlesForUnloading(loadData, conf); assertTrue(bundlesToUnload.isEmpty()); - conf.setEnableLowerBoundaryShedding(true); + conf.setLowerBoundarySheddingEnabled(true); bundlesToUnload = thresholdShedder.findBundlesForUnloading(loadData, conf); assertTrue(bundlesToUnload.isEmpty()); } From 1ecb7d623c1352e351392da07f07f8693021cc25 Mon Sep 17 00:00:00 2001 From: feynmanlin <315157973@qq.com> Date: Wed, 14 Sep 2022 01:13:41 +0800 Subject: [PATCH 07/11] Add unit test --- .../loadbalance/impl/ThresholdShedder.java | 2 +- .../impl/ThresholdShedderTest.java | 43 ++++++++++++++++++- 2 files changed, 42 insertions(+), 3 deletions(-) diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/impl/ThresholdShedder.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/impl/ThresholdShedder.java index 3ffc4978f050d..3f4452bb73569 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/impl/ThresholdShedder.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/impl/ThresholdShedder.java @@ -272,7 +272,7 @@ private void tryLowerBoundaryShedding(LoadData loadData, ServiceConfiguration co private Pair getMaxUsageBroker( LoadData loadData, double threshold, double avgUsage) { String maxUsageBrokerName = ""; - double maxUsage = avgUsage + threshold; + double maxUsage = -1; boolean hasBrokerBelowLowerBound = false; for (Map.Entry entry : loadData.getBrokerData().entrySet()) { String broker = entry.getKey(); diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/impl/ThresholdShedderTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/impl/ThresholdShedderTest.java index e58cc9ea6c104..8ca9c57084be0 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/impl/ThresholdShedderTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/impl/ThresholdShedderTest.java @@ -245,7 +245,7 @@ public void testRangeThroughput() { timeAverageMessageData.setMsgThroughputIn(i == lowLoadNode ? 0 : throughput); timeAverageMessageData.setMsgThroughputOut(i == lowLoadNode ? 0 : throughput); bundle.setShortTermData(timeAverageMessageData); - String broker2BundleName = "broker-" + i + "-bundle-" + (numBundles + i); + String broker2BundleName = "broker-" + i + "-bundle-" + j; loadData.getBundleData().put(broker2BundleName, bundle); broker.getBundles().add(broker2BundleName); } @@ -279,7 +279,7 @@ public void testNoBrokerToOffload() { timeAverageMessageData.setMsgThroughputIn(throughput); timeAverageMessageData.setMsgThroughputOut(throughput); bundle.setShortTermData(timeAverageMessageData); - String broker2BundleName = "broker-" + i + "-bundle-" + (numBundles + i); + String broker2BundleName = "broker-" + i + "-bundle-" + j; loadData.getBundleData().put(broker2BundleName, bundle); broker.getBundles().add(broker2BundleName); } @@ -296,4 +296,43 @@ public void testNoBrokerToOffload() { bundlesToUnload = thresholdShedder.findBundlesForUnloading(loadData, conf); assertTrue(bundlesToUnload.isEmpty()); } + + @Test + public void testBrokerWithOneBundle() { + int brokerNum = 11; + int lowLoadNode = 5; + int brokerWithManyBundles = 3; + LoadData loadData = new LoadData(); + double throughput = 100 * 1024 * 1024; + //There are 11 Brokers, of which 10 are loaded at 80% and 1 is loaded at 0%. + //Only broker3 has 10 bundles. + for (int i = 0; i < brokerNum; i++) { + LocalBrokerData broker = new LocalBrokerData(); + //Broker3 has 10 bundles + int numBundles = i == brokerWithManyBundles ? 10 : 1; + for (int j = 0; j < numBundles; j++) { + BundleData bundle = new BundleData(); + TimeAverageMessageData timeAverageMessageData = new TimeAverageMessageData(); + timeAverageMessageData.setMsgThroughputIn(i == lowLoadNode ? 0 : throughput); + timeAverageMessageData.setMsgThroughputOut(i == lowLoadNode ? 0 : throughput); + bundle.setShortTermData(timeAverageMessageData); + String broker2BundleName = "broker-" + i + "-bundle-" + j; + loadData.getBundleData().put(broker2BundleName, bundle); + broker.getBundles().add(broker2BundleName); + } + broker.setBandwidthIn(new ResourceUsage(i == lowLoadNode ? 0 : 80, 100)); + broker.setBandwidthOut(new ResourceUsage(i == lowLoadNode ? 0 : 80, 100)); + broker.setMsgThroughputIn(i == lowLoadNode ? 0 : throughput); + broker.setMsgThroughputOut(i == lowLoadNode ? 0 : throughput); + loadData.getBrokerData().put("broker-" + i, new BrokerData(broker)); + } + ThresholdShedder shedder = new ThresholdShedder(); + Multimap bundlesToUnload = shedder.findBundlesForUnloading(loadData, conf); + assertTrue(bundlesToUnload.isEmpty()); + conf.setLowerBoundarySheddingEnabled(true); + bundlesToUnload = thresholdShedder.findBundlesForUnloading(loadData, conf); + assertFalse(bundlesToUnload.isEmpty()); + assertEquals(bundlesToUnload.size(), 1); + assertTrue(bundlesToUnload.containsKey("broker-3")); + } } From 633a27c8dfaffc207965eb021c4114d3461c3244 Mon Sep 17 00:00:00 2001 From: feynmanlin <315157973@qq.com> Date: Wed, 14 Sep 2022 23:51:43 +0800 Subject: [PATCH 08/11] Set initial value --- .../apache/pulsar/broker/loadbalance/impl/ThresholdShedder.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/impl/ThresholdShedder.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/impl/ThresholdShedder.java index 3f4452bb73569..928f045369cff 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/impl/ThresholdShedder.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/impl/ThresholdShedder.java @@ -272,7 +272,7 @@ private void tryLowerBoundaryShedding(LoadData loadData, ServiceConfiguration co private Pair getMaxUsageBroker( LoadData loadData, double threshold, double avgUsage) { String maxUsageBrokerName = ""; - double maxUsage = -1; + double maxUsage = avgUsage - threshold; boolean hasBrokerBelowLowerBound = false; for (Map.Entry entry : loadData.getBrokerData().entrySet()) { String broker = entry.getKey(); From d3ce511b1b0adf18aabbbdc37e2e621f2cc45c13 Mon Sep 17 00:00:00 2001 From: feynmanlin <315157973@qq.com> Date: Fri, 16 Sep 2022 10:41:45 +0800 Subject: [PATCH 09/11] Update pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/impl/ThresholdShedderTest.java Co-authored-by: Penghui Li --- .../pulsar/broker/loadbalance/impl/ThresholdShedderTest.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/impl/ThresholdShedderTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/impl/ThresholdShedderTest.java index 8ca9c57084be0..f10b6faa27930 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/impl/ThresholdShedderTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/impl/ThresholdShedderTest.java @@ -264,7 +264,7 @@ public void testRangeThroughput() { } @Test - public void testNoBrokerToOffload() { + public void testLowerBoundarySheddingNoBrokerToOffload() { int numBundles = 10; int brokerNum = 11; LoadData loadData = new LoadData(); From 257e9983b00a11ea6707000116f22535b3712ba1 Mon Sep 17 00:00:00 2001 From: feynmanlin <315157973@qq.com> Date: Fri, 16 Sep 2022 10:41:55 +0800 Subject: [PATCH 10/11] Update pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/impl/ThresholdShedderTest.java Co-authored-by: Penghui Li --- .../pulsar/broker/loadbalance/impl/ThresholdShedderTest.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/impl/ThresholdShedderTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/impl/ThresholdShedderTest.java index f10b6faa27930..9c415fbe7d1d4 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/impl/ThresholdShedderTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/impl/ThresholdShedderTest.java @@ -227,7 +227,7 @@ public void testPrintResourceUsage() { } @Test - public void testRangeThroughput() { + public void testLowerBoundaryShedding() { int numBundles = 10; int brokerNum = 11; int lowLoadNode = 10; From cb2403dae80fc294f15642f8d27ccf998403b4c5 Mon Sep 17 00:00:00 2001 From: feynmanlin <315157973@qq.com> Date: Fri, 16 Sep 2022 10:42:10 +0800 Subject: [PATCH 11/11] Update pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/impl/ThresholdShedderTest.java Co-authored-by: Penghui Li --- .../pulsar/broker/loadbalance/impl/ThresholdShedderTest.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/impl/ThresholdShedderTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/impl/ThresholdShedderTest.java index 9c415fbe7d1d4..8461f8ce74c43 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/impl/ThresholdShedderTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/impl/ThresholdShedderTest.java @@ -298,7 +298,7 @@ public void testLowerBoundarySheddingNoBrokerToOffload() { } @Test - public void testBrokerWithOneBundle() { + public void testLowerBoundarySheddingBrokerWithOneBundle() { int brokerNum = 11; int lowLoadNode = 5; int brokerWithManyBundles = 3;