Skip to content

Commit

Permalink
[fix][broker] fix UniformLoadShedder seleet wrong overloadbroker and …
Browse files Browse the repository at this point in the history 
…underloadbroker (apache#21025)
  • Loading branch information
@aloyszhang
aloyszhang committed Sep 14, 2023
1 parent bcdcae2 commit c60ce86
Show file tree
Hide file tree
Showing 2 changed files with 178 additions and 63 deletions.
163 changes: 100 additions & 63 deletions ...ar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/impl/UniformLoadShedder.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -25,7 +25,7 @@
import org.apache.commons.lang3.mutable.MutableDouble;
import org.apache.commons.lang3.mutable.MutableInt;
import org.apache.commons.lang3.mutable.MutableObject;
import org.apache.commons.lang3.tuple.Triple;
import org.apache.commons.lang3.tuple.Pair;
import org.apache.pulsar.broker.ServiceConfiguration;
import org.apache.pulsar.broker.loadbalance.LoadData;
import org.apache.pulsar.broker.loadbalance.LoadSheddingStrategy;
Expand All @@ -36,7 +36,7 @@

/**
* This strategy tends to distribute load uniformly across all brokers. This strategy checks load difference between
* broker with highest load and broker with lowest load. If the difference is higher than configured thresholds
* broker with the highest load and broker with the lowest load. If the difference is higher than configured thresholds
* {@link ServiceConfiguration#getLoadBalancerMsgRateDifferenceShedderThreshold()} or
* {@link ServiceConfiguration#getLoadBalancerMsgThroughputMultiplierDifferenceShedderThreshold()} then it finds out
* bundles which can be unloaded to distribute traffic evenly across all brokers.
Expand All @@ -63,25 +63,37 @@ public Multimap<String, String> findBundlesForUnloading(final LoadData loadData,
Map<String, BundleData> loadBundleData = loadData.getBundleDataForLoadShedding();
Map<String, Long> recentlyUnloadedBundles = loadData.getRecentlyUnloadedBundles();

MutableObject<String> overloadedBroker = new MutableObject<>();
MutableObject<String> underloadedBroker = new MutableObject<>();
MutableObject<String> msgRateOverloadedBroker = new MutableObject<>();
MutableObject<String> msgThroughputOverloadedBroker = new MutableObject<>();
MutableObject<String> msgRateUnderloadedBroker = new MutableObject<>();
MutableObject<String> msgThroughputUnderloadedBroker = new MutableObject<>();
MutableDouble maxMsgRate = new MutableDouble(-1);
MutableDouble maxThroughputRate = new MutableDouble(-1);
MutableDouble maxThroughput = new MutableDouble(-1);
MutableDouble minMsgRate = new MutableDouble(Integer.MAX_VALUE);
MutableDouble minThroughputRate = new MutableDouble(Integer.MAX_VALUE);
MutableDouble minThroughput = new MutableDouble(Integer.MAX_VALUE);

brokersData.forEach((broker, data) -> {
double msgRate = data.getLocalData().getMsgRateIn() + data.getLocalData().getMsgRateOut();
double throughputRate = data.getLocalData().getMsgThroughputIn()
+ data.getLocalData().getMsgThroughputOut();
if (msgRate > maxMsgRate.getValue() || throughputRate > maxThroughputRate.getValue()) {
overloadedBroker.setValue(broker);
if (msgRate > maxMsgRate.getValue()) {
msgRateOverloadedBroker.setValue(broker);
maxMsgRate.setValue(msgRate);
maxThroughputRate.setValue(throughputRate);
}
if (msgRate < minMsgRate.getValue() || throughputRate < minThroughputRate.getValue()) {
underloadedBroker.setValue(broker);

if (throughputRate > maxThroughput.getValue()) {
msgThroughputOverloadedBroker.setValue(broker);
maxThroughput.setValue(throughputRate);
}

if (msgRate < minMsgRate.getValue()) {
msgRateUnderloadedBroker.setValue(broker);
minMsgRate.setValue(msgRate);
minThroughputRate.setValue(throughputRate);
}

if (throughputRate < minThroughput.getValue()) {
msgThroughputUnderloadedBroker.setValue(broker);
minThroughput.setValue(throughputRate);
}
});

Expand All @@ -91,12 +103,12 @@ public Multimap<String, String> findBundlesForUnloading(final LoadData loadData,
if (minMsgRate.getValue() <= EPS && minMsgRate.getValue() >= -EPS) {
minMsgRate.setValue(1.0);
}
if (minThroughputRate.getValue() <= EPS && minThroughputRate.getValue() >= -EPS) {
minThroughputRate.setValue(1.0);
if (minThroughput.getValue() <= EPS && minThroughput.getValue() >= -EPS) {
minThroughput.setValue(1.0);
}
double msgRateDifferencePercentage = ((maxMsgRate.getValue() - minMsgRate.getValue()) * 100)
/ (minMsgRate.getValue());
double msgThroughputDifferenceRate = maxThroughputRate.getValue() / minThroughputRate.getValue();
double msgThroughputDifferenceRate = maxThroughput.getValue() / minThroughput.getValue();

// if the threshold matches then find out how much load needs to be unloaded by considering number of msgRate
// and throughput.
Expand All @@ -105,66 +117,91 @@ public Multimap<String, String> findBundlesForUnloading(final LoadData loadData,
boolean isMsgThroughputThresholdExceeded = conf
.getLoadBalancerMsgThroughputMultiplierDifferenceShedderThreshold() > 0
&& msgThroughputDifferenceRate > conf
.getLoadBalancerMsgThroughputMultiplierDifferenceShedderThreshold();
.getLoadBalancerMsgThroughputMultiplierDifferenceShedderThreshold();

if (isMsgRateThresholdExceeded || isMsgThroughputThresholdExceeded) {
if (log.isDebugEnabled()) {
log.debug(
"Found bundles for uniform load balancing. "
+ "overloaded broker {} with (msgRate,throughput)= ({},{}) "
+ "and underloaded broker {} with (msgRate,throughput)= ({},{})",
overloadedBroker.getValue(), maxMsgRate.getValue(), maxThroughputRate.getValue(),
underloadedBroker.getValue(), minMsgRate.getValue(), minThroughputRate.getValue());
}
MutableInt msgRateRequiredFromUnloadedBundles = new MutableInt(
(int) ((maxMsgRate.getValue() - minMsgRate.getValue()) * conf.getMaxUnloadPercentage()));
MutableInt msgThroughputRequiredFromUnloadedBundles = new MutableInt(
(int) ((maxThroughputRate.getValue() - minThroughputRate.getValue())
(int) ((maxThroughput.getValue() - minThroughput.getValue())
* conf.getMaxUnloadPercentage()));
LocalBrokerData overloadedBrokerData = brokersData.get(overloadedBroker.getValue()).getLocalData();

if (overloadedBrokerData.getBundles().size() > 1
&& (msgRateRequiredFromUnloadedBundles.getValue() >= conf.getMinUnloadMessage()
|| msgThroughputRequiredFromUnloadedBundles.getValue() >= conf.getMinUnloadMessageThroughput())) {
// Sort bundles by throughput, then pick the bundle which can help to reduce load uniformly with
// under-loaded broker
loadBundleData.entrySet().stream()
.filter(e -> overloadedBrokerData.getBundles().contains(e.getKey()))
.map((e) -> {
String bundle = e.getKey();
BundleData bundleData = e.getValue();
TimeAverageMessageData shortTermData = bundleData.getShortTermData();
double throughput = isMsgRateThresholdExceeded
? shortTermData.getMsgRateIn() + shortTermData.getMsgRateOut()
: shortTermData.getMsgThroughputIn() + shortTermData.getMsgThroughputOut();
return Triple.of(bundle, bundleData, throughput);
}).filter(e -> !recentlyUnloadedBundles.containsKey(e.getLeft()))
.sorted((e1, e2) -> Double.compare(e2.getRight(), e1.getRight())).forEach((e) -> {
if (conf.getMaxUnloadBundleNumPerShedding() != -1
&& selectedBundlesCache.size() >= conf.getMaxUnloadBundleNumPerShedding()) {
return;
}
String bundle = e.getLeft();
BundleData bundleData = e.getMiddle();
TimeAverageMessageData shortTermData = bundleData.getShortTermData();
double throughput = shortTermData.getMsgThroughputIn()
+ shortTermData.getMsgThroughputOut();
double bundleMsgRate = shortTermData.getMsgRateIn() + shortTermData.getMsgRateOut();
if (isMsgRateThresholdExceeded) {
if (isMsgRateThresholdExceeded) {
if (log.isDebugEnabled()) {
log.debug("Found bundles for uniform load balancing. "
+ "msgRate overloaded broker: {} with msgRate: {}, "
+ "msgRate underloaded broker: {} with msgRate: {}",
msgRateOverloadedBroker.getValue(), maxMsgRate.getValue(),
msgRateUnderloadedBroker.getValue(), minMsgRate.getValue());
}
LocalBrokerData overloadedBrokerData =
brokersData.get(msgRateOverloadedBroker.getValue()).getLocalData();
if (overloadedBrokerData.getBundles().size() > 1
&& (msgRateRequiredFromUnloadedBundles.getValue() >= conf.getMinUnloadMessage())) {
// Sort bundles by throughput, then pick the bundle which can help to reduce load uniformly with
// under-loaded broker
loadBundleData.entrySet().stream()
.filter(e -> overloadedBrokerData.getBundles().contains(e.getKey()))
.map((e) -> {
String bundle = e.getKey();
TimeAverageMessageData shortTermData = e.getValue().getShortTermData();
double msgRate = shortTermData.getMsgRateIn() + shortTermData.getMsgRateOut();
return Pair.of(bundle, msgRate);
}).filter(e -> !recentlyUnloadedBundles.containsKey(e.getLeft()))
.sorted((e1, e2) -> Double.compare(e2.getRight(), e1.getRight())).forEach((e) -> {
if (conf.getMaxUnloadBundleNumPerShedding() != -1
&& selectedBundlesCache.size() >= conf.getMaxUnloadBundleNumPerShedding()) {
return;
}
String bundle = e.getLeft();
double bundleMsgRate = e.getRight();
if (bundleMsgRate <= (msgRateRequiredFromUnloadedBundles.getValue()
+ 1000/* delta */)) {
log.info("Found bundle to unload with msgRate {}", bundleMsgRate);
msgRateRequiredFromUnloadedBundles.add(-bundleMsgRate);
selectedBundlesCache.put(overloadedBroker.getValue(), bundle);
selectedBundlesCache.put(msgRateOverloadedBroker.getValue(), bundle);
}
} else {
if (throughput <= (msgThroughputRequiredFromUnloadedBundles.getValue())) {
log.info("Found bundle to unload with throughput {}", throughput);
msgThroughputRequiredFromUnloadedBundles.add(-throughput);
selectedBundlesCache.put(overloadedBroker.getValue(), bundle);
});
}
} else {
if (log.isDebugEnabled()) {
log.debug("Found bundles for uniform load balancing. "
+ "msgThroughput overloaded broker: {} with msgThroughput {}, "
+ "msgThroughput underloaded broker: {} with msgThroughput: {}",
msgThroughputOverloadedBroker.getValue(), maxThroughput.getValue(),
msgThroughputUnderloadedBroker.getValue(), minThroughput.getValue());
}
LocalBrokerData overloadedBrokerData =
brokersData.get(msgThroughputOverloadedBroker.getValue()).getLocalData();
if (overloadedBrokerData.getBundles().size() > 1
&&
msgThroughputRequiredFromUnloadedBundles.getValue() >= conf.getMinUnloadMessageThroughput()) {
// Sort bundles by throughput, then pick the bundle which can help to reduce load uniformly with
// under-loaded broker
loadBundleData.entrySet().stream()
.filter(e -> overloadedBrokerData.getBundles().contains(e.getKey()))
.map((e) -> {
String bundle = e.getKey();
TimeAverageMessageData shortTermData = e.getValue().getShortTermData();
double msgThroughtput = shortTermData.getMsgThroughputIn()
+ shortTermData.getMsgThroughputOut();
return Pair.of(bundle, msgThroughtput);
}).filter(e -> !recentlyUnloadedBundles.containsKey(e.getLeft()))
.sorted((e1, e2) -> Double.compare(e2.getRight(), e1.getRight())).forEach((e) -> {
if (conf.getMaxUnloadBundleNumPerShedding() != -1
&& selectedBundlesCache.size() >= conf.getMaxUnloadBundleNumPerShedding()) {
return;
}
}
});
String bundle = e.getLeft();
double msgThroughput = e.getRight();
if (msgThroughput <= (msgThroughputRequiredFromUnloadedBundles.getValue()
+ 1000/* delta */)) {
log.info("Found bundle to unload with msgThroughput {}", msgThroughput);
msgThroughputRequiredFromUnloadedBundles.add(-msgThroughput);
selectedBundlesCache.put(msgThroughputOverloadedBroker.getValue(), bundle);
}
});
}

}
}

Expand Down
78 changes: 78 additions & 0 deletions ...roker/src/test/java/org/apache/pulsar/broker/loadbalance/impl/UniformLoadShedderTest.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -26,6 +26,7 @@
import org.testng.annotations.Test;
import static org.testng.Assert.assertEquals;
import static org.testng.Assert.assertFalse;
import static org.testng.Assert.assertTrue;

@Test(groups = "broker")
public class UniformLoadShedderTest {
Expand Down Expand Up @@ -119,4 +120,81 @@ public void testBrokerWithMultipleBundles() {
assertFalse(bundlesToUnload.isEmpty());
}

@Test
public void testOverloadBrokerSelect() {
conf.setMaxUnloadBundleNumPerShedding(1);
conf.setMaxUnloadPercentage(0.5);
int numBrokers = 5;
int numBundles = 5;
LoadData loadData = new LoadData();

LocalBrokerData[] localBrokerDatas = new LocalBrokerData[]{
new LocalBrokerData(),
new LocalBrokerData(),
new LocalBrokerData(),
new LocalBrokerData(),
new LocalBrokerData()};

String[] brokerNames = new String[]{"broker0", "broker1", "broker2", "broker3", "broker4"};

double[] brokerMsgRates = new double[]{
50000, // broker0
60000, // broker1
70000, // broker2
10000, // broker3
20000};// broker4

double[] brokerMsgThroughputs = new double[]{
50 * 1024 * 1024, // broker0
60 * 1024 * 1024, // broker1
70 * 1024 * 1024, // broker2
80 * 1024 * 1024, // broker3
10 * 1024 * 1024};// broker4


for (int brokerId = 0; brokerId < numBrokers; brokerId++) {
double msgRate = brokerMsgRates[brokerId] / numBundles;
double throughput = brokerMsgThroughputs[brokerId] / numBundles;
for (int i = 0; i < numBundles; ++i) {
String bundleName = "broker-" + brokerId + "-bundle-" + i;
localBrokerDatas[brokerId].getBundles().add(bundleName);
localBrokerDatas[brokerId].setMsgRateIn(brokerMsgRates[brokerId]);
localBrokerDatas[brokerId].setMsgThroughputIn(brokerMsgThroughputs[brokerId]);
BundleData bundle = new BundleData();

TimeAverageMessageData timeAverageMessageData = new TimeAverageMessageData();
timeAverageMessageData.setMsgRateIn(msgRate);
timeAverageMessageData.setMsgThroughputIn(throughput);
bundle.setShortTermData(timeAverageMessageData);
loadData.getBundleData().put(bundleName, bundle);
}
loadData.getBrokerData().put(brokerNames[brokerId], new BrokerData(localBrokerDatas[brokerId]));
}

// disable throughput based load shedding, enable rate based load shedding only
conf.setLoadBalancerMsgRateDifferenceShedderThreshold(50);
conf.setLoadBalancerMsgThroughputMultiplierDifferenceShedderThreshold(0);

Multimap<String, String> bundlesToUnload = uniformLoadShedder.findBundlesForUnloading(loadData, conf);
assertEquals(bundlesToUnload.size(), 1);
assertTrue(bundlesToUnload.containsKey("broker2"));


// disable rate based load shedding, enable throughput based load shedding only
conf.setLoadBalancerMsgRateDifferenceShedderThreshold(0);
conf.setLoadBalancerMsgThroughputMultiplierDifferenceShedderThreshold(2);

bundlesToUnload = uniformLoadShedder.findBundlesForUnloading(loadData, conf);
assertEquals(bundlesToUnload.size(), 1);
assertTrue(bundlesToUnload.containsKey("broker3"));

// enable both rate and throughput based load shedding, but rate based load shedding has higher priority
conf.setLoadBalancerMsgRateDifferenceShedderThreshold(50);
conf.setLoadBalancerMsgThroughputMultiplierDifferenceShedderThreshold(2);

bundlesToUnload = uniformLoadShedder.findBundlesForUnloading(loadData, conf);
assertEquals(bundlesToUnload.size(), 1);
assertTrue(bundlesToUnload.containsKey("broker2"));
}

}

0 comments on commit c60ce86

Please sign in to comment.