Skip to content

When the traffic to be unloaded is too small, it means that maxThroug… #13140

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
lordcheng10 wants to merge 3 commits into from
Closed

When the traffic to be unloaded is too small, it means that maxThroug… #13140

Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
1a6d546
When the traffic to be unloaded is too small, it means that maxThroug…
lordcheng10 Dec 5, 2021
1548d46
check style
lordcheng10 Dec 5, 2021
3f25595
check style
lordcheng10 Dec 5, 2021
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
14 changes: 14 additions & 0 deletions ...ar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/impl/UniformLoadShedder.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -49,6 +49,8 @@ public class UniformLoadShedder implements LoadSheddingStrategy {

private static final Logger log = LoggerFactory.getLogger(UniformLoadShedder.class);

private static final double MB = 1024 * 1024;

private final Multimap<String, String> selectedBundlesCache = ArrayListMultimap.create();

/**
Expand All @@ -73,6 +75,9 @@ public Multimap<String, String> findBundlesForUnloading(final LoadData loadData,
MutableDouble maxThroughputRate = new MutableDouble(-1);
MutableDouble minMsgRate = new MutableDouble(Integer.MAX_VALUE);
MutableDouble minThroughputgRate = new MutableDouble(Integer.MAX_VALUE);

double minThroughputThreshold = conf.getLoadBalancerBundleUnloadMinThroughputThreshold() * MB;

brokersData.forEach((broker, data) -> {
double msgRate = data.getLocalData().getMsgRateIn() + data.getLocalData().getMsgRateOut();
double throughputRate = data.getLocalData().getMsgThroughputIn()
Expand Down Expand Up @@ -122,6 +127,15 @@ public Multimap<String, String> findBundlesForUnloading(final LoadData loadData,
(int) ((maxThroughputRate.getValue() - minThroughputgRate.getValue()) / 2));
LocalBrokerData overloadedBrokerData = brokersData.get(overloadedBroker.getValue()).getLocalData();

if (msgThroughtputRequiredFromUnloadedBundles.getValue() < minThroughputThreshold) {
if (log.isDebugEnabled()) {
log.debug("Planning to shed throughput {} MByte/s less than "
+ "minimumThroughputThreshold {} MByte/s, stop unload.",
msgThroughtputRequiredFromUnloadedBundles.getValue() / MB, minThroughputThreshold / MB);
}
return selectedBundlesCache;
}

if (overloadedBrokerData.getBundles().size() > 1) {
// Sort bundles by throughput, then pick the bundle which can help to reduce load uniformly with
// under-loaded broker
Expand Down