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 30ca29faad889..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 @@ -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 lowerBoundarySheddingEnabled = 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/ThresholdShedder.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/impl/ThresholdShedder.java index 05c69062663bb..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 @@ -52,6 +52,9 @@ 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 LOWER_BOUNDARY_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 @@ -80,8 +83,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, sampleLog); + final double avgUsage = getBrokerAvgUsage(loadData, conf, sampleLog); if (sampleLog) { log.info("brokers' resource avgUsage:{}%", toPercentage(avgUsage)); } @@ -122,17 +124,34 @@ 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); + } + }); + if (selectedBundlesCache.isEmpty() && conf.isLowerBoundarySheddingEnabled()) { + tryLowerBoundaryShedding(loadData, conf); + } + return selectedBundlesCache; + } + + private 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 -> @@ -147,21 +166,11 @@ 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; } - private double getBrokerAvgUsage(final LoadData loadData, final double historyPercentage, + private double getBrokerAvgUsage(final LoadData loadData, final ServiceConfiguration conf, boolean sampleLog) { + double historyPercentage = conf.getLoadBalancerHistoryResourcePercentage(); double totalUsage = 0.0; int totalBrokers = 0; @@ -227,4 +236,60 @@ 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) { + log.info("Load data is null or bundle <=1, skipping bundle unload."); + 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_BOUNDARY_THRESHOLD_MARGIN; + double minThroughputThreshold = conf.getLoadBalancerBundleUnloadMinThroughputThreshold() * MB; + if (minThroughputThreshold > minimumThroughputToOffload) { + 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; + } + filterAndSelectBundle(loadData, loadData.getRecentlyUnloadedBundles(), maxUsageBroker, localData, + minimumThroughputToOffload); + } + + private Pair getMaxUsageBroker( + LoadData loadData, double threshold, double avgUsage) { + String maxUsageBrokerName = ""; + 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 && brokerData.getLocalData() != null + && brokerData.getLocalData().getBundles().size() > 1) { + 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/impl/ThresholdShedderTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/impl/ThresholdShedderTest.java index 80760a7afc307..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 @@ -47,6 +47,7 @@ public ThresholdShedderTest() { @BeforeMethod public void setup() { + conf.setLowerBoundarySheddingEnabled(false); thresholdShedder = new ThresholdShedder(); } @@ -224,4 +225,114 @@ public void testPrintResourceUsage() { assertEquals(data.printResourceUsage(), "cpu: 10.00%, memory: 50.00%, directMemory: 90.00%, bandwidthIn: 30.00%, bandwidthOut: 20.00%"); } + + @Test + public void testLowerBoundaryShedding() { + 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-" + 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()); + } + + @Test + public void testLowerBoundarySheddingNoBrokerToOffload() { + 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-" + j; + 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.setLowerBoundarySheddingEnabled(true); + bundlesToUnload = thresholdShedder.findBundlesForUnloading(loadData, conf); + assertTrue(bundlesToUnload.isEmpty()); + } + + @Test + public void testLowerBoundarySheddingBrokerWithOneBundle() { + 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")); + } }