You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by kw...@apache.org on 2023/03/29 09:30:34 UTC
[pulsar] branch master updated: [improve][broker] PIP-192 Improved Auto Unload Logic (#19813)
This is an automated email from the ASF dual-hosted git repository.
kwang pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/pulsar.git
The following commit(s) were added to refs/heads/master by this push:
new ee5ac8607eb [improve][broker] PIP-192 Improved Auto Unload Logic (#19813)
ee5ac8607eb is described below
commit ee5ac8607ebd7f90843789d4eead48de8600c1f0
Author: Heesung Sohn <10...@users.noreply.github.com>
AuthorDate: Wed Mar 29 02:30:24 2023 -0700
[improve][broker] PIP-192 Improved Auto Unload Logic (#19813)
Master Issue: Master Issue: https://github.com/apache/pulsar/issues/16691, https://github.com/apache/pulsar/issues/18099
### Motivation
Raising a PR to implement Master Issue: https://github.com/apache/pulsar/issues/16691, https://github.com/apache/pulsar/issues/18099
We want to reduce unload frequencies from flaky traffic.
### Modifications
This PR
- Introduced a config `loadBalancerSheddingConditionHitCountThreshold` to further restrict shedding conditions based on the hit count.
- Normalized offload traffic
- Lowered the default `loadBalanceSheddingDelayInSeconds` value from 600 to 180, as 10 mins are too long. 3 mins can be long enough to catch the new load after unloads.
- Changed the config `loadBalancerBundleLoadReportPercentage` to `loadBalancerMaxNumberOfBundlesInBundleLoadReport` to make the topk bundle count absolute instead of relative.
- Renamed `loadBalancerNamespaceBundleSplitConditionThreshold` to `loadBalancerNamespaceBundleSplitConditionHitCountThreshold` to be consistent with `*ConditionHitCountThreshold`.
- Renamed `loadBalancerMaxNumberOfBrokerTransfersPerCycle ` to `loadBalancerMaxNumberOfBrokerSheddingPerCycle`.
- Added LoadDataStore cleanup logic in BSC monitor.
- Added `msgThroughputEMA` in BrokerLoadData to smooth the broker throughput info.
- Updated Topk bundles sorted in a ascending order (instead of descending)
- Update some info logs to only show in the debug mode.
- Added load data tombstone upon Own, Releasing, Splitting
- Added the bundle ownership(isOwned) check upon split and unload.
- Added swap unload logic
---
conf/broker.conf | 85 ++++
.../apache/pulsar/broker/ServiceConfiguration.java | 23 +-
.../extensions/ExtensibleLoadManagerImpl.java | 67 ++-
.../channel/ServiceUnitStateChannel.java | 25 +
.../channel/ServiceUnitStateChannelImpl.java | 85 +++-
.../extensions/data/BrokerLoadData.java | 34 +-
.../loadbalance/extensions/models/TopKBundles.java | 2 +-
.../extensions/models/UnloadCounter.java | 4 +-
.../extensions/models/UnloadDecision.java | 2 +-
.../reporter/BrokerLoadDataReporter.java | 90 +++-
.../reporter/TopBundleLoadDataReporter.java | 63 ++-
.../scheduler/NamespaceUnloadStrategy.java | 8 +
.../extensions/scheduler/SplitScheduler.java | 16 +-
.../extensions/scheduler/TransferShedder.java | 469 ++++++++++++++-----
.../extensions/scheduler/UnloadScheduler.java | 19 +-
.../DefaultNamespaceBundleSplitStrategyImpl.java | 144 ++++--
.../strategy/LeastResourceUsageWithWeight.java | 8 +-
.../extensions/ExtensibleLoadManagerImplTest.java | 6 +-
.../channel/ServiceUnitStateChannelTest.java | 66 +++
.../extensions/data/BrokerLoadDataTest.java | 10 +-
.../extensions/models/TopKBundlesTest.java | 14 +-
.../reporter/BrokerLoadDataReporterTest.java | 93 +++-
.../reporter/TopBundleLoadDataReporterTest.java | 90 +++-
.../extensions/scheduler/TransferShedderTest.java | 515 ++++++++++++++++++---
.../extensions/scheduler/UnloadSchedulerTest.java | 12 +-
.../DefaultNamespaceBundleSplitStrategyTest.java | 49 +-
.../data/loadbalancer/NamespaceBundleStats.java | 2 +
.../policies/data/loadbalancer/ResourceUsage.java | 2 +
28 files changed, 1648 insertions(+), 355 deletions(-)
diff --git a/conf/broker.conf b/conf/broker.conf
index d52adb25456..1bbb12313f9 100644
--- a/conf/broker.conf
+++ b/conf/broker.conf
@@ -1373,6 +1373,91 @@ loadBalancerBundleUnloadMinThroughputThreshold=10
# Time to wait for the unloading of a namespace bundle
namespaceBundleUnloadingTimeoutMs=60000
+### --- Load balancer extension --- ###
+
+# Option to enable the debug mode for the load balancer logics.
+# The debug mode prints more logs to provide more information such as load balance states and decisions.
+# (only used in load balancer extension logics)
+loadBalancerDebugModeEnabled=false
+
+# The target standard deviation of the resource usage across brokers
+# (100% resource usage is 1.0 load).
+# The shedder logic tries to distribute bundle load across brokers to meet this target std.
+# The smaller value will incur load balancing more frequently.
+# (only used in load balancer extension TransferSheddeer)
+loadBalancerBrokerLoadTargetStd=0.25
+
+# Threshold to the consecutive count of fulfilled shedding(unload) conditions.
+# If the unload scheduler consecutively finds bundles that meet unload conditions
+# many times bigger than this threshold, the scheduler will shed the bundles.
+# The bigger value will incur less bundle unloading/transfers.
+# (only used in load balancer extension TransferSheddeer)
+loadBalancerSheddingConditionHitCountThreshold=3
+
+# Option to enable the bundle transfer mode when distributing bundle loads.
+# On: transfer bundles from overloaded brokers to underloaded
+# -- pre-assigns the destination broker upon unloading).
+# Off: unload bundles from overloaded brokers
+# -- post-assigns the destination broker upon lookups).
+# (only used in load balancer extension TransferSheddeer)
+loadBalancerTransferEnabled=true
+
+# Maximum number of brokers to unload bundle load for each unloading cycle.
+# The bigger value will incur more unloading/transfers for each unloading cycle.
+# (only used in load balancer extension TransferSheddeer)
+loadBalancerMaxNumberOfBrokerSheddingPerCycle=3
+
+# Delay (in seconds) to the next unloading cycle after unloading.
+# The logic tries to give enough time for brokers to recompute load after unloading.
+# The bigger value will delay the next unloading cycle longer.
+# (only used in load balancer extension TransferSheddeer)
+loadBalanceSheddingDelayInSeconds=180
+
+# Broker load data time to live (TTL in seconds).
+# The logic tries to avoid (possibly unavailable) brokers with out-dated load data,
+# and those brokers will be ignored in the load computation.
+# When tuning this value, please consider loadBalancerReportUpdateMaxIntervalMinutes.
+#The current default is loadBalancerReportUpdateMaxIntervalMinutes * 2.
+# (only used in load balancer extension TransferSheddeer)
+loadBalancerBrokerLoadDataTTLInSeconds=1800
+
+# Max number of bundles in bundle load report from each broker.
+# The load balancer distributes bundles across brokers,
+# based on topK bundle load data and other broker load data.
+# The bigger value will increase the overhead of reporting many bundles in load data.
+# (only used in load balancer extension logics)
+loadBalancerMaxNumberOfBundlesInBundleLoadReport=10
+
+# Service units'(bundles) split interval. Broker periodically checks whether
+# some service units(e.g. bundles) should split if they become hot-spots.
+# (only used in load balancer extension logics)
+loadBalancerSplitIntervalMinutes=1
+
+# Max number of bundles to split to per cycle.
+# (only used in load balancer extension logics)
+loadBalancerMaxNumberOfBundlesToSplitPerCycle=10
+
+# Threshold to the consecutive count of fulfilled split conditions.
+# If the split scheduler consecutively finds bundles that meet split conditions
+# many times bigger than this threshold, the scheduler will trigger splits on the bundles
+# (if the number of bundles is less than loadBalancerNamespaceMaximumBundles).
+# (only used in load balancer extension logics)
+loadBalancerNamespaceBundleSplitConditionHitCountThreshold=3
+
+# After this delay, the service-unit state channel tombstones any service units (e.g., bundles)
+# in semi-terminal states. For example, after splits, parent bundles will be `deleted`,
+# and then after this delay, the parent bundles' state will be `tombstoned`
+# in the service-unit state channel.
+# Pulsar does not immediately remove such semi-terminal states
+# to avoid unnecessary system confusion,
+# as the bundles in the `tombstoned` state might temporarily look available to reassign.
+# Rarely, one could lower this delay in order to aggressively clean
+# the service-unit state channel when there are a large number of bundles.
+# minimum value = 30 secs
+# (only used in load balancer extension logics)
+loadBalancerServiceUnitStateCleanUpDelayTimeInSeconds=604800
+
+
### --- Replication --- ###
# Enable replication metrics
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 e2bcf03353b..03e0fec0a53 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
@@ -2506,6 +2506,17 @@ public class ServiceConfiguration implements PulsarConfiguration {
)
private double loadBalancerBrokerLoadTargetStd = 0.25;
+ @FieldContext(
+ category = CATEGORY_LOAD_BALANCER,
+ dynamic = true,
+ doc = "Threshold to the consecutive count of fulfilled shedding(unload) conditions. "
+ + "If the unload scheduler consecutively finds bundles that meet unload conditions "
+ + "many times bigger than this threshold, the scheduler will shed the bundles. "
+ + "The bigger value will incur less bundle unloading/transfers. "
+ + "(only used in load balancer extension TransferSheddeer)"
+ )
+ private int loadBalancerSheddingConditionHitCountThreshold = 3;
+
@FieldContext(
category = CATEGORY_LOAD_BALANCER,
dynamic = true,
@@ -2521,11 +2532,11 @@ public class ServiceConfiguration implements PulsarConfiguration {
@FieldContext(
category = CATEGORY_LOAD_BALANCER,
dynamic = true,
- doc = "Maximum number of brokers to transfer bundle load for each unloading cycle. "
+ doc = "Maximum number of brokers to unload bundle load for each unloading cycle. "
+ "The bigger value will incur more unloading/transfers for each unloading cycle. "
+ "(only used in load balancer extension TransferSheddeer)"
)
- private int loadBalancerMaxNumberOfBrokerTransfersPerCycle = 3;
+ private int loadBalancerMaxNumberOfBrokerSheddingPerCycle = 3;
@FieldContext(
category = CATEGORY_LOAD_BALANCER,
@@ -2535,7 +2546,7 @@ public class ServiceConfiguration implements PulsarConfiguration {
+ "The bigger value will delay the next unloading cycle longer. "
+ "(only used in load balancer extension TransferSheddeer)"
)
- private long loadBalanceUnloadDelayInSeconds = 600;
+ private long loadBalanceSheddingDelayInSeconds = 180;
@FieldContext(
category = CATEGORY_LOAD_BALANCER,
@@ -2552,13 +2563,13 @@ public class ServiceConfiguration implements PulsarConfiguration {
@FieldContext(
dynamic = true,
category = CATEGORY_LOAD_BALANCER,
- doc = "Percentage of bundles to compute topK bundle load data from each broker. "
+ doc = "Max number of bundles in bundle load report from each broker. "
+ "The load balancer distributes bundles across brokers, "
+ "based on topK bundle load data and other broker load data."
+ "The bigger value will increase the overhead of reporting many bundles in load data. "
+ "(only used in load balancer extension logics)"
)
- private double loadBalancerBundleLoadReportPercentage = 10;
+ private int loadBalancerMaxNumberOfBundlesInBundleLoadReport = 10;
@FieldContext(
category = CATEGORY_LOAD_BALANCER,
doc = "Service units'(bundles) split interval. Broker periodically checks whether "
@@ -2582,7 +2593,7 @@ public class ServiceConfiguration implements PulsarConfiguration {
+ "(if the number of bundles is less than loadBalancerNamespaceMaximumBundles). "
+ "(only used in load balancer extension logics)"
)
- private int loadBalancerNamespaceBundleSplitConditionThreshold = 5;
+ private int loadBalancerNamespaceBundleSplitConditionHitCountThreshold = 3;
@FieldContext(
category = CATEGORY_LOAD_BALANCER,
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/extensions/ExtensibleLoadManagerImpl.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/extensions/ExtensibleLoadManagerImpl.java
index 7aefef596e7..3e078d0a5eb 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/extensions/ExtensibleLoadManagerImpl.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/extensions/ExtensibleLoadManagerImpl.java
@@ -85,6 +85,7 @@ import org.apache.pulsar.common.stats.Metrics;
import org.apache.pulsar.common.util.FutureUtil;
import org.apache.pulsar.common.util.collections.ConcurrentOpenHashMap;
import org.apache.pulsar.metadata.api.coordination.LeaderElectionState;
+import org.slf4j.Logger;
@Slf4j
public class ExtensibleLoadManagerImpl implements ExtensibleLoadManager {
@@ -101,6 +102,8 @@ public class ExtensibleLoadManagerImpl implements ExtensibleLoadManager {
private static final long MAX_ROLE_CHANGE_RETRY_DELAY_IN_MILLIS = 200;
+ private static final long MONITOR_INTERVAL_IN_MILLIS = 120_000;
+
private PulsarService pulsar;
private ServiceConfiguration conf;
@@ -108,10 +111,12 @@ public class ExtensibleLoadManagerImpl implements ExtensibleLoadManager {
@Getter
private BrokerRegistry brokerRegistry;
+ @Getter
private ServiceUnitStateChannel serviceUnitStateChannel;
private AntiAffinityGroupPolicyFilter antiAffinityGroupPolicyFilter;
+ @Getter
private AntiAffinityGroupPolicyHelper antiAffinityGroupPolicyHelper;
private LoadDataStore<BrokerLoadData> brokerLoadDataStore;
@@ -139,6 +144,8 @@ public class ExtensibleLoadManagerImpl implements ExtensibleLoadManager {
private ScheduledFuture brokerLoadDataReportTask;
private ScheduledFuture topBundlesLoadDataReportTask;
+
+ private ScheduledFuture monitorTask;
private SplitScheduler splitScheduler;
private UnloadManager unloadManager;
@@ -148,6 +155,7 @@ public class ExtensibleLoadManagerImpl implements ExtensibleLoadManager {
private boolean started = false;
private final AssignCounter assignCounter = new AssignCounter();
+ @Getter
private final UnloadCounter unloadCounter = new UnloadCounter();
private final SplitCounter splitCounter = new SplitCounter();
@@ -162,14 +170,14 @@ public class ExtensibleLoadManagerImpl implements ExtensibleLoadManager {
lookupRequests = ConcurrentOpenHashMap.<String,
CompletableFuture<Optional<BrokerLookupData>>>newBuilder()
.build();
- private final CountDownLatch loadStoreInitWaiter = new CountDownLatch(1);
+ private final CountDownLatch initWaiter = new CountDownLatch(1);
public enum Role {
Leader,
Follower
}
- private Role role;
+ private volatile Role role;
/**
* Life cycle: Constructor -> initialize -> start -> close.
@@ -194,6 +202,10 @@ public class ExtensibleLoadManagerImpl implements ExtensibleLoadManager {
return loadManagerWrapper.get();
}
+ public static boolean debug(ServiceConfiguration config, Logger log) {
+ return config.isLoadBalancerDebugModeEnabled() || log.isDebugEnabled();
+ }
+
@Override
public void start() throws PulsarServerException {
if (this.started) {
@@ -231,7 +243,6 @@ public class ExtensibleLoadManagerImpl implements ExtensibleLoadManager {
this.brokerLoadDataStore.startTableView();
this.topBundlesLoadDataStore = LoadDataStoreFactory
.create(pulsar.getClient(), TOP_BUNDLES_LOAD_DATA_STORE_TOPIC, TopBundlesLoadData.class);
- this.loadStoreInitWaiter.countDown();
} catch (LoadDataStoreException e) {
throw new PulsarServerException(e);
}
@@ -247,7 +258,8 @@ public class ExtensibleLoadManagerImpl implements ExtensibleLoadManager {
this.topBundleLoadDataReporter =
new TopBundleLoadDataReporter(pulsar, brokerRegistry.getBrokerId(), topBundlesLoadDataStore);
-
+ this.serviceUnitStateChannel.listen(brokerLoadDataReporter);
+ this.serviceUnitStateChannel.listen(topBundleLoadDataReporter);
var interval = conf.getLoadBalancerReportUpdateMinIntervalMillis();
this.brokerLoadDataReportTask = this.pulsar.getLoadManagerExecutor()
.scheduleAtFixedRate(() -> {
@@ -273,13 +285,21 @@ public class ExtensibleLoadManagerImpl implements ExtensibleLoadManager {
interval,
interval, TimeUnit.MILLISECONDS);
+ this.monitorTask = this.pulsar.getLoadManagerExecutor()
+ .scheduleAtFixedRate(() -> {
+ monitor();
+ },
+ MONITOR_INTERVAL_IN_MILLIS,
+ MONITOR_INTERVAL_IN_MILLIS, TimeUnit.MILLISECONDS);
+
this.unloadScheduler = new UnloadScheduler(
- pulsar, pulsar.getLoadManagerExecutor(), unloadManager,
- context, serviceUnitStateChannel, antiAffinityGroupPolicyHelper, unloadCounter, unloadMetrics);
+ pulsar, pulsar.getLoadManagerExecutor(), unloadManager, context, serviceUnitStateChannel,
+ unloadCounter, unloadMetrics);
this.unloadScheduler.start();
this.splitScheduler = new SplitScheduler(
pulsar, serviceUnitStateChannel, splitManager, splitCounter, splitMetrics, context);
this.splitScheduler.start();
+ this.initWaiter.countDown();
this.started = true;
}
@@ -327,7 +347,7 @@ public class ExtensibleLoadManagerImpl implements ExtensibleLoadManager {
return owner.thenCompose(broker -> {
if (broker.isEmpty()) {
String errorMsg = String.format(
- "Failed to look up a broker registry:%s for bundle:%s", broker, bundle);
+ "Failed to get or assign the owner for bundle:%s", bundle);
log.error(errorMsg);
throw new IllegalStateException(errorMsg);
}
@@ -498,6 +518,10 @@ public class ExtensibleLoadManagerImpl implements ExtensibleLoadManager {
topBundlesLoadDataReportTask.cancel(true);
}
+ if (monitorTask != null) {
+ monitorTask.cancel(true);
+ }
+
this.brokerLoadDataStore.close();
this.topBundlesLoadDataStore.close();
this.unloadScheduler.close();
@@ -539,8 +563,8 @@ public class ExtensibleLoadManagerImpl implements ExtensibleLoadManager {
int retry = 0;
while (true) {
try {
+ initWaiter.await();
serviceUnitStateChannel.scheduleOwnershipMonitor();
- loadStoreInitWaiter.await();
topBundlesLoadDataStore.startTableView();
unloadScheduler.start();
break;
@@ -575,8 +599,8 @@ public class ExtensibleLoadManagerImpl implements ExtensibleLoadManager {
int retry = 0;
while (true) {
try {
+ initWaiter.await();
serviceUnitStateChannel.cancelOwnershipMonitor();
- loadStoreInitWaiter.await();
topBundlesLoadDataStore.closeTableView();
unloadScheduler.close();
break;
@@ -631,4 +655,29 @@ public class ExtensibleLoadManagerImpl implements ExtensibleLoadManager {
return metricsCollection;
}
+
+ private void monitor() {
+ try {
+ initWaiter.await();
+
+ // Monitor role
+ // Periodically check the role in case ZK watcher fails.
+ var isChannelOwner = serviceUnitStateChannel.isChannelOwner();
+ if (isChannelOwner) {
+ if (role != Leader) {
+ log.warn("Current role:{} does not match with the channel ownership:{}. "
+ + "Playing the leader role.", role, isChannelOwner);
+ playLeader();
+ }
+ } else {
+ if (role != Follower) {
+ log.warn("Current role:{} does not match with the channel ownership:{}. "
+ + "Playing the follower role.", role, isChannelOwner);
+ playFollower();
+ }
+ }
+ } catch (Throwable e) {
+ log.error("Failed to get the channel ownership.", e);
+ }
+ }
}
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/extensions/channel/ServiceUnitStateChannel.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/extensions/channel/ServiceUnitStateChannel.java
index 719c72a67b4..4782a31fe0c 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/extensions/channel/ServiceUnitStateChannel.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/extensions/channel/ServiceUnitStateChannel.java
@@ -64,6 +64,12 @@ public interface ServiceUnitStateChannel extends Closeable {
*/
CompletableFuture<Boolean> isChannelOwnerAsync();
+ /**
+ * Checks if the current broker is the owner broker of the system topic in this channel.
+ * @return True if the current broker is the owner. Otherwise, false.
+ */
+ boolean isChannelOwner();
+
/**
* Handles the metadata session events to track
* if the connection between the broker and metadata store is stable or not.
@@ -125,6 +131,25 @@ public interface ServiceUnitStateChannel extends Closeable {
*/
CompletableFuture<Optional<String>> getOwnerAsync(String serviceUnit);
+ /**
+ * Checks if the target broker is the owner of the service unit.
+ *
+ *
+ * @param serviceUnit (e.g. bundle)
+ * @param targetBroker
+ * @return true if the target broker is the owner. false if unknown.
+ */
+ boolean isOwner(String serviceUnit, String targetBroker);
+
+ /**
+ * Checks if the current broker is the owner of the service unit.
+ *
+ *
+ * @param serviceUnit (e.g. bundle))
+ * @return true if the current broker is the owner. false if unknown.
+ */
+ boolean isOwner(String serviceUnit);
+
/**
* Asynchronously publishes the service unit assignment event to the system topic in this channel.
* It de-duplicates assignment events if there is any ongoing assignment event for the same service unit.
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/extensions/channel/ServiceUnitStateChannelImpl.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/extensions/channel/ServiceUnitStateChannelImpl.java
index ca9e55923e7..e7fb59450b2 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/extensions/channel/ServiceUnitStateChannelImpl.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/extensions/channel/ServiceUnitStateChannelImpl.java
@@ -70,6 +70,7 @@ import org.apache.pulsar.broker.PulsarService;
import org.apache.pulsar.broker.ServiceConfiguration;
import org.apache.pulsar.broker.loadbalance.LeaderElectionService;
import org.apache.pulsar.broker.loadbalance.extensions.BrokerRegistry;
+import org.apache.pulsar.broker.loadbalance.extensions.ExtensibleLoadManagerImpl;
import org.apache.pulsar.broker.loadbalance.extensions.ExtensibleLoadManagerWrapper;
import org.apache.pulsar.broker.loadbalance.extensions.LoadManagerContext;
import org.apache.pulsar.broker.loadbalance.extensions.manager.StateChangeListener;
@@ -233,7 +234,7 @@ public class ServiceUnitStateChannelImpl implements ServiceUnitStateChannel {
log.info("Failed to monitor the ownerships. will retry..", e);
}
},
- ownershipMonitorDelayTimeInSecs, ownershipMonitorDelayTimeInSecs, SECONDS);
+ 0, ownershipMonitorDelayTimeInSecs, SECONDS);
log.info("This leader broker:{} started the ownership monitor.",
lookupServiceAddress);
}
@@ -390,7 +391,7 @@ public class ServiceUnitStateChannelImpl implements ServiceUnitStateChannel {
}
private boolean debug() {
- return config.isLoadBalancerDebugModeEnabled() || log.isDebugEnabled();
+ return ExtensibleLoadManagerImpl.debug(config, log);
}
public CompletableFuture<Optional<String>> getChannelOwnerAsync() {
@@ -426,7 +427,7 @@ public class ServiceUnitStateChannelImpl implements ServiceUnitStateChannel {
});
}
- private boolean isChannelOwner() {
+ public boolean isChannelOwner() {
try {
return isChannelOwnerAsync().get(
MAX_CHANNEL_OWNER_ELECTION_WAITING_TIME_IN_SECS, TimeUnit.SECONDS);
@@ -437,6 +438,25 @@ public class ServiceUnitStateChannelImpl implements ServiceUnitStateChannel {
}
}
+ public boolean isOwner(String serviceUnit, String targetBroker) {
+ if (!validateChannelState(Started, true)) {
+ throw new IllegalStateException("Invalid channel state:" + channelState.name());
+ }
+ var ownerFuture = getOwnerAsync(serviceUnit);
+ if (!ownerFuture.isDone() || ownerFuture.isCompletedExceptionally() || ownerFuture.isCancelled()) {
+ return false;
+ }
+ var owner = ownerFuture.join();
+ if (owner.isPresent() && StringUtils.equals(targetBroker, owner.get())) {
+ return true;
+ }
+ return false;
+ }
+
+ public boolean isOwner(String serviceUnit) {
+ return isOwner(serviceUnit, lookupServiceAddress);
+ }
+
public CompletableFuture<Optional<String>> getOwnerAsync(String serviceUnit) {
if (!validateChannelState(Started, true)) {
return CompletableFuture.failedFuture(
@@ -951,9 +971,11 @@ public class ServiceUnitStateChannelImpl implements ServiceUnitStateChannel {
+ " Active cleanup job count:{}",
broker, cleanupJobs.size());
} else {
- log.info("Failed to cancel the ownership cleanup for broker:{}."
- + " There was no scheduled cleanup job. Active cleanup job count:{}",
- broker, cleanupJobs.size());
+ if (debug()) {
+ log.info("No needs to cancel the ownership cleanup for broker:{}."
+ + " There was no scheduled cleanup job. Active cleanup job count:{}",
+ broker, cleanupJobs.size());
+ }
}
}
@@ -989,6 +1011,8 @@ public class ServiceUnitStateChannelImpl implements ServiceUnitStateChannel {
log.error("Failed to run the cleanup job for the broker {}, "
+ "totalCleanupErrorCnt:{}.",
broker, totalCleanupErrorCnt.incrementAndGet(), e);
+ } finally {
+ cleanupJobs.remove(broker);
}
}
, delayed);
@@ -1057,7 +1081,11 @@ public class ServiceUnitStateChannelImpl implements ServiceUnitStateChannel {
double cleanupTime = TimeUnit.NANOSECONDS
.toMillis((System.nanoTime() - startTime));
- // TODO: clean load data stores
+
+ // clean load data stores
+ getContext().topBundleLoadDataStore().removeAsync(broker);
+ getContext().brokerLoadDataStore().removeAsync(broker);
+
log.info("Completed a cleanup for the inactive broker:{} in {} ms. "
+ "Cleaned up orphan service units: orphanServiceUnitCleanupCnt:{}, "
+ "approximate cleanupErrorCnt:{}, metrics:{} ",
@@ -1066,7 +1094,7 @@ public class ServiceUnitStateChannelImpl implements ServiceUnitStateChannel {
orphanServiceUnitCleanupCnt,
totalCleanupErrorCntStart - totalCleanupErrorCnt.get(),
printCleanupMetrics());
- cleanupJobs.remove(broker);
+
}
private Optional<ServiceUnitStateData> getRollForwardStateData(
@@ -1123,7 +1151,11 @@ public class ServiceUnitStateChannelImpl implements ServiceUnitStateChannel {
return;
}
- log.info("Started the ownership monitor run for activeBrokerCount:{}", brokers.size());
+ var debug = debug();
+ if (debug) {
+ log.info("Started the ownership monitor run for activeBrokerCount:{}", brokers.size());
+ }
+
long startTime = System.nanoTime();
Set<String> inactiveBrokers = new HashSet<>();
Set<String> activeBrokers = new HashSet<>(brokers);
@@ -1199,27 +1231,32 @@ public class ServiceUnitStateChannelImpl implements ServiceUnitStateChannel {
log.error("Failed to flush the in-flight messages.", e);
}
+ boolean cleaned = false;
if (serviceUnitTombstoneCleanupCnt > 0) {
this.totalServiceUnitTombstoneCleanupCnt += serviceUnitTombstoneCleanupCnt;
+ cleaned = true;
}
if (orphanServiceUnitCleanupCnt > 0) {
this.totalOrphanServiceUnitCleanupCnt += orphanServiceUnitCleanupCnt;
+ cleaned = true;
}
- double monitorTime = TimeUnit.NANOSECONDS
- .toMillis((System.nanoTime() - startTime));
- log.info("Completed the ownership monitor run in {} ms. "
- + "Scheduled cleanups for inactive brokers:{}. inactiveBrokerCount:{}. "
- + "Published cleanups for orphan service units, orphanServiceUnitCleanupCnt:{}. "
- + "Tombstoned semi-terminal state service units, serviceUnitTombstoneCleanupCnt:{}. "
- + "Approximate cleanupErrorCnt:{}, metrics:{}. ",
- monitorTime,
- inactiveBrokers, inactiveBrokers.size(),
- orphanServiceUnitCleanupCnt,
- serviceUnitTombstoneCleanupCnt,
- totalCleanupErrorCntStart - totalCleanupErrorCnt.get(),
- printCleanupMetrics());
+ if (debug || cleaned) {
+ double monitorTime = TimeUnit.NANOSECONDS
+ .toMillis((System.nanoTime() - startTime));
+ log.info("Completed the ownership monitor run in {} ms. "
+ + "Scheduled cleanups for inactive brokers:{}. inactiveBrokerCount:{}. "
+ + "Published cleanups for orphan service units, orphanServiceUnitCleanupCnt:{}. "
+ + "Tombstoned semi-terminal state service units, serviceUnitTombstoneCleanupCnt:{}. "
+ + "Approximate cleanupErrorCnt:{}, metrics:{}. ",
+ monitorTime,
+ inactiveBrokers, inactiveBrokers.size(),
+ orphanServiceUnitCleanupCnt,
+ serviceUnitTombstoneCleanupCnt,
+ totalCleanupErrorCntStart - totalCleanupErrorCnt.get(),
+ printCleanupMetrics());
+ }
}
@@ -1353,4 +1390,8 @@ public class ServiceUnitStateChannelImpl implements ServiceUnitStateChannel {
public Set<Map.Entry<String, ServiceUnitStateData>> getOwnershipEntrySet() {
return tableview.entrySet();
}
+
+ public static ServiceUnitStateChannel get(PulsarService pulsar) {
+ return ExtensibleLoadManagerImpl.get(pulsar.getLoadManager().get()).getServiceUnitStateChannel();
+ }
}
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/extensions/data/BrokerLoadData.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/extensions/data/BrokerLoadData.java
index 8b373bc5954..48665d39a0d 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/extensions/data/BrokerLoadData.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/extensions/data/BrokerLoadData.java
@@ -75,6 +75,7 @@ public class BrokerLoadData {
* The historical resource percentage is configured by loadBalancerHistoryResourcePercentage.
*/
private double weightedMaxEMA;
+ private double msgThroughputEMA;
private long updatedAt;
@Setter
@@ -88,6 +89,7 @@ public class BrokerLoadData {
bandwidthOut = new ResourceUsage();
maxResourceUsage = DEFAULT_RESOURCE_USAGE;
weightedMaxEMA = DEFAULT_RESOURCE_USAGE;
+ msgThroughputEMA = 0;
}
/**
@@ -142,6 +144,7 @@ public class BrokerLoadData {
bundleCount = other.bundleCount;
topics = other.topics;
weightedMaxEMA = other.weightedMaxEMA;
+ msgThroughputEMA = other.msgThroughputEMA;
maxResourceUsage = other.maxResourceUsage;
updatedAt = other.updatedAt;
reportedAt = other.reportedAt;
@@ -161,6 +164,7 @@ public class BrokerLoadData {
private void updateFeatures(ServiceConfiguration conf) {
updateMaxResourceUsage();
updateWeightedMaxEMA(conf);
+ updateMsgThroughputEMA(conf);
}
private void updateMaxResourceUsage() {
@@ -188,6 +192,32 @@ public class BrokerLoadData {
weightedMaxEMA * historyPercentage + (1 - historyPercentage) * weightedMax;
}
+ private void updateMsgThroughputEMA(ServiceConfiguration conf) {
+ var historyPercentage = conf.getLoadBalancerHistoryResourcePercentage();
+ double msgThroughput = msgThroughputIn + msgThroughputOut;
+ msgThroughputEMA = updatedAt == 0 ? msgThroughput :
+ msgThroughputEMA * historyPercentage + (1 - historyPercentage) * msgThroughput;
+ }
+
+ public void clear() {
+ cpu = new ResourceUsage();
+ memory = new ResourceUsage();
+ directMemory = new ResourceUsage();
+ bandwidthIn = new ResourceUsage();
+ bandwidthOut = new ResourceUsage();
+ maxResourceUsage = DEFAULT_RESOURCE_USAGE;
+ weightedMaxEMA = DEFAULT_RESOURCE_USAGE;
+ msgThroughputEMA = 0;
+ msgThroughputIn = 0;
+ msgThroughputOut = 0;
+ msgRateIn = 0.0;
+ msgRateOut = 0.0;
+ bundleCount = 0;
+ topics = 0;
+ updatedAt = 0;
+ reportedAt = 0;
+ }
+
public String toString(ServiceConfiguration conf) {
return String.format("cpu= %.2f%%, memory= %.2f%%, directMemory= %.2f%%, "
+ "bandwithIn= %.2f%%, bandwithOut= %.2f%%, "
@@ -195,7 +225,7 @@ public class BrokerLoadData {
+ "bandwithInResourceWeight= %f, bandwithOutResourceWeight= %f, "
+ "msgThroughputIn= %.2f, msgThroughputOut= %.2f, msgRateIn= %.2f, msgRateOut= %.2f, "
+ "bundleCount= %d, "
- + "maxResourceUsage= %.2f%%, weightedMaxEMA= %.2f%%, "
+ + "maxResourceUsage= %.2f%%, weightedMaxEMA= %.2f%%, msgThroughputEMA= %.2f, "
+ "updatedAt= %d, reportedAt= %d",
cpu.percentUsage(), memory.percentUsage(), directMemory.percentUsage(),
@@ -207,7 +237,7 @@ public class BrokerLoadData {
conf.getLoadBalancerBandwithOutResourceWeight(),
msgThroughputIn, msgThroughputOut, msgRateIn, msgRateOut,
bundleCount,
- maxResourceUsage * 100, weightedMaxEMA * 100,
+ maxResourceUsage * 100, weightedMaxEMA * 100, msgThroughputEMA,
updatedAt, reportedAt
);
}
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/extensions/models/TopKBundles.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/extensions/models/TopKBundles.java
index d49361741be..2f5c32197c1 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/extensions/models/TopKBundles.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/extensions/models/TopKBundles.java
@@ -86,7 +86,7 @@ public class TopKBundles {
topk = Math.min(topk, arr.size());
partitionSort(arr, topk);
- for (int i = 0; i < topk; i++) {
+ for (int i = topk - 1; i >= 0; i--) {
var etr = arr.get(i);
topKBundlesLoadData.add(
new TopBundlesLoadData.BundleLoadData(etr.getKey(), (NamespaceBundleStats) etr.getValue()));
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/extensions/models/UnloadCounter.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/extensions/models/UnloadCounter.java
index 37483b58b53..4a5d41f7576 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/extensions/models/UnloadCounter.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/extensions/models/UnloadCounter.java
@@ -22,8 +22,8 @@ import static org.apache.pulsar.broker.loadbalance.extensions.models.UnloadDecis
import static org.apache.pulsar.broker.loadbalance.extensions.models.UnloadDecision.Label.Skip;
import static org.apache.pulsar.broker.loadbalance.extensions.models.UnloadDecision.Label.Success;
import static org.apache.pulsar.broker.loadbalance.extensions.models.UnloadDecision.Reason.Admin;
-import static org.apache.pulsar.broker.loadbalance.extensions.models.UnloadDecision.Reason.Balanced;
import static org.apache.pulsar.broker.loadbalance.extensions.models.UnloadDecision.Reason.CoolDown;
+import static org.apache.pulsar.broker.loadbalance.extensions.models.UnloadDecision.Reason.HitCount;
import static org.apache.pulsar.broker.loadbalance.extensions.models.UnloadDecision.Reason.NoBrokers;
import static org.apache.pulsar.broker.loadbalance.extensions.models.UnloadDecision.Reason.NoBundles;
import static org.apache.pulsar.broker.loadbalance.extensions.models.UnloadDecision.Reason.NoLoadData;
@@ -68,7 +68,7 @@ public class UnloadCounter {
Underloaded, new AtomicLong(),
Admin, new AtomicLong()),
Skip, Map.of(
- Balanced, new AtomicLong(),
+ HitCount, new AtomicLong(),
NoBundles, new AtomicLong(),
CoolDown, new AtomicLong(),
OutDatedData, new AtomicLong(),
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/extensions/models/UnloadDecision.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/extensions/models/UnloadDecision.java
index e1087ab6e53..1301b0708a5 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/extensions/models/UnloadDecision.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/extensions/models/UnloadDecision.java
@@ -41,7 +41,7 @@ public class UnloadDecision {
public enum Reason {
Overloaded,
Underloaded,
- Balanced,
+ HitCount,
NoBundles,
CoolDown,
OutDatedData,
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/extensions/reporter/BrokerLoadDataReporter.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/extensions/reporter/BrokerLoadDataReporter.java
index 256e52c4554..b07acfda7f7 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/extensions/reporter/BrokerLoadDataReporter.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/extensions/reporter/BrokerLoadDataReporter.java
@@ -18,15 +18,21 @@
*/
package org.apache.pulsar.broker.loadbalance.extensions.reporter;
+import com.google.common.annotations.VisibleForTesting;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.TimeUnit;
import lombok.Getter;
import lombok.extern.slf4j.Slf4j;
+import org.apache.commons.lang3.StringUtils;
import org.apache.commons.lang3.SystemUtils;
import org.apache.pulsar.broker.PulsarService;
import org.apache.pulsar.broker.ServiceConfiguration;
import org.apache.pulsar.broker.loadbalance.BrokerHostUsage;
+import org.apache.pulsar.broker.loadbalance.extensions.ExtensibleLoadManagerImpl;
+import org.apache.pulsar.broker.loadbalance.extensions.channel.ServiceUnitState;
+import org.apache.pulsar.broker.loadbalance.extensions.channel.ServiceUnitStateData;
import org.apache.pulsar.broker.loadbalance.extensions.data.BrokerLoadData;
+import org.apache.pulsar.broker.loadbalance.extensions.manager.StateChangeListener;
import org.apache.pulsar.broker.loadbalance.extensions.store.LoadDataStore;
import org.apache.pulsar.broker.loadbalance.impl.GenericBrokerHostUsageImpl;
import org.apache.pulsar.broker.loadbalance.impl.LinuxBrokerHostUsageImpl;
@@ -37,7 +43,9 @@ import org.apache.pulsar.policies.data.loadbalancer.SystemResourceUsage;
* The broker load data reporter.
*/
@Slf4j
-public class BrokerLoadDataReporter implements LoadDataReporter<BrokerLoadData> {
+public class BrokerLoadDataReporter implements LoadDataReporter<BrokerLoadData>, StateChangeListener {
+
+ private static final long TOMBSTONE_DELAY_IN_MILLIS = 1000 * 10;
private final PulsarService pulsar;
@@ -54,6 +62,10 @@ public class BrokerLoadDataReporter implements LoadDataReporter<BrokerLoadData>
private final BrokerLoadData lastData;
+ private volatile long lastTombstonedAt;
+
+ private long tombstoneDelayInMillis;
+
public BrokerLoadDataReporter(PulsarService pulsar,
String lookupServiceAddress,
LoadDataStore<BrokerLoadData> brokerLoadDataStore) {
@@ -68,6 +80,7 @@ public class BrokerLoadDataReporter implements LoadDataReporter<BrokerLoadData>
}
this.localData = new BrokerLoadData();
this.lastData = new BrokerLoadData();
+ this.tombstoneDelayInMillis = TOMBSTONE_DELAY_IN_MILLIS;
}
@@ -92,8 +105,11 @@ public class BrokerLoadDataReporter implements LoadDataReporter<BrokerLoadData>
@Override
public CompletableFuture<Void> reportAsync(boolean force) {
BrokerLoadData newLoadData = this.generateLoadData();
+ boolean debug = ExtensibleLoadManagerImpl.debug(conf, log);
if (force || needBrokerDataUpdate()) {
- log.info("publishing load report:{}", localData.toString(conf));
+ if (debug) {
+ log.info("publishing load report:{}", localData.toString(conf));
+ }
CompletableFuture<Void> future =
this.brokerLoadDataStore.pushAsync(this.lookupServiceAddress, newLoadData);
future.whenComplete((__, ex) -> {
@@ -106,7 +122,9 @@ public class BrokerLoadDataReporter implements LoadDataReporter<BrokerLoadData>
});
return future;
} else {
- log.info("skipping load report:{}", localData.toString(conf));
+ if (debug) {
+ log.info("skipping load report:{}", localData.toString(conf));
+ }
}
return CompletableFuture.completedFuture(null);
}
@@ -117,10 +135,13 @@ public class BrokerLoadDataReporter implements LoadDataReporter<BrokerLoadData>
final long updateMaxIntervalMillis = TimeUnit.MINUTES
.toMillis(loadBalancerReportUpdateMaxIntervalMinutes);
long timeSinceLastReportWrittenToStore = System.currentTimeMillis() - localData.getReportedAt();
+ boolean debug = ExtensibleLoadManagerImpl.debug(conf, log);
if (timeSinceLastReportWrittenToStore > updateMaxIntervalMillis) {
- log.info("Writing local data to metadata store because time since last"
- + " update exceeded threshold of {} minutes",
- loadBalancerReportUpdateMaxIntervalMinutes);
+ if (debug) {
+ log.info("Writing local data to metadata store because time since last"
+ + " update exceeded threshold of {} minutes",
+ loadBalancerReportUpdateMaxIntervalMinutes);
+ }
// Always update after surpassing the maximum interval.
return true;
}
@@ -133,10 +154,13 @@ public class BrokerLoadDataReporter implements LoadDataReporter<BrokerLoadData>
localData.getMsgThroughputIn() + localData.getMsgThroughputOut()),
percentChange(lastData.getBundleCount(), localData.getBundleCount()))));
if (maxChange > loadBalancerReportUpdateThresholdPercentage) {
- log.info("Writing local data to metadata store because maximum change {}% exceeded threshold {}%; "
- + "time since last report written is {} seconds", maxChange,
- loadBalancerReportUpdateThresholdPercentage,
- timeSinceLastReportWrittenToStore / 1000.0);
+ if (debug) {
+ log.info(String.format("Writing local data to metadata store "
+ + "because maximum change %.2f%% exceeded threshold %d%%. "
+ + "Time since last report written is %.2f%% seconds", maxChange,
+ loadBalancerReportUpdateThresholdPercentage,
+ timeSinceLastReportWrittenToStore / 1000.0));
+ }
return true;
}
return false;
@@ -152,4 +176,50 @@ public class BrokerLoadDataReporter implements LoadDataReporter<BrokerLoadData>
}
return 100 * Math.abs((oldValue - newValue) / oldValue);
}
+
+ @VisibleForTesting
+ protected void tombstone() {
+ var now = System.currentTimeMillis();
+ if (now - lastTombstonedAt < tombstoneDelayInMillis) {
+ return;
+ }
+ var lastSuccessfulTombstonedAt = lastTombstonedAt;
+ lastTombstonedAt = now; // dedup first
+ brokerLoadDataStore.removeAsync(lookupServiceAddress)
+ .whenComplete((__, e) -> {
+ if (e != null) {
+ log.error("Failed to clean broker load data.", e);
+ lastTombstonedAt = lastSuccessfulTombstonedAt;
+ } else {
+ boolean debug = ExtensibleLoadManagerImpl.debug(conf, log);
+ if (debug) {
+ log.info("Cleaned broker load data.");
+ }
+ }
+ }
+ );
+
+ }
+
+ @Override
+ public void handleEvent(String serviceUnit, ServiceUnitStateData data, Throwable t) {
+ if (t != null) {
+ return;
+ }
+ ServiceUnitState state = ServiceUnitStateData.state(data);
+ switch (state) {
+ case Releasing, Splitting -> {
+ if (StringUtils.equals(data.sourceBroker(), lookupServiceAddress)) {
+ localData.clear();
+ tombstone();
+ }
+ }
+ case Owned -> {
+ if (StringUtils.equals(data.dstBroker(), lookupServiceAddress)) {
+ localData.clear();
+ tombstone();
+ }
+ }
+ }
+ }
}
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/extensions/reporter/TopBundleLoadDataReporter.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/extensions/reporter/TopBundleLoadDataReporter.java
index cc1a39add5d..0fa37d3687c 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/extensions/reporter/TopBundleLoadDataReporter.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/extensions/reporter/TopBundleLoadDataReporter.java
@@ -18,10 +18,16 @@
*/
package org.apache.pulsar.broker.loadbalance.extensions.reporter;
+import com.google.common.annotations.VisibleForTesting;
import java.util.concurrent.CompletableFuture;
import lombok.extern.slf4j.Slf4j;
+import org.apache.commons.lang3.StringUtils;
import org.apache.pulsar.broker.PulsarService;
+import org.apache.pulsar.broker.loadbalance.extensions.ExtensibleLoadManagerImpl;
+import org.apache.pulsar.broker.loadbalance.extensions.channel.ServiceUnitState;
+import org.apache.pulsar.broker.loadbalance.extensions.channel.ServiceUnitStateData;
import org.apache.pulsar.broker.loadbalance.extensions.data.TopBundlesLoadData;
+import org.apache.pulsar.broker.loadbalance.extensions.manager.StateChangeListener;
import org.apache.pulsar.broker.loadbalance.extensions.models.TopKBundles;
import org.apache.pulsar.broker.loadbalance.extensions.store.LoadDataStore;
@@ -29,7 +35,9 @@ import org.apache.pulsar.broker.loadbalance.extensions.store.LoadDataStore;
* The top k highest-loaded bundles' load data reporter.
*/
@Slf4j
-public class TopBundleLoadDataReporter implements LoadDataReporter<TopBundlesLoadData> {
+public class TopBundleLoadDataReporter implements LoadDataReporter<TopBundlesLoadData>, StateChangeListener {
+
+ private static final long TOMBSTONE_DELAY_IN_MILLIS = 1000 * 10;
private final PulsarService pulsar;
@@ -41,6 +49,9 @@ public class TopBundleLoadDataReporter implements LoadDataReporter<TopBundlesLoa
private long lastBundleStatsUpdatedAt;
+ private volatile long lastTombstonedAt;
+ private long tombstoneDelayInMillis;
+
public TopBundleLoadDataReporter(PulsarService pulsar,
String lookupServiceAddress,
LoadDataStore<TopBundlesLoadData> bundleLoadDataStore) {
@@ -49,6 +60,7 @@ public class TopBundleLoadDataReporter implements LoadDataReporter<TopBundlesLoa
this.bundleLoadDataStore = bundleLoadDataStore;
this.lastBundleStatsUpdatedAt = 0;
this.topKBundles = new TopKBundles(pulsar);
+ this.tombstoneDelayInMillis = TOMBSTONE_DELAY_IN_MILLIS;
}
@Override
@@ -60,8 +72,7 @@ public class TopBundleLoadDataReporter implements LoadDataReporter<TopBundlesLoa
var pulsarStatsUpdatedAt = pulsarStats.getUpdatedAt();
if (pulsarStatsUpdatedAt > lastBundleStatsUpdatedAt) {
var bundleStats = pulsar.getBrokerService().getBundleStats();
- double percentage = pulsar.getConfiguration().getLoadBalancerBundleLoadReportPercentage();
- int topk = Math.max(1, (int) (bundleStats.size() * percentage / 100.0));
+ int topk = pulsar.getConfiguration().getLoadBalancerMaxNumberOfBundlesInBundleLoadReport();
topKBundles.update(bundleStats, topk);
lastBundleStatsUpdatedAt = pulsarStatsUpdatedAt;
result = topKBundles.getLoadData();
@@ -74,6 +85,9 @@ public class TopBundleLoadDataReporter implements LoadDataReporter<TopBundlesLoa
public CompletableFuture<Void> reportAsync(boolean force) {
var topBundlesLoadData = generateLoadData();
if (topBundlesLoadData != null || force) {
+ if (ExtensibleLoadManagerImpl.debug(pulsar.getConfiguration(), log)) {
+ log.info("Reporting TopBundlesLoadData:{}", topKBundles.getLoadData());
+ }
return this.bundleLoadDataStore.pushAsync(lookupServiceAddress, topKBundles.getLoadData())
.exceptionally(e -> {
log.error("Failed to report top-bundles load data.", e);
@@ -83,4 +97,47 @@ public class TopBundleLoadDataReporter implements LoadDataReporter<TopBundlesLoa
return CompletableFuture.completedFuture(null);
}
}
+
+ @VisibleForTesting
+ protected void tombstone() {
+ var now = System.currentTimeMillis();
+ if (now - lastTombstonedAt < tombstoneDelayInMillis) {
+ return;
+ }
+ var lastSuccessfulTombstonedAt = lastTombstonedAt;
+ lastTombstonedAt = now; // dedup first
+ bundleLoadDataStore.removeAsync(lookupServiceAddress)
+ .whenComplete((__, e) -> {
+ if (e != null) {
+ log.error("Failed to clean broker load data.", e);
+ lastTombstonedAt = lastSuccessfulTombstonedAt;
+ } else {
+ boolean debug = ExtensibleLoadManagerImpl.debug(pulsar.getConfiguration(), log);
+ if (debug) {
+ log.info("Cleaned broker load data.");
+ }
+ }
+ }
+ );
+ }
+
+ @Override
+ public void handleEvent(String serviceUnit, ServiceUnitStateData data, Throwable t) {
+ if (t != null) {
+ return;
+ }
+ ServiceUnitState state = ServiceUnitStateData.state(data);
+ switch (state) {
+ case Releasing, Splitting -> {
+ if (StringUtils.equals(data.sourceBroker(), lookupServiceAddress)) {
+ tombstone();
+ }
+ }
+ case Owned -> {
+ if (StringUtils.equals(data.dstBroker(), lookupServiceAddress)) {
+ tombstone();
+ }
+ }
+ }
+ }
}
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/extensions/scheduler/NamespaceUnloadStrategy.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/extensions/scheduler/NamespaceUnloadStrategy.java
index 42af396abca..421e6240c8f 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/extensions/scheduler/NamespaceUnloadStrategy.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/extensions/scheduler/NamespaceUnloadStrategy.java
@@ -20,6 +20,7 @@ package org.apache.pulsar.broker.loadbalance.extensions.scheduler;
import java.util.Map;
import java.util.Set;
+import org.apache.pulsar.broker.PulsarService;
import org.apache.pulsar.broker.loadbalance.extensions.LoadManagerContext;
import org.apache.pulsar.broker.loadbalance.extensions.models.UnloadDecision;
@@ -43,4 +44,11 @@ public interface NamespaceUnloadStrategy {
Map<String, Long> recentlyUnloadedBundles,
Map<String, Long> recentlyUnloadedBrokers);
+ /**
+ * Initializes the internals.
+ *
+ * @param pulsar The pulsar service instance.
+ */
+ void initialize(PulsarService pulsar);
+
}
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/extensions/scheduler/SplitScheduler.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/extensions/scheduler/SplitScheduler.java
index 589df80fc5c..816fde0038a 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/extensions/scheduler/SplitScheduler.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/extensions/scheduler/SplitScheduler.java
@@ -22,6 +22,7 @@ import static org.apache.pulsar.broker.loadbalance.extensions.models.SplitDecisi
import java.util.ArrayList;
import java.util.List;
import java.util.Set;
+import java.util.StringJoiner;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledFuture;
@@ -30,6 +31,7 @@ import java.util.concurrent.atomic.AtomicReference;
import lombok.extern.slf4j.Slf4j;
import org.apache.pulsar.broker.PulsarService;
import org.apache.pulsar.broker.ServiceConfiguration;
+import org.apache.pulsar.broker.loadbalance.extensions.ExtensibleLoadManagerImpl;
import org.apache.pulsar.broker.loadbalance.extensions.LoadManagerContext;
import org.apache.pulsar.broker.loadbalance.extensions.channel.ServiceUnitStateChannel;
import org.apache.pulsar.broker.loadbalance.extensions.manager.SplitManager;
@@ -98,7 +100,7 @@ public class SplitScheduler implements LoadManagerScheduler {
@Override
public void execute() {
- boolean debugMode = conf.isLoadBalancerDebugModeEnabled() || log.isDebugEnabled();
+ boolean debugMode = ExtensibleLoadManagerImpl.debug(conf, log);
if (debugMode) {
log.info("Load balancer enabled: {}, Split enabled: {}.",
conf.isLoadBalancerEnabled(), conf.isLoadBalancerAutoBundleSplitEnabled());
@@ -113,8 +115,10 @@ public class SplitScheduler implements LoadManagerScheduler {
synchronized (bundleSplitStrategy) {
final Set<SplitDecision> decisions = bundleSplitStrategy.findBundlesToSplit(context, pulsar);
+ if (debugMode) {
+ log.info("Split Decisions:", decisions);
+ }
if (!decisions.isEmpty()) {
-
// currently following the unloading timeout
var asyncOpTimeoutMs = conf.getNamespaceBundleUnloadingTimeoutMs();
List<CompletableFuture<Void>> futures = new ArrayList<>();
@@ -156,6 +160,14 @@ public class SplitScheduler implements LoadManagerScheduler {
task = loadManagerExecutor.scheduleAtFixedRate(() -> {
try {
execute();
+ var debugMode = ExtensibleLoadManagerImpl.debug(conf, log);
+ if (debugMode) {
+ StringJoiner joiner = new StringJoiner("\n");
+ joiner.add("### OwnershipEntrySet start ###");
+ serviceUnitStateChannel.getOwnershipEntrySet().forEach(e -> joiner.add(e.toString()));
+ joiner.add("### OwnershipEntrySet end ###");
+ log.info(joiner.toString());
+ }
} catch (Throwable e) {
log.error("Failed to run the split job.", e);
}
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/extensions/scheduler/TransferShedder.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/extensions/scheduler/TransferShedder.java
index effddaf9c41..669219daba2 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/extensions/scheduler/TransferShedder.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/extensions/scheduler/TransferShedder.java
@@ -20,8 +20,8 @@ package org.apache.pulsar.broker.loadbalance.extensions.scheduler;
import static org.apache.pulsar.broker.loadbalance.extensions.models.UnloadDecision.Label.Failure;
import static org.apache.pulsar.broker.loadbalance.extensions.models.UnloadDecision.Label.Skip;
-import static org.apache.pulsar.broker.loadbalance.extensions.models.UnloadDecision.Reason.Balanced;
import static org.apache.pulsar.broker.loadbalance.extensions.models.UnloadDecision.Reason.CoolDown;
+import static org.apache.pulsar.broker.loadbalance.extensions.models.UnloadDecision.Reason.HitCount;
import static org.apache.pulsar.broker.loadbalance.extensions.models.UnloadDecision.Reason.NoBrokers;
import static org.apache.pulsar.broker.loadbalance.extensions.models.UnloadDecision.Reason.NoBundles;
import static org.apache.pulsar.broker.loadbalance.extensions.models.UnloadDecision.Reason.NoLoadData;
@@ -30,19 +30,26 @@ import static org.apache.pulsar.broker.loadbalance.extensions.models.UnloadDecis
import static org.apache.pulsar.broker.loadbalance.extensions.models.UnloadDecision.Reason.Underloaded;
import static org.apache.pulsar.broker.loadbalance.extensions.models.UnloadDecision.Reason.Unknown;
import com.google.common.annotations.VisibleForTesting;
-import com.google.common.collect.MinMaxPriorityQueue;
+import java.util.ArrayList;
+import java.util.Comparator;
import java.util.HashSet;
+import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.Set;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
+import java.util.stream.Collectors;
import lombok.Getter;
+import lombok.NoArgsConstructor;
import lombok.experimental.Accessors;
import org.apache.pulsar.broker.PulsarService;
import org.apache.pulsar.broker.ServiceConfiguration;
+import org.apache.pulsar.broker.loadbalance.extensions.ExtensibleLoadManagerImpl;
import org.apache.pulsar.broker.loadbalance.extensions.LoadManagerContext;
+import org.apache.pulsar.broker.loadbalance.extensions.channel.ServiceUnitStateChannel;
+import org.apache.pulsar.broker.loadbalance.extensions.channel.ServiceUnitStateChannelImpl;
import org.apache.pulsar.broker.loadbalance.extensions.data.BrokerLoadData;
import org.apache.pulsar.broker.loadbalance.extensions.data.BrokerLookupData;
import org.apache.pulsar.broker.loadbalance.extensions.data.TopBundlesLoadData;
@@ -80,17 +87,23 @@ import org.slf4j.LoggerFactory;
* (loadBalancerMaxNumberOfBrokerTransfersPerCycle).
* 9. Print more logs with a debug option(loadBalancerDebugModeEnabled=true).
*/
+@NoArgsConstructor
public class TransferShedder implements NamespaceUnloadStrategy {
private static final Logger log = LoggerFactory.getLogger(TransferShedder.class);
private static final double KB = 1024;
+ private static final String CANNOT_CONTINUE_UNLOAD_MSG = "Can't continue the unload cycle.";
+ private static final String CANNOT_UNLOAD_BROKER_MSG = "Can't unload broker:%s.";
+ private static final String CANNOT_UNLOAD_BUNDLE_MSG = "Can't unload bundle:%s.";
private final LoadStats stats = new LoadStats();
- private final PulsarService pulsar;
- private final SimpleResourceAllocationPolicies allocationPolicies;
- private final IsolationPoliciesHelper isolationPoliciesHelper;
- private final AntiAffinityGroupPolicyHelper antiAffinityGroupPolicyHelper;
-
- private final Set<UnloadDecision> decisionCache;
- private final UnloadCounter counter;
+ private PulsarService pulsar;
+ private SimpleResourceAllocationPolicies allocationPolicies;
+ private IsolationPoliciesHelper isolationPoliciesHelper;
+ private AntiAffinityGroupPolicyHelper antiAffinityGroupPolicyHelper;
+ private Set<UnloadDecision> decisionCache;
+ @Getter
+ private UnloadCounter counter;
+ private ServiceUnitStateChannel channel;
+ private int unloadConditionHitCount = 0;
@VisibleForTesting
public TransferShedder(UnloadCounter counter){
@@ -110,9 +123,22 @@ public class TransferShedder implements NamespaceUnloadStrategy {
this.allocationPolicies = new SimpleResourceAllocationPolicies(pulsar);
this.counter = counter;
this.isolationPoliciesHelper = new IsolationPoliciesHelper(allocationPolicies);
+ this.channel = ServiceUnitStateChannelImpl.get(pulsar);
this.antiAffinityGroupPolicyHelper = antiAffinityGroupPolicyHelper;
}
+ @Override
+ public void initialize(PulsarService pulsar){
+ this.pulsar = pulsar;
+ this.decisionCache = new HashSet<>();
+ this.allocationPolicies = new SimpleResourceAllocationPolicies(pulsar);
+ var manager = ExtensibleLoadManagerImpl.get(pulsar.getLoadManager().get());
+ this.counter = manager.getUnloadCounter();
+ this.isolationPoliciesHelper = new IsolationPoliciesHelper(allocationPolicies);
+ this.channel = ServiceUnitStateChannelImpl.get(pulsar);
+ this.antiAffinityGroupPolicyHelper = manager.getAntiAffinityGroupPolicyHelper();
+ }
+
@Getter
@Accessors(fluent = true)
@@ -122,17 +148,14 @@ public class TransferShedder implements NamespaceUnloadStrategy {
private int totalBrokers;
private double avg;
private double std;
- private MinMaxPriorityQueue<String> minBrokers;
- private MinMaxPriorityQueue<String> maxBrokers;
private LoadDataStore<BrokerLoadData> loadDataStore;
-
+ private List<Map.Entry<String, BrokerLoadData>> brokersSortedByLoad;
+ int maxBrokerIndex;
+ int minBrokerIndex;
+ int numberOfBrokerSheddingPerCycle;
+ int maxNumberOfBrokerSheddingPerCycle;
LoadStats() {
- this.minBrokers = MinMaxPriorityQueue.orderedBy((a, b) -> Double.compare(
- loadDataStore.get((String) b).get().getWeightedMaxEMA(),
- loadDataStore.get((String) a).get().getWeightedMaxEMA())).create();
- this.maxBrokers = MinMaxPriorityQueue.orderedBy((a, b) -> Double.compare(
- loadDataStore.get((String) a).get().getWeightedMaxEMA(),
- loadDataStore.get((String) b).get().getWeightedMaxEMA())).create();
+ brokersSortedByLoad = new ArrayList<>();
}
private void update(double sum, double sqSum, int totalBrokers) {
@@ -143,8 +166,7 @@ public class TransferShedder implements NamespaceUnloadStrategy {
if (totalBrokers == 0) {
this.avg = 0;
this.std = 0;
- minBrokers.clear();
- maxBrokers.clear();
+
} else {
this.avg = sum / totalBrokers;
this.std = Math.sqrt(sqSum / totalBrokers - avg * avg);
@@ -156,34 +178,42 @@ public class TransferShedder implements NamespaceUnloadStrategy {
double maxd = Math.max(0, max - offload);
double mind = min + offload;
sqSum += maxd * maxd + mind * mind;
- std = Math.sqrt(sqSum / totalBrokers - avg * avg);
+ std = Math.sqrt(Math.abs(sqSum / totalBrokers - avg * avg));
+ numberOfBrokerSheddingPerCycle++;
+ minBrokerIndex++;
}
- void clear(){
+ void clear() {
sum = 0.0;
sqSum = 0.0;
totalBrokers = 0;
avg = 0.0;
std = 0.0;
- minBrokers.clear();
- maxBrokers.clear();
+ maxBrokerIndex = 0;
+ minBrokerIndex = 0;
+ numberOfBrokerSheddingPerCycle = 0;
+ maxNumberOfBrokerSheddingPerCycle = 0;
+ brokersSortedByLoad.clear();
+ loadDataStore = null;
}
Optional<UnloadDecision.Reason> update(final LoadDataStore<BrokerLoadData> loadStore,
+ final Map<String, BrokerLookupData> availableBrokers,
Map<String, Long> recentlyUnloadedBrokers,
final ServiceConfiguration conf) {
-
+ maxNumberOfBrokerSheddingPerCycle = conf.getLoadBalancerMaxNumberOfBrokerSheddingPerCycle();
+ var debug = ExtensibleLoadManagerImpl.debug(conf, log);
UnloadDecision.Reason decisionReason = null;
double sum = 0.0;
double sqSum = 0.0;
int totalBrokers = 0;
- int maxTransfers = conf.getLoadBalancerMaxNumberOfBrokerTransfersPerCycle();
long now = System.currentTimeMillis();
+ var missingLoadDataBrokers = new HashSet<>(availableBrokers.keySet());
for (Map.Entry<String, BrokerLoadData> entry : loadStore.entrySet()) {
BrokerLoadData localBrokerData = entry.getValue();
String broker = entry.getKey();
-
+ missingLoadDataBrokers.remove(broker);
// We don't want to use the outdated load data.
if (now - localBrokerData.getUpdatedAt()
> conf.getLoadBalancerBrokerLoadDataTTLInSeconds() * 1000) {
@@ -196,12 +226,16 @@ public class TransferShedder implements NamespaceUnloadStrategy {
// Also, we should give enough time for each broker to recompute its load after transfers.
if (recentlyUnloadedBrokers.containsKey(broker)) {
- if (localBrokerData.getUpdatedAt() - recentlyUnloadedBrokers.get(broker)
- < conf.getLoadBalanceUnloadDelayInSeconds() * 1000) {
- log.warn(
- "Broker:{} load data timestamp:{} is too early since "
- + "the last transfer timestamp:{}. Stop unloading.",
- broker, localBrokerData.getUpdatedAt(), recentlyUnloadedBrokers.get(broker));
+ var elapsed = localBrokerData.getUpdatedAt() - recentlyUnloadedBrokers.get(broker);
+ if (elapsed < conf.getLoadBalanceSheddingDelayInSeconds() * 1000) {
+ if (debug) {
+ log.warn(
+ "Broker:{} load data is too early since "
+ + "the last transfer. elapsed {} secs < threshold {} secs",
+ broker,
+ TimeUnit.MILLISECONDS.toSeconds(elapsed),
+ conf.getLoadBalanceSheddingDelayInSeconds());
+ }
update(0.0, 0.0, 0);
return Optional.of(CoolDown);
} else {
@@ -211,25 +245,28 @@ public class TransferShedder implements NamespaceUnloadStrategy {
double load = localBrokerData.getWeightedMaxEMA();
- minBrokers.offer(broker);
- if (minBrokers.size() > maxTransfers) {
- minBrokers.poll();
- }
- maxBrokers.offer(broker);
- if (maxBrokers.size() > maxTransfers) {
- maxBrokers.poll();
- }
sum += load;
sqSum += load * load;
totalBrokers++;
}
-
if (totalBrokers == 0) {
if (decisionReason == null) {
decisionReason = NoBrokers;
}
update(0.0, 0.0, 0);
+ if (debug) {
+ log.info("There is no broker load data.");
+ }
+ return Optional.of(decisionReason);
+ }
+
+ if (!missingLoadDataBrokers.isEmpty()) {
+ decisionReason = NoLoadData;
+ update(0.0, 0.0, 0);
+ if (debug) {
+ log.info("There is missing load data from brokers:{}", missingLoadDataBrokers);
+ }
return Optional.of(decisionReason);
}
@@ -237,21 +274,35 @@ public class TransferShedder implements NamespaceUnloadStrategy {
return Optional.empty();
}
- boolean hasTransferableBrokers() {
- return !(maxBrokers.isEmpty() || minBrokers.isEmpty()
- || maxBrokers.peekLast().equals(minBrokers().peekLast()));
- }
-
void setLoadDataStore(LoadDataStore<BrokerLoadData> loadDataStore) {
this.loadDataStore = loadDataStore;
+ brokersSortedByLoad.addAll(loadDataStore.entrySet());
+ brokersSortedByLoad.sort(Comparator.comparingDouble(
+ a -> a.getValue().getWeightedMaxEMA()));
+ maxBrokerIndex = brokersSortedByLoad.size() - 1;
+ minBrokerIndex = 0;
+ }
+
+ String peekMinBroker() {
+ return brokersSortedByLoad.get(minBrokerIndex).getKey();
+ }
+
+ String pollMaxBroker() {
+ return brokersSortedByLoad.get(maxBrokerIndex--).getKey();
}
@Override
public String toString() {
return String.format(
- "sum:%.2f, sqSum:%.2f, avg:%.2f, std:%.2f, totalBrokers:%d, "
- + "minBrokers:%s, maxBrokers:%s",
- sum, sqSum, avg, std, totalBrokers, minBrokers, maxBrokers);
+ "sum:%.2f, sqSum:%.2f, avg:%.2f, std:%.2f, totalBrokers:%d, brokersSortedByLoad:%s",
+ sum, sqSum, avg, std, totalBrokers,
+ brokersSortedByLoad.stream().map(v->v.getKey()).collect(Collectors.toList()));
+ }
+
+
+ boolean hasTransferableBrokers() {
+ return numberOfBrokerSheddingPerCycle < maxNumberOfBrokerSheddingPerCycle
+ && minBrokerIndex < maxBrokerIndex;
}
}
@@ -263,20 +314,36 @@ public class TransferShedder implements NamespaceUnloadStrategy {
final var conf = context.brokerConfiguration();
decisionCache.clear();
stats.clear();
+ Map<String, BrokerLookupData> availableBrokers;
+ try {
+ availableBrokers = context.brokerRegistry().getAvailableBrokerLookupDataAsync()
+ .get(context.brokerConfiguration().getMetadataStoreOperationTimeoutSeconds(), TimeUnit.SECONDS);
+ } catch (ExecutionException | InterruptedException | TimeoutException e) {
+ counter.update(Failure, Unknown);
+ log.warn("Failed to fetch available brokers. Stop unloading.", e);
+ return decisionCache;
+ }
try {
final var loadStore = context.brokerLoadDataStore();
stats.setLoadDataStore(loadStore);
- boolean debugMode = conf.isLoadBalancerDebugModeEnabled() || log.isDebugEnabled();
+ boolean debugMode = ExtensibleLoadManagerImpl.debug(conf, log);
- var skipReason = stats.update(context.brokerLoadDataStore(), recentlyUnloadedBrokers, conf);
+ var skipReason = stats.update(
+ context.brokerLoadDataStore(), availableBrokers, recentlyUnloadedBrokers, conf);
if (skipReason.isPresent()) {
- log.warn("Failed to update load stat. Reason:{}. Stop unloading.", skipReason.get());
+ if (debugMode) {
+ log.warn(CANNOT_CONTINUE_UNLOAD_MSG
+ + " Skipped the load stat update. Reason:{}.",
+ skipReason.get());
+ }
counter.update(Skip, skipReason.get());
return decisionCache;
}
counter.updateLoadData(stats.avg, stats.std);
+
+
if (debugMode) {
log.info("brokers' load stats:{}", stats);
}
@@ -287,138 +354,282 @@ public class TransferShedder implements NamespaceUnloadStrategy {
final double targetStd = conf.getLoadBalancerBrokerLoadTargetStd();
boolean transfer = conf.isLoadBalancerTransferEnabled();
+ if (stats.std() > targetStd || isUnderLoaded(context, stats.peekMinBroker(), stats.avg)) {
+ unloadConditionHitCount++;
+ } else {
+ unloadConditionHitCount = 0;
+ }
- Map<String, BrokerLookupData> availableBrokers;
- try {
- availableBrokers = context.brokerRegistry().getAvailableBrokerLookupDataAsync()
- .get(context.brokerConfiguration().getMetadataStoreOperationTimeoutSeconds(), TimeUnit.SECONDS);
- } catch (ExecutionException | InterruptedException | TimeoutException e) {
- counter.update(Skip, Unknown);
- log.warn("Failed to fetch available brokers. Reason: Unknown. Stop unloading.", e);
+ if (unloadConditionHitCount <= conf.getLoadBalancerSheddingConditionHitCountThreshold()) {
+ if (debugMode) {
+ log.info(CANNOT_CONTINUE_UNLOAD_MSG
+ + " Shedding condition hit count:{} is less than or equal to the threshold:{}.",
+ unloadConditionHitCount, conf.getLoadBalancerSheddingConditionHitCountThreshold());
+ }
+ counter.update(Skip, HitCount);
return decisionCache;
}
while (true) {
if (!stats.hasTransferableBrokers()) {
if (debugMode) {
- log.info("Exhausted target transfer brokers. Stop unloading");
+ log.info(CANNOT_CONTINUE_UNLOAD_MSG
+ + " Exhausted target transfer brokers.");
}
break;
}
UnloadDecision.Reason reason;
if (stats.std() <= targetStd) {
- if (hasMsgThroughput(context, stats.minBrokers.peekLast())) {
+ if (!isUnderLoaded(context, stats.peekMinBroker(), stats.avg)) {
if (debugMode) {
- log.info("std:{} <= targetStd:{} and minBroker:{} has msg throughput. Stop unloading.",
- stats.std, targetStd, stats.minBrokers.peekLast());
+ log.info(CANNOT_CONTINUE_UNLOAD_MSG
+ + "The overall cluster load meets the target, std:{} <= targetStd:{},"
+ + " and minBroker:{} is not underloaded.",
+ stats.std(), targetStd, stats.peekMinBroker());
}
break;
} else {
reason = Underloaded;
+ if (debugMode) {
+ log.info(String.format("broker:%s is underloaded:%s although "
+ + "load std:%.2f <= targetStd:%.2f. "
+ + "Continuing unload for this underloaded broker.",
+ stats.peekMinBroker(),
+ context.brokerLoadDataStore().get(stats.peekMinBroker()).get(),
+ stats.std(), targetStd));
+ }
}
} else {
reason = Overloaded;
}
- String maxBroker = stats.maxBrokers().pollLast();
- String minBroker = stats.minBrokers().pollLast();
+ String maxBroker = stats.pollMaxBroker();
+ String minBroker = stats.peekMinBroker();
Optional<BrokerLoadData> maxBrokerLoadData = context.brokerLoadDataStore().get(maxBroker);
Optional<BrokerLoadData> minBrokerLoadData = context.brokerLoadDataStore().get(minBroker);
if (maxBrokerLoadData.isEmpty()) {
- log.error("maxBroker:{} maxBrokerLoadData is empty. Skip unloading from this max broker.",
- maxBroker);
+ log.error(String.format(CANNOT_UNLOAD_BROKER_MSG
+ + " MaxBrokerLoadData is empty.", maxBroker));
numOfBrokersWithEmptyLoadData++;
continue;
}
if (minBrokerLoadData.isEmpty()) {
- log.error("minBroker:{} minBrokerLoadData is empty. Skip unloading to this min broker.", minBroker);
+ log.error("Can't transfer load to broker:{}. MinBrokerLoadData is empty.", minBroker);
numOfBrokersWithEmptyLoadData++;
continue;
}
-
- double max = maxBrokerLoadData.get().getWeightedMaxEMA();
- double min = minBrokerLoadData.get().getWeightedMaxEMA();
- double offload = (max - min) / 2;
+ double maxLoad = maxBrokerLoadData.get().getWeightedMaxEMA();
+ double minLoad = minBrokerLoadData.get().getWeightedMaxEMA();
+ double offload = (maxLoad - minLoad) / 2;
BrokerLoadData brokerLoadData = maxBrokerLoadData.get();
- double brokerThroughput = brokerLoadData.getMsgThroughputIn() + brokerLoadData.getMsgThroughputOut();
- double offloadThroughput = brokerThroughput * offload;
+ double maxBrokerThroughput = brokerLoadData.getMsgThroughputIn()
+ + brokerLoadData.getMsgThroughputOut();
+ double minBrokerThroughput = minBrokerLoadData.get().getMsgThroughputIn()
+ + minBrokerLoadData.get().getMsgThroughputOut();
+ double offloadThroughput = maxBrokerThroughput * offload / maxLoad;
if (debugMode) {
- log.info(
- "Attempting to shed load from broker:{}{}, which has the max resource "
- + "usage {}%, targetStd:{},"
- + " -- Offloading {}%, at least {} KByte/s of traffic, left throughput {} KByte/s",
+ log.info(String.format(
+ "Attempting to shed load from broker:%s%s, which has the max resource "
+ + "usage:%.2f%%, targetStd:%.2f,"
+ + " -- Trying to offload %.2f%%, %.2f KByte/s of traffic.",
maxBroker, transfer ? " to broker:" + minBroker : "",
- 100 * max, targetStd,
- offload * 100, offloadThroughput / KB, (brokerThroughput - offloadThroughput) / KB);
+ maxLoad * 100,
+ targetStd,
+ offload * 100,
+ offloadThroughput / KB
+ ));
}
double trafficMarkedToOffload = 0;
- boolean atLeastOneBundleSelected = false;
+ double trafficMarkedToGain = 0;
Optional<TopBundlesLoadData> bundlesLoadData = context.topBundleLoadDataStore().get(maxBroker);
if (bundlesLoadData.isEmpty() || bundlesLoadData.get().getTopBundlesLoadData().isEmpty()) {
- log.error("maxBroker:{} topBundlesLoadData is empty. Skip unloading from this broker.", maxBroker);
+ log.error(String.format(CANNOT_UNLOAD_BROKER_MSG
+ + " TopBundlesLoadData is empty.", maxBroker));
numOfBrokersWithEmptyLoadData++;
continue;
}
- var topBundlesLoadData = bundlesLoadData.get().getTopBundlesLoadData();
- if (topBundlesLoadData.size() > 1) {
- int remainingTopBundles = topBundlesLoadData.size();
- for (var e : topBundlesLoadData) {
- String bundle = e.bundleName();
- if (!recentlyUnloadedBundles.containsKey(bundle)
- && isTransferable(context, availableBrokers,
- bundle, maxBroker, Optional.of(minBroker))) {
- var bundleData = e.stats();
- double throughput = bundleData.msgThroughputIn + bundleData.msgThroughputOut;
- if (remainingTopBundles > 1
- && (trafficMarkedToOffload < offloadThroughput
- || !atLeastOneBundleSelected)) {
- Unload unload;
- if (transfer) {
- unload = new Unload(maxBroker, bundle, Optional.of(minBroker));
- } else {
- unload = new Unload(maxBroker, bundle);
+ var maxBrokerTopBundlesLoadData = bundlesLoadData.get().getTopBundlesLoadData();
+ if (maxBrokerTopBundlesLoadData.size() == 1) {
+ numOfBrokersWithFewBundles++;
+ log.warn(String.format(CANNOT_UNLOAD_BROKER_MSG
+ + " Sole namespace bundle:%s is overloading the broker. ",
+ maxBroker, maxBrokerTopBundlesLoadData.iterator().next()));
+ continue;
+ }
+ Optional<TopBundlesLoadData> minBundlesLoadData = context.topBundleLoadDataStore().get(minBroker);
+ var minBrokerTopBundlesLoadDataIter =
+ minBundlesLoadData.isPresent() ? minBundlesLoadData.get().getTopBundlesLoadData().iterator() :
+ null;
+
+
+ if (maxBrokerTopBundlesLoadData.isEmpty()) {
+ numOfBrokersWithFewBundles++;
+ log.warn(String.format(CANNOT_UNLOAD_BROKER_MSG
+ + " Broker overloaded despite having no bundles", maxBroker));
+ continue;
+ }
+
+ int remainingTopBundles = maxBrokerTopBundlesLoadData.size();
+ for (var e : maxBrokerTopBundlesLoadData) {
+ String bundle = e.bundleName();
+ if (channel != null && !channel.isOwner(bundle, maxBroker)) {
+ if (debugMode) {
+ log.warn(String.format(CANNOT_UNLOAD_BUNDLE_MSG
+ + " MaxBroker:%s is not the owner.", bundle, maxBroker));
+ }
+ continue;
+ }
+ if (recentlyUnloadedBundles.containsKey(bundle)) {
+ if (debugMode) {
+ log.info(String.format(CANNOT_UNLOAD_BUNDLE_MSG
+ + " Bundle has been recently unloaded at ts:%d.",
+ bundle, recentlyUnloadedBundles.get(bundle)));
+ }
+ continue;
+ }
+ if (!isTransferable(context, availableBrokers, bundle, maxBroker, Optional.of(minBroker))) {
+ if (debugMode) {
+ log.info(String.format(CANNOT_UNLOAD_BUNDLE_MSG
+ + " This unload can't meet "
+ + "affinity(isolation) or anti-affinity group policies.", bundle));
+ }
+ continue;
+ }
+ if (remainingTopBundles <= 1) {
+ if (debugMode) {
+ log.info(String.format(CANNOT_UNLOAD_BUNDLE_MSG
+ + " The remaining bundles in TopBundlesLoadData from the maxBroker:%s is"
+ + " less than or equal to 1.",
+ bundle, maxBroker));
+ }
+ break;
+ }
+
+ var bundleData = e.stats();
+ double maxBrokerBundleThroughput = bundleData.msgThroughputIn + bundleData.msgThroughputOut;
+ boolean swap = false;
+ List<Unload> minToMaxUnloads = new ArrayList<>();
+ double minBrokerBundleSwapThroughput = 0.0;
+ if (trafficMarkedToOffload - trafficMarkedToGain + maxBrokerBundleThroughput > offloadThroughput) {
+ // see if we can swap bundles from min to max broker to balance better.
+ if (transfer && minBrokerTopBundlesLoadDataIter != null) {
+ var maxBrokerNewThroughput =
+ maxBrokerThroughput - trafficMarkedToOffload + trafficMarkedToGain
+ - maxBrokerBundleThroughput;
+ var minBrokerNewThroughput =
+ minBrokerThroughput + trafficMarkedToOffload - trafficMarkedToGain
+ + maxBrokerBundleThroughput;
+ while (minBrokerTopBundlesLoadDataIter.hasNext()) {
+ var minBrokerBundleData = minBrokerTopBundlesLoadDataIter.next();
+ if (!isTransferable(context, availableBrokers,
+ minBrokerBundleData.bundleName(), minBroker, Optional.of(maxBroker))) {
+ continue;
+ }
+ var minBrokerBundleThroughput =
+ minBrokerBundleData.stats().msgThroughputIn
+ + minBrokerBundleData.stats().msgThroughputOut;
+ var maxBrokerNewThroughputTmp = maxBrokerNewThroughput + minBrokerBundleThroughput;
+ var minBrokerNewThroughputTmp = minBrokerNewThroughput - minBrokerBundleThroughput;
+ if (maxBrokerNewThroughputTmp < maxBrokerThroughput
+ && minBrokerNewThroughputTmp < maxBrokerThroughput) {
+ minToMaxUnloads.add(new Unload(minBroker,
+ minBrokerBundleData.bundleName(), Optional.of(maxBroker)));
+ maxBrokerNewThroughput = maxBrokerNewThroughputTmp;
+ minBrokerNewThroughput = minBrokerNewThroughputTmp;
+ minBrokerBundleSwapThroughput += minBrokerBundleThroughput;
+ if (minBrokerNewThroughput <= maxBrokerNewThroughput
+ && maxBrokerNewThroughput < maxBrokerThroughput * 0.75) {
+ swap = true;
+ break;
+ }
+ }
+ }
+ }
+ if (!swap) {
+ if (debugMode) {
+ log.info(String.format(CANNOT_UNLOAD_BUNDLE_MSG
+ + " The traffic to unload:%.2f - gain:%.2f = %.2f KByte/s is "
+ + "greater than the target :%.2f KByte/s.",
+ bundle,
+ (trafficMarkedToOffload + maxBrokerBundleThroughput) / KB,
+ trafficMarkedToGain / KB,
+ (trafficMarkedToOffload - trafficMarkedToGain + maxBrokerBundleThroughput) / KB,
+ offloadThroughput / KB));
+ }
+ break;
+ }
+ }
+ Unload unload;
+ if (transfer) {
+ if (swap) {
+ minToMaxUnloads.forEach(minToMaxUnload -> {
+ if (debugMode) {
+ log.info("Decided to gain bundle:{} from min broker:{}",
+ minToMaxUnload.serviceUnit(), minToMaxUnload.sourceBroker());
}
var decision = new UnloadDecision();
- decision.setUnload(unload);
+ decision.setUnload(minToMaxUnload);
decision.succeed(reason);
decisionCache.add(decision);
- trafficMarkedToOffload += throughput;
- atLeastOneBundleSelected = true;
- remainingTopBundles--;
+ });
+ if (debugMode) {
+ log.info(String.format(
+ "Total traffic %.2f KByte/s to transfer from min broker:%s to max broker:%s.",
+ minBrokerBundleSwapThroughput / KB, minBroker, maxBroker));
+ trafficMarkedToGain += minBrokerBundleSwapThroughput;
}
}
+ unload = new Unload(maxBroker, bundle, Optional.of(minBroker));
+ } else {
+ unload = new Unload(maxBroker, bundle);
}
- if (!atLeastOneBundleSelected) {
- numOfBrokersWithFewBundles++;
+ var decision = new UnloadDecision();
+ decision.setUnload(unload);
+ decision.succeed(reason);
+ decisionCache.add(decision);
+ trafficMarkedToOffload += maxBrokerBundleThroughput;
+ remainingTopBundles--;
+
+ if (debugMode) {
+ log.info(String.format("Decided to unload bundle:%s, throughput:%.2f KByte/s."
+ + " The traffic marked to unload:%.2f - gain:%.2f = %.2f KByte/s."
+ + " Target:%.2f KByte/s.",
+ bundle, maxBrokerBundleThroughput / KB,
+ trafficMarkedToOffload / KB,
+ trafficMarkedToGain / KB,
+ (trafficMarkedToOffload - trafficMarkedToGain) / KB,
+ offloadThroughput / KB));
}
- } else if (topBundlesLoadData.size() == 1) {
- numOfBrokersWithFewBundles++;
- log.warn(
- "HIGH USAGE WARNING : Sole namespace bundle {} is overloading broker {}. "
- + "No Load Shedding will be done on this broker",
- topBundlesLoadData.iterator().next(), maxBroker);
- } else {
- numOfBrokersWithFewBundles++;
- log.warn("Broker {} is overloaded despite having no bundles", maxBroker);
}
-
if (trafficMarkedToOffload > 0) {
- stats.offload(max, min, offload);
+ var adjustedOffload =
+ (trafficMarkedToOffload - trafficMarkedToGain) * maxLoad / maxBrokerThroughput;
+ stats.offload(maxLoad, minLoad, adjustedOffload);
if (debugMode) {
log.info(
String.format("brokers' load stats:%s, after offload{max:%.2f, min:%.2f, offload:%.2f}",
- stats, max, min, offload));
+ stats, maxLoad, minLoad, adjustedOffload));
}
+ } else {
+ numOfBrokersWithFewBundles++;
+ log.warn(String.format(CANNOT_UNLOAD_BROKER_MSG
+ + " There is no bundle that can be unloaded in top bundles load data. "
+ + "Consider splitting bundles owned by the broker "
+ + "to make each bundle serve less traffic "
+ + "or increasing loadBalancerMaxNumberOfBundlesInBundleLoadReport"
+ + " to report more bundles in the top bundles load data.", maxBroker));
}
- }
+
+ } // while end
if (debugMode) {
log.info("decisionCache:{}", decisionCache);
}
+
if (decisionCache.isEmpty()) {
UnloadDecision.Reason reason;
if (numOfBrokersWithEmptyLoadData > 0) {
@@ -426,10 +637,13 @@ public class TransferShedder implements NamespaceUnloadStrategy {
} else if (numOfBrokersWithFewBundles > 0) {
reason = NoBundles;
} else {
- reason = Balanced;
+ reason = HitCount;
}
counter.update(Skip, reason);
+ } else {
+ unloadConditionHitCount = 0;
}
+
} catch (Throwable e) {
log.error("Failed to process unloading. ", e);
this.counter.update(Failure, Unknown);
@@ -438,13 +652,19 @@ public class TransferShedder implements NamespaceUnloadStrategy {
}
- private boolean hasMsgThroughput(LoadManagerContext context, String broker) {
+ private boolean isUnderLoaded(LoadManagerContext context, String broker, double avgLoad) {
var brokerLoadDataOptional = context.brokerLoadDataStore().get(broker);
if (brokerLoadDataOptional.isEmpty()) {
return false;
}
var brokerLoadData = brokerLoadDataOptional.get();
- return brokerLoadData.getMsgThroughputIn() + brokerLoadData.getMsgThroughputOut() > 0.0;
+ if (brokerLoadData.getMsgThroughputEMA() < 1) {
+ return true;
+ }
+
+ return brokerLoadData.getWeightedMaxEMA()
+ < avgLoad * Math.min(0.5, Math.max(0.0,
+ context.brokerConfiguration().getLoadBalancerBrokerLoadTargetStd() / 2));
}
@@ -467,7 +687,8 @@ public class TransferShedder implements NamespaceUnloadStrategy {
return false;
}
- if (!antiAffinityGroupPolicyHelper.canUnload(availableBrokers, bundle, srcBroker, dstBroker)) {
+ if (antiAffinityGroupPolicyHelper != null
+ && !antiAffinityGroupPolicyHelper.canUnload(availableBrokers, bundle, srcBroker, dstBroker)) {
return false;
}
return true;
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/extensions/scheduler/UnloadScheduler.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/extensions/scheduler/UnloadScheduler.java
index 31310c5c9cc..d6c754c90fc 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/extensions/scheduler/UnloadScheduler.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/extensions/scheduler/UnloadScheduler.java
@@ -40,7 +40,6 @@ import org.apache.pulsar.broker.loadbalance.extensions.manager.UnloadManager;
import org.apache.pulsar.broker.loadbalance.extensions.models.Unload;
import org.apache.pulsar.broker.loadbalance.extensions.models.UnloadCounter;
import org.apache.pulsar.broker.loadbalance.extensions.models.UnloadDecision;
-import org.apache.pulsar.broker.loadbalance.extensions.policies.AntiAffinityGroupPolicyHelper;
import org.apache.pulsar.common.stats.Metrics;
import org.apache.pulsar.common.util.FutureUtil;
import org.apache.pulsar.common.util.Reflections;
@@ -81,11 +80,10 @@ public class UnloadScheduler implements LoadManagerScheduler {
UnloadManager unloadManager,
LoadManagerContext context,
ServiceUnitStateChannel channel,
- AntiAffinityGroupPolicyHelper antiAffinityGroupPolicyHelper,
UnloadCounter counter,
AtomicReference<List<Metrics>> unloadMetrics) {
this(pulsar, loadManagerExecutor, unloadManager, context, channel,
- createNamespaceUnloadStrategy(pulsar, antiAffinityGroupPolicyHelper, counter), counter, unloadMetrics);
+ createNamespaceUnloadStrategy(pulsar), counter, unloadMetrics);
}
@VisibleForTesting
@@ -212,19 +210,22 @@ public class UnloadScheduler implements LoadManagerScheduler {
this.recentlyUnloadedBrokers.clear();
}
- private static NamespaceUnloadStrategy createNamespaceUnloadStrategy(PulsarService pulsar,
- AntiAffinityGroupPolicyHelper helper,
- UnloadCounter counter) {
+ private static NamespaceUnloadStrategy createNamespaceUnloadStrategy(PulsarService pulsar) {
ServiceConfiguration conf = pulsar.getConfiguration();
+ NamespaceUnloadStrategy unloadStrategy;
try {
- return Reflections.createInstance(conf.getLoadBalancerLoadSheddingStrategy(), NamespaceUnloadStrategy.class,
+ unloadStrategy = Reflections.createInstance(conf.getLoadBalancerLoadSheddingStrategy(),
+ NamespaceUnloadStrategy.class,
Thread.currentThread().getContextClassLoader());
+ log.info("Created namespace unload strategy:{}", unloadStrategy.getClass().getCanonicalName());
} catch (Exception e) {
log.error("Error when trying to create namespace unload strategy: {}",
conf.getLoadBalancerLoadPlacementStrategy(), e);
+ log.error("create namespace unload strategy failed. using TransferShedder instead.");
+ unloadStrategy = new TransferShedder();
}
- log.error("create namespace unload strategy failed. using TransferShedder instead.");
- return new TransferShedder(pulsar, counter, helper);
+ unloadStrategy.initialize(pulsar);
+ return unloadStrategy;
}
private boolean isLoadBalancerSheddingEnabled() {
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/extensions/strategy/DefaultNamespaceBundleSplitStrategyImpl.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/extensions/strategy/DefaultNamespaceBundleSplitStrategyImpl.java
index da04a287f1a..15bfdc747f1 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/extensions/strategy/DefaultNamespaceBundleSplitStrategyImpl.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/extensions/strategy/DefaultNamespaceBundleSplitStrategyImpl.java
@@ -32,6 +32,7 @@ import lombok.extern.slf4j.Slf4j;
import org.apache.pulsar.broker.PulsarService;
import org.apache.pulsar.broker.ServiceConfiguration;
import org.apache.pulsar.broker.loadbalance.extensions.LoadManagerContext;
+import org.apache.pulsar.broker.loadbalance.extensions.channel.ServiceUnitStateChannelImpl;
import org.apache.pulsar.broker.loadbalance.extensions.models.Split;
import org.apache.pulsar.broker.loadbalance.extensions.models.SplitCounter;
import org.apache.pulsar.broker.loadbalance.extensions.models.SplitDecision;
@@ -47,15 +48,17 @@ import org.apache.pulsar.policies.data.loadbalancer.NamespaceBundleStats;
*/
@Slf4j
public class DefaultNamespaceBundleSplitStrategyImpl implements NamespaceBundleSplitStrategy {
+ private static final String CANNOT_CONTINUE_SPLIT_MSG = "Can't continue the split cycle.";
+ private static final String CANNOT_SPLIT_BUNDLE_MSG = "Can't split broker:%s.";
private final Set<SplitDecision> decisionCache;
private final Map<String, Integer> namespaceBundleCount;
- private final Map<String, Integer> bundleHighTrafficFrequency;
+ private final Map<String, Integer> splitConditionHitCounts;
private final SplitCounter counter;
public DefaultNamespaceBundleSplitStrategyImpl(SplitCounter counter) {
decisionCache = new HashSet<>();
namespaceBundleCount = new HashMap<>();
- bundleHighTrafficFrequency = new HashMap<>();
+ splitConditionHitCounts = new HashMap<>();
this.counter = counter;
}
@@ -71,22 +74,33 @@ public class DefaultNamespaceBundleSplitStrategyImpl implements NamespaceBundleS
long maxBundleMsgRate = conf.getLoadBalancerNamespaceBundleMaxMsgRate();
long maxBundleBandwidth = conf.getLoadBalancerNamespaceBundleMaxBandwidthMbytes() * LoadManagerShared.MIBI;
long maxSplitCount = conf.getLoadBalancerMaxNumberOfBundlesToSplitPerCycle();
- long splitConditionThreshold = conf.getLoadBalancerNamespaceBundleSplitConditionThreshold();
+ long splitConditionHitCountThreshold = conf.getLoadBalancerNamespaceBundleSplitConditionHitCountThreshold();
boolean debug = log.isDebugEnabled() || conf.isLoadBalancerDebugModeEnabled();
+ var channel = ServiceUnitStateChannelImpl.get(pulsar);
Map<String, NamespaceBundleStats> bundleStatsMap = pulsar.getBrokerService().getBundleStats();
NamespaceBundleFactory namespaceBundleFactory =
pulsar.getNamespaceService().getNamespaceBundleFactory();
- // clean bundleHighTrafficFrequency
- bundleHighTrafficFrequency.keySet().retainAll(bundleStatsMap.keySet());
+ // clean splitConditionHitCounts
+ splitConditionHitCounts.keySet().retainAll(bundleStatsMap.keySet());
for (var entry : bundleStatsMap.entrySet()) {
final String bundle = entry.getKey();
final NamespaceBundleStats stats = entry.getValue();
if (stats.topics < 2) {
if (debug) {
- log.info("The count of topics on the bundle {} is less than 2, skip split!", bundle);
+ log.info(String.format(CANNOT_SPLIT_BUNDLE_MSG
+ + " The topic count is less than 2.", bundle));
+ }
+ continue;
+ }
+
+ if (!channel.isOwner(bundle)) {
+ if (debug) {
+ log.error(String.format(CANNOT_SPLIT_BUNDLE_MSG
+ + " This broker is not the owner.", bundle));
+ counter.update(Failure, Unknown);
}
continue;
}
@@ -96,7 +110,8 @@ public class DefaultNamespaceBundleSplitStrategyImpl implements NamespaceBundleS
if (!namespaceBundleFactory
.canSplitBundle(namespaceBundleFactory.getBundle(namespaceName, bundleRange))) {
if (debug) {
- log.info("Can't split the bundle:{}. invalid bundle range:{}. ", bundle, bundleRange);
+ log.info(String.format(CANNOT_SPLIT_BUNDLE_MSG
+ + " Invalid bundle range:%s.", bundle, bundleRange));
}
counter.update(Failure, Unknown);
continue;
@@ -117,54 +132,85 @@ public class DefaultNamespaceBundleSplitStrategyImpl implements NamespaceBundleS
}
if (reason != Unknown) {
- bundleHighTrafficFrequency.put(bundle, bundleHighTrafficFrequency.getOrDefault(bundle, 0) + 1);
+ splitConditionHitCounts.put(bundle, splitConditionHitCounts.getOrDefault(bundle, 0) + 1);
} else {
- bundleHighTrafficFrequency.remove(bundle);
+ splitConditionHitCounts.remove(bundle);
+ }
+
+ if (splitConditionHitCounts.getOrDefault(bundle, 0) <= splitConditionHitCountThreshold) {
+ if (debug) {
+ log.info(String.format(
+ CANNOT_SPLIT_BUNDLE_MSG
+ + " Split condition hit count: %d is"
+ + " less than or equal to threshold: %d. "
+ + "Topics: %d/%d, "
+ + "Sessions: (%d+%d)/%d, "
+ + "Message Rate: %.2f/%d (msgs/s), "
+ + "Message Throughput: %.2f/%d (MB/s).",
+ bundle,
+ splitConditionHitCounts.getOrDefault(bundle, 0),
+ splitConditionHitCountThreshold,
+ stats.topics, maxBundleTopics,
+ stats.producerCount, stats.consumerCount, maxBundleSessions,
+ totalMessageRate, maxBundleMsgRate,
+ totalMessageThroughput / LoadManagerShared.MIBI,
+ maxBundleBandwidth / LoadManagerShared.MIBI
+ ));
+ }
+ continue;
}
- if (bundleHighTrafficFrequency.getOrDefault(bundle, 0) > splitConditionThreshold) {
- final String namespace = LoadManagerShared.getNamespaceNameFromBundleName(bundle);
- try {
- final int bundleCount = pulsar.getNamespaceService()
- .getBundleCount(NamespaceName.get(namespace));
- if ((bundleCount + namespaceBundleCount.getOrDefault(namespace, 0))
- < maxBundleCount) {
- if (debug) {
- log.info("The bundle {} is considered to split. Topics: {}/{}, Sessions: ({}+{})/{}, "
- + "Message Rate: {}/{} (msgs/s), Message Throughput: {}/{} (MB/s)",
- bundle, stats.topics, maxBundleTopics, stats.producerCount, stats.consumerCount,
- maxBundleSessions, totalMessageRate, maxBundleMsgRate,
- totalMessageThroughput / LoadManagerShared.MIBI,
- maxBundleBandwidth / LoadManagerShared.MIBI);
- }
- var decision = new SplitDecision();
- decision.setSplit(new Split(bundle, context.brokerRegistry().getBrokerId()));
- decision.succeed(reason);
- decisionCache.add(decision);
- int bundleNum = namespaceBundleCount.getOrDefault(namespace, 0);
- namespaceBundleCount.put(namespace, bundleNum + 1);
- bundleHighTrafficFrequency.remove(bundle);
- // Clear namespace bundle-cache
- namespaceBundleFactory.invalidateBundleCache(NamespaceName.get(namespaceName));
- if (decisionCache.size() == maxSplitCount) {
- if (debug) {
- log.info("Too many bundles to split in this split cycle {} / {}. Stop.",
- decisionCache.size(), maxSplitCount);
- }
- break;
- }
- } else {
- if (debug) {
- log.info(
- "Could not split namespace bundle {} because namespace {} has too many bundles:"
- + "{}", bundle, namespace, bundleCount);
- }
+ final String namespace = LoadManagerShared.getNamespaceNameFromBundleName(bundle);
+ try {
+ final int bundleCount = pulsar.getNamespaceService()
+ .getBundleCount(NamespaceName.get(namespace));
+ if ((bundleCount + namespaceBundleCount.getOrDefault(namespace, 0))
+ >= maxBundleCount) {
+ if (debug) {
+ log.info(String.format(CANNOT_SPLIT_BUNDLE_MSG + " Namespace:%s has too many bundles:%d",
+ bundle, namespace, bundleCount));
}
- } catch (Exception e) {
- counter.update(Failure, Unknown);
- log.warn("Error while computing bundle splits for namespace {}", namespace, e);
+ continue;
+ }
+ } catch (Exception e) {
+ counter.update(Failure, Unknown);
+ log.warn("Failed to get bundle count in namespace:{}", namespace, e);
+ continue;
+ }
+
+ if (debug) {
+ log.info(String.format(
+ "Splitting bundle: %s. "
+ + "Topics: %d/%d, "
+ + "Sessions: (%d+%d)/%d, "
+ + "Message Rate: %.2f/%d (msgs/s), "
+ + "Message Throughput: %.2f/%d (MB/s)",
+ bundle,
+ stats.topics, maxBundleTopics,
+ stats.producerCount, stats.consumerCount, maxBundleSessions,
+ totalMessageRate, maxBundleMsgRate,
+ totalMessageThroughput / LoadManagerShared.MIBI,
+ maxBundleBandwidth / LoadManagerShared.MIBI
+ ));
+ }
+ var decision = new SplitDecision();
+ decision.setSplit(new Split(bundle, context.brokerRegistry().getBrokerId()));
+ decision.succeed(reason);
+ decisionCache.add(decision);
+ int bundleNum = namespaceBundleCount.getOrDefault(namespace, 0);
+ namespaceBundleCount.put(namespace, bundleNum + 1);
+ splitConditionHitCounts.remove(bundle);
+ // Clear namespace bundle-cache
+ namespaceBundleFactory.invalidateBundleCache(NamespaceName.get(namespaceName));
+ if (decisionCache.size() == maxSplitCount) {
+ if (debug) {
+ log.info(CANNOT_CONTINUE_SPLIT_MSG
+ + "Too many bundles split in this cycle {} / {}.",
+ decisionCache.size(), maxSplitCount);
}
+ break;
}
+
}
return decisionCache;
}
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/extensions/strategy/LeastResourceUsageWithWeight.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/extensions/strategy/LeastResourceUsageWithWeight.java
index 678927dac92..902cfdaf73f 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/extensions/strategy/LeastResourceUsageWithWeight.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/extensions/strategy/LeastResourceUsageWithWeight.java
@@ -82,7 +82,7 @@ public class LeastResourceUsageWithWeight implements BrokerSelectionStrategy {
Set<String> candidates, ServiceUnitId bundleToAssign, LoadManagerContext context) {
var conf = context.brokerConfiguration();
if (candidates.isEmpty()) {
- log.info("There are no available brokers as candidates at this point for bundle: {}", bundleToAssign);
+ log.warn("There are no available brokers as candidates at this point for bundle: {}", bundleToAssign);
return Optional.empty();
}
@@ -131,8 +131,10 @@ public class LeastResourceUsageWithWeight implements BrokerSelectionStrategy {
if (bestBrokers.isEmpty()) {
// Assign randomly as all brokers are overloaded.
- log.warn("Assign randomly as none of the brokers are underloaded. candidatesSize:{}, "
- + "noLoadDataBrokersSize:{}", candidates.size(), noLoadDataBrokers.size());
+ if (debugMode) {
+ log.info("Assign randomly as none of the brokers are underloaded. candidatesSize:{}, "
+ + "noLoadDataBrokersSize:{}", candidates.size(), noLoadDataBrokers.size());
+ }
for (String broker : candidates) {
bestBrokers.add(broker);
}
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/extensions/ExtensibleLoadManagerImplTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/extensions/ExtensibleLoadManagerImplTest.java
index f8a7a9b629f..4c57e6b93fa 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/extensions/ExtensibleLoadManagerImplTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/extensions/ExtensibleLoadManagerImplTest.java
@@ -26,8 +26,8 @@ import static org.apache.pulsar.broker.loadbalance.extensions.models.SplitDecisi
import static org.apache.pulsar.broker.loadbalance.extensions.models.UnloadDecision.Label.Failure;
import static org.apache.pulsar.broker.loadbalance.extensions.models.UnloadDecision.Label.Skip;
import static org.apache.pulsar.broker.loadbalance.extensions.models.UnloadDecision.Label.Success;
-import static org.apache.pulsar.broker.loadbalance.extensions.models.UnloadDecision.Reason.Balanced;
import static org.apache.pulsar.broker.loadbalance.extensions.models.UnloadDecision.Reason.CoolDown;
+import static org.apache.pulsar.broker.loadbalance.extensions.models.UnloadDecision.Reason.HitCount;
import static org.apache.pulsar.broker.loadbalance.extensions.models.UnloadDecision.Reason.NoBrokers;
import static org.apache.pulsar.broker.loadbalance.extensions.models.UnloadDecision.Reason.NoBundles;
import static org.apache.pulsar.broker.loadbalance.extensions.models.UnloadDecision.Reason.NoLoadData;
@@ -671,7 +671,7 @@ public class ExtensibleLoadManagerImplTest extends MockedPulsarServiceBaseTest {
put(Underloaded, new AtomicLong(2));
}},
Skip, new LinkedHashMap<>() {{
- put(Balanced, new AtomicLong(3));
+ put(HitCount, new AtomicLong(3));
put(NoBundles, new AtomicLong(4));
put(CoolDown, new AtomicLong(5));
put(OutDatedData, new AtomicLong(6));
@@ -756,7 +756,7 @@ public class ExtensibleLoadManagerImplTest extends MockedPulsarServiceBaseTest {
dimensions=[{broker=localhost, feature=max, metric=loadBalancing}], metrics=[{brk_lb_resource_usage=0.04}]
dimensions=[{broker=localhost, metric=bundleUnloading}], metrics=[{brk_lb_unload_broker_total=2, brk_lb_unload_bundle_total=3}]
dimensions=[{broker=localhost, metric=bundleUnloading, reason=Unknown, result=Failure}], metrics=[{brk_lb_unload_broker_breakdown_total=10}]
- dimensions=[{broker=localhost, metric=bundleUnloading, reason=Balanced, result=Skip}], metrics=[{brk_lb_unload_broker_breakdown_total=3}]
+ dimensions=[{broker=localhost, metric=bundleUnloading, reason=HitCount, result=Skip}], metrics=[{brk_lb_unload_broker_breakdown_total=3}]
dimensions=[{broker=localhost, metric=bundleUnloading, reason=NoBundles, result=Skip}], metrics=[{brk_lb_unload_broker_breakdown_total=4}]
dimensions=[{broker=localhost, metric=bundleUnloading, reason=CoolDown, result=Skip}], metrics=[{brk_lb_unload_broker_breakdown_total=5}]
dimensions=[{broker=localhost, metric=bundleUnloading, reason=OutDatedData, result=Skip}], metrics=[{brk_lb_unload_broker_breakdown_total=6}]
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/extensions/channel/ServiceUnitStateChannelTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/extensions/channel/ServiceUnitStateChannelTest.java
index c05c8ac7415..9eda98e5d84 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/extensions/channel/ServiceUnitStateChannelTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/extensions/channel/ServiceUnitStateChannelTest.java
@@ -78,6 +78,7 @@ import org.apache.pulsar.broker.loadbalance.extensions.BrokerRegistryImpl;
import org.apache.pulsar.broker.loadbalance.extensions.LoadManagerContext;
import org.apache.pulsar.broker.loadbalance.extensions.models.Split;
import org.apache.pulsar.broker.loadbalance.extensions.models.Unload;
+import org.apache.pulsar.broker.loadbalance.extensions.store.LoadDataStore;
import org.apache.pulsar.broker.loadbalance.extensions.strategy.BrokerSelectionStrategy;
import org.apache.pulsar.broker.namespace.NamespaceService;
import org.apache.pulsar.broker.testcontext.PulsarTestContext;
@@ -132,6 +133,8 @@ public class ServiceUnitStateChannelTest extends MockedPulsarServiceBaseTest {
pulsar1 = pulsar;
registry = new BrokerRegistryImpl(pulsar);
loadManagerContext = mock(LoadManagerContext.class);
+ doReturn(mock(LoadDataStore.class)).when(loadManagerContext).brokerLoadDataStore();
+ doReturn(mock(LoadDataStore.class)).when(loadManagerContext).topBundleLoadDataStore();
brokerSelector = mock(BrokerSelectionStrategy.class);
additionalPulsarTestContext = createAdditionalPulsarTestContext(getDefaultConf());
pulsar2 = additionalPulsarTestContext.getPulsarService();
@@ -1208,6 +1211,69 @@ public class ServiceUnitStateChannelTest extends MockedPulsarServiceBaseTest {
}
+ @Test(priority = 15)
+ public void testIsOwner() throws IllegalAccessException {
+
+ var owner1 = channel1.isOwner(bundle);
+ var owner2 = channel2.isOwner(bundle);
+
+ assertFalse(owner1);
+ assertFalse(owner2);
+
+ owner1 = channel1.isOwner(bundle, lookupServiceAddress2);
+ owner2 = channel2.isOwner(bundle, lookupServiceAddress1);
+
+ assertFalse(owner1);
+ assertFalse(owner2);
+
+ channel1.publishAssignEventAsync(bundle, lookupServiceAddress1);
+ owner2 = channel2.isOwner(bundle);
+ assertFalse(owner2);
+
+ waitUntilOwnerChanges(channel1, bundle, null);
+ waitUntilOwnerChanges(channel2, bundle, null);
+
+ owner1 = channel1.isOwner(bundle);
+ owner2 = channel2.isOwner(bundle);
+
+ assertTrue(owner1);
+ assertFalse(owner2);
+
+ owner1 = channel1.isOwner(bundle, lookupServiceAddress1);
+ owner2 = channel2.isOwner(bundle, lookupServiceAddress2);
+
+ assertTrue(owner1);
+ assertFalse(owner2);
+
+ owner1 = channel2.isOwner(bundle, lookupServiceAddress1);
+ owner2 = channel1.isOwner(bundle, lookupServiceAddress2);
+
+ assertTrue(owner1);
+ assertFalse(owner2);
+
+ overrideTableView(channel1, bundle, new ServiceUnitStateData(Assigning, lookupServiceAddress1, 1));
+ assertFalse(channel1.isOwner(bundle));
+
+ overrideTableView(channel1, bundle, new ServiceUnitStateData(Owned, lookupServiceAddress1, 1));
+ assertTrue(channel1.isOwner(bundle));
+
+ overrideTableView(channel1, bundle, new ServiceUnitStateData(Releasing, null, lookupServiceAddress1, 1));
+ assertFalse(channel1.isOwner(bundle));
+
+ overrideTableView(channel1, bundle, new ServiceUnitStateData(Splitting, null, lookupServiceAddress1, 1));
+ assertTrue(channel1.isOwner(bundle));
+
+ overrideTableView(channel1, bundle, new ServiceUnitStateData(Free, null, lookupServiceAddress1, 1));
+ assertFalse(channel1.isOwner(bundle));
+
+ overrideTableView(channel1, bundle, new ServiceUnitStateData(Deleted, null, lookupServiceAddress1, 1));
+ assertFalse(channel1.isOwner(bundle));
+
+ overrideTableView(channel1, bundle, null);
+ assertFalse(channel1.isOwner(bundle));
+ }
+
+
private static ConcurrentOpenHashMap<String, CompletableFuture<Optional<String>>> getOwnerRequests(
ServiceUnitStateChannel channel) throws IllegalAccessException {
return (ConcurrentOpenHashMap<String, CompletableFuture<Optional<String>>>)
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/extensions/data/BrokerLoadDataTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/extensions/data/BrokerLoadDataTest.java
index 5ba7629dd11..85792a7ba93 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/extensions/data/BrokerLoadDataTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/extensions/data/BrokerLoadDataTest.java
@@ -74,6 +74,7 @@ public class BrokerLoadDataTest {
assertEquals(data.getTopics(), 6);
assertEquals(data.getMaxResourceUsage(), 0.04); // skips memory usage
assertEquals(data.getWeightedMaxEMA(), 2);
+ assertEquals(data.getMsgThroughputEMA(), 3);
assertThat(data.getUpdatedAt(), greaterThanOrEqualTo(now));
now = System.currentTimeMillis();
@@ -103,6 +104,7 @@ public class BrokerLoadDataTest {
assertEquals(data.getTopics(), 10);
assertEquals(data.getMaxResourceUsage(), 3.0);
assertEquals(data.getWeightedMaxEMA(), 1.875);
+ assertEquals(data.getMsgThroughputEMA(), 5);
assertThat(data.getUpdatedAt(), greaterThanOrEqualTo(now));
assertEquals(data.getReportedAt(), 0l);
assertEquals(data.toString(conf), "cpu= 300.00%, memory= 100.00%, directMemory= 2.00%, "
@@ -111,8 +113,11 @@ public class BrokerLoadDataTest {
+ "bandwithInResourceWeight= 0.500000, bandwithOutResourceWeight= 0.500000, "
+ "msgThroughputIn= 5.00, msgThroughputOut= 6.00, "
+ "msgRateIn= 7.00, msgRateOut= 8.00, bundleCount= 9, "
- + "maxResourceUsage= 300.00%, weightedMaxEMA= 187.50%, "
+ + "maxResourceUsage= 300.00%, weightedMaxEMA= 187.50%, msgThroughputEMA= 5.00, "
+ "updatedAt= " + data.getUpdatedAt() + ", reportedAt= " + data.getReportedAt());
+
+ data.clear();
+ assertEquals(data, new BrokerLoadData());
}
@Test
@@ -143,6 +148,9 @@ public class BrokerLoadDataTest {
data.update(other);
assertEquals(data, other);
+
+ data.clear();
+ assertEquals(data, new BrokerLoadData());
}
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/extensions/models/TopKBundlesTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/extensions/models/TopKBundlesTest.java
index 9b42163bd66..472d44df890 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/extensions/models/TopKBundlesTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/extensions/models/TopKBundlesTest.java
@@ -87,15 +87,15 @@ public class TopKBundlesTest {
Map<String, NamespaceBundleStats> bundleStats = new HashMap<>();
var topKBundles = new TopKBundles(pulsar);
NamespaceBundleStats stats1 = new NamespaceBundleStats();
- stats1.msgRateIn = 500;
+ stats1.msgRateIn = 100000;
bundleStats.put(bundle1, stats1);
NamespaceBundleStats stats2 = new NamespaceBundleStats();
- stats2.msgRateIn = 10000;
+ stats2.msgRateIn = 500;
bundleStats.put(bundle2, stats2);
NamespaceBundleStats stats3 = new NamespaceBundleStats();
- stats3.msgRateIn = 100000;
+ stats3.msgRateIn = 10000;
bundleStats.put(bundle3, stats3);
NamespaceBundleStats stats4 = new NamespaceBundleStats();
@@ -107,8 +107,8 @@ public class TopKBundlesTest {
var top1 = topKBundles.getLoadData().getTopBundlesLoadData().get(1);
var top2 = topKBundles.getLoadData().getTopBundlesLoadData().get(2);
- assertEquals(top0.bundleName(), bundle3);
- assertEquals(top1.bundleName(), bundle2);
+ assertEquals(top0.bundleName(), bundle2);
+ assertEquals(top1.bundleName(), bundle3);
assertEquals(top2.bundleName(), bundle1);
}
@@ -225,8 +225,8 @@ public class TopKBundlesTest {
var top0 = topKBundles.getLoadData().getTopBundlesLoadData().get(0);
var top1 = topKBundles.getLoadData().getTopBundlesLoadData().get(1);
- assertEquals(top0.bundleName(), bundle2);
- assertEquals(top1.bundleName(), bundle1);
+ assertEquals(top0.bundleName(), bundle1);
+ assertEquals(top1.bundleName(), bundle2);
configuration.setLoadBalancerSheddingBundlesWithPoliciesEnabled(false);
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/extensions/reporter/BrokerLoadDataReporterTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/extensions/reporter/BrokerLoadDataReporterTest.java
index ee7e708667a..93ab35981e1 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/extensions/reporter/BrokerLoadDataReporterTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/extensions/reporter/BrokerLoadDataReporterTest.java
@@ -18,28 +18,37 @@
*/
package org.apache.pulsar.broker.loadbalance.extensions.reporter;
+import static org.apache.pulsar.broker.loadbalance.extensions.channel.ServiceUnitStateChannelImpl.VERSION_ID_INIT;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.eq;
import static org.mockito.Mockito.doReturn;
import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.spy;
import static org.mockito.Mockito.times;
import static org.mockito.Mockito.verify;
import static org.testng.Assert.assertEquals;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
import org.apache.commons.lang.reflect.FieldUtils;
import org.apache.pulsar.broker.PulsarService;
import org.apache.pulsar.broker.ServiceConfiguration;
+import org.apache.pulsar.broker.loadbalance.extensions.channel.ServiceUnitState;
+import org.apache.pulsar.broker.loadbalance.extensions.channel.ServiceUnitStateData;
import org.apache.pulsar.broker.loadbalance.extensions.data.BrokerLoadData;
import org.apache.pulsar.broker.loadbalance.extensions.store.LoadDataStore;
import org.apache.pulsar.broker.loadbalance.impl.LoadManagerShared;
import org.apache.pulsar.broker.service.BrokerService;
import org.apache.pulsar.broker.service.PulsarStats;
import org.apache.pulsar.broker.stats.BrokerStats;
+import org.apache.pulsar.client.util.ExecutorProvider;
import org.apache.pulsar.policies.data.loadbalancer.ResourceUsage;
import org.apache.pulsar.policies.data.loadbalancer.SystemResourceUsage;
import org.mockito.MockedStatic;
import org.mockito.Mockito;
+import org.testcontainers.shaded.org.awaitility.Awaitility;
+import org.testng.annotations.AfterMethod;
import org.testng.annotations.BeforeMethod;
import org.testng.annotations.Test;
@@ -52,17 +61,24 @@ public class BrokerLoadDataReporterTest {
ServiceConfiguration config;
BrokerStats brokerStats;
SystemResourceUsage usage;
+ String broker = "broker1";
+ String bundle = "bundle1";
+ ScheduledExecutorService executor;
@BeforeMethod
void setup() {
config = new ServiceConfiguration();
+ config.setLoadBalancerDebugModeEnabled(true);
pulsar = mock(PulsarService.class);
store = mock(LoadDataStore.class);
brokerService = mock(BrokerService.class);
pulsarStats = mock(PulsarStats.class);
doReturn(brokerService).when(pulsar).getBrokerService();
doReturn(config).when(pulsar).getConfiguration();
- doReturn(Executors.newSingleThreadScheduledExecutor()).when(pulsar).getLoadManagerExecutor();
+ executor = Executors
+ .newSingleThreadScheduledExecutor(new
+ ExecutorProvider.ExtendedThreadFactory("pulsar-load-manager"));
+ doReturn(executor).when(pulsar).getLoadManagerExecutor();
doReturn(pulsarStats).when(brokerService).getPulsarStats();
brokerStats = new BrokerStats(0);
brokerStats.topics = 6;
@@ -74,6 +90,7 @@ public class BrokerLoadDataReporterTest {
doReturn(pulsarStats).when(brokerService).getPulsarStats();
doReturn(brokerStats).when(pulsarStats).getBrokerStats();
doReturn(CompletableFuture.completedFuture(null)).when(store).pushAsync(any(), any());
+ doReturn(CompletableFuture.completedFuture(null)).when(store).removeAsync(any());
usage = new SystemResourceUsage();
usage.setCpu(new ResourceUsage(1.0, 100.0));
@@ -83,6 +100,11 @@ public class BrokerLoadDataReporterTest {
usage.setBandwidthOut(new ResourceUsage(4.0, 100.0));
}
+ @AfterMethod
+ void shutdown(){
+ executor.shutdown();
+ }
+
public void testGenerate() throws IllegalAccessException {
try (MockedStatic<LoadManagerShared> mockLoadManagerShared = Mockito.mockStatic(LoadManagerShared.class)) {
mockLoadManagerShared.when(() -> LoadManagerShared.getSystemResourceUsage(any())).thenReturn(usage);
@@ -124,4 +146,73 @@ public class BrokerLoadDataReporterTest {
}
}
+ @Test
+ public void testTombstone() throws IllegalAccessException, InterruptedException {
+
+ var target = spy(new BrokerLoadDataReporter(pulsar, broker, store));
+
+ target.handleEvent(bundle,
+ new ServiceUnitStateData(ServiceUnitState.Assigning, broker, VERSION_ID_INIT), null);
+ verify(store, times(0)).removeAsync(eq(broker));
+ verify(target, times(0)).tombstone();
+
+ target.handleEvent(bundle,
+ new ServiceUnitStateData(ServiceUnitState.Deleted, broker, VERSION_ID_INIT), null);
+ verify(store, times(0)).removeAsync(eq(broker));
+ verify(target, times(0)).tombstone();
+
+
+ target.handleEvent(bundle,
+ new ServiceUnitStateData(ServiceUnitState.Init, broker, VERSION_ID_INIT), null);
+ verify(store, times(0)).removeAsync(eq(broker));
+ verify(target, times(0)).tombstone();
+
+ target.handleEvent(bundle,
+ new ServiceUnitStateData(ServiceUnitState.Free, broker, VERSION_ID_INIT), null);
+ verify(store, times(0)).removeAsync(eq(broker));
+ verify(target, times(0)).tombstone();
+
+ target.handleEvent(bundle,
+ new ServiceUnitStateData(ServiceUnitState.Releasing, "broker-2", broker, VERSION_ID_INIT),
+ new RuntimeException());
+ verify(store, times(0)).removeAsync(eq(broker));
+ verify(target, times(0)).tombstone();
+
+ target.handleEvent(bundle,
+ new ServiceUnitStateData(ServiceUnitState.Releasing, "broker-2", broker, VERSION_ID_INIT), null);
+ Awaitility.waitAtMost(3, TimeUnit.SECONDS).untilAsserted(() -> {
+ verify(target, times(1)).tombstone();
+ verify(store, times(1)).removeAsync(eq(broker));
+ var localData = (BrokerLoadData) FieldUtils.readDeclaredField(target, "localData", true);
+ assertEquals(localData, new BrokerLoadData());
+ });
+
+ target.handleEvent(bundle,
+ new ServiceUnitStateData(ServiceUnitState.Releasing, "broker-2", broker, VERSION_ID_INIT), null);
+ Awaitility.waitAtMost(3, TimeUnit.SECONDS).untilAsserted(() -> {
+ verify(target, times(2)).tombstone();
+ verify(store, times(1)).removeAsync(eq(broker));
+ var localData = (BrokerLoadData) FieldUtils.readDeclaredField(target, "localData", true);
+ assertEquals(localData, new BrokerLoadData());
+ });
+
+ FieldUtils.writeDeclaredField(target, "tombstoneDelayInMillis", 0, true);
+ target.handleEvent(bundle,
+ new ServiceUnitStateData(ServiceUnitState.Splitting, "broker-2", broker, VERSION_ID_INIT), null);
+ Awaitility.waitAtMost(3, TimeUnit.SECONDS).untilAsserted(() -> {
+ verify(target, times(3)).tombstone();
+ verify(store, times(2)).removeAsync(eq(broker));
+ var localData = (BrokerLoadData) FieldUtils.readDeclaredField(target, "localData", true);
+ assertEquals(localData, new BrokerLoadData());
+ });
+
+ target.handleEvent(bundle,
+ new ServiceUnitStateData(ServiceUnitState.Owned, broker, VERSION_ID_INIT), null);
+ Awaitility.waitAtMost(3, TimeUnit.SECONDS).untilAsserted(() -> {
+ verify(target, times(4)).tombstone();
+ verify(store, times(3)).removeAsync(eq(broker));
+ var localData = (BrokerLoadData) FieldUtils.readDeclaredField(target, "localData", true);
+ assertEquals(localData, new BrokerLoadData());
+ });
+ }
}
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/extensions/reporter/TopBundleLoadDataReporterTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/extensions/reporter/TopBundleLoadDataReporterTest.java
index b5c415c405f..be8c6af2b04 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/extensions/reporter/TopBundleLoadDataReporterTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/extensions/reporter/TopBundleLoadDataReporterTest.java
@@ -18,9 +18,12 @@
*/
package org.apache.pulsar.broker.loadbalance.extensions.reporter;
+import static org.apache.pulsar.broker.loadbalance.extensions.channel.ServiceUnitStateChannelImpl.VERSION_ID_INIT;
import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.eq;
import static org.mockito.Mockito.doReturn;
import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.spy;
import static org.mockito.Mockito.times;
import static org.mockito.Mockito.verify;
import static org.testng.Assert.assertEquals;
@@ -29,9 +32,12 @@ import java.util.HashMap;
import java.util.Map;
import java.util.Optional;
import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.TimeUnit;
import org.apache.commons.lang.reflect.FieldUtils;
import org.apache.pulsar.broker.PulsarService;
import org.apache.pulsar.broker.ServiceConfiguration;
+import org.apache.pulsar.broker.loadbalance.extensions.channel.ServiceUnitState;
+import org.apache.pulsar.broker.loadbalance.extensions.channel.ServiceUnitStateData;
import org.apache.pulsar.broker.loadbalance.extensions.data.TopBundlesLoadData;
import org.apache.pulsar.broker.loadbalance.extensions.models.TopKBundles;
import org.apache.pulsar.broker.loadbalance.extensions.store.LoadDataStore;
@@ -42,6 +48,7 @@ import org.apache.pulsar.broker.service.BrokerService;
import org.apache.pulsar.broker.service.PulsarStats;
import org.apache.pulsar.metadata.api.MetadataStoreException;
import org.apache.pulsar.policies.data.loadbalancer.NamespaceBundleStats;
+import org.testcontainers.shaded.org.awaitility.Awaitility;
import org.testng.annotations.BeforeMethod;
import org.testng.annotations.Test;
@@ -58,10 +65,13 @@ public class TopBundleLoadDataReporterTest {
private LocalPoliciesResources localPoliciesResources;
String bundle1 = "my-tenant/my-namespace1/0x00000000_0x0FFFFFFF";
String bundle2 = "my-tenant/my-namespace2/0x00000000_0x0FFFFFFF";
+ String bundle = bundle1;
+ String broker = "broker-1";
@BeforeMethod
void setup() throws MetadataStoreException {
config = new ServiceConfiguration();
+ config.setLoadBalancerDebugModeEnabled(true);
pulsar = mock(PulsarService.class);
store = mock(LoadDataStore.class);
brokerService = mock(BrokerService.class);
@@ -75,6 +85,7 @@ public class TopBundleLoadDataReporterTest {
doReturn(config).when(pulsar).getConfiguration();
doReturn(pulsarStats).when(brokerService).getPulsarStats();
doReturn(CompletableFuture.completedFuture(null)).when(store).pushAsync(any(), any());
+ doReturn(CompletableFuture.completedFuture(null)).when(store).removeAsync(any());
doReturn(pulsarResources).when(pulsar).getPulsarResources();
doReturn(namespaceResources).when(pulsarResources).getNamespaceResources();
@@ -102,22 +113,23 @@ public class TopBundleLoadDataReporterTest {
public void testGenerateLoadData() throws IllegalAccessException {
doReturn(1l).when(pulsarStats).getUpdatedAt();
- config.setLoadBalancerBundleLoadReportPercentage(100);
+ config.setLoadBalancerMaxNumberOfBundlesInBundleLoadReport(2);
var target = new TopBundleLoadDataReporter(pulsar, "", store);
var expected = new TopKBundles(pulsar);
expected.update(bundleStats, 2);
assertEquals(target.generateLoadData(), expected.getLoadData());
- config.setLoadBalancerBundleLoadReportPercentage(50);
+ config.setLoadBalancerMaxNumberOfBundlesInBundleLoadReport(1);
FieldUtils.writeDeclaredField(target, "lastBundleStatsUpdatedAt", 0l, true);
expected = new TopKBundles(pulsar);
expected.update(bundleStats, 1);
assertEquals(target.generateLoadData(), expected.getLoadData());
- config.setLoadBalancerBundleLoadReportPercentage(1);
+ config.setLoadBalancerMaxNumberOfBundlesInBundleLoadReport(0);
FieldUtils.writeDeclaredField(target, "lastBundleStatsUpdatedAt", 0l, true);
+
expected = new TopKBundles(pulsar);
- expected.update(bundleStats, 1);
+ expected.update(bundleStats, 0);
assertEquals(target.generateLoadData(), expected.getLoadData());
doReturn(new HashMap()).when(brokerService).getBundleStats();
@@ -128,21 +140,83 @@ public class TopBundleLoadDataReporterTest {
public void testReportForce() {
- var target = new TopBundleLoadDataReporter(pulsar, "broker-1", store);
+ var target = new TopBundleLoadDataReporter(pulsar, broker, store);
target.reportAsync(false);
verify(store, times(0)).pushAsync(any(), any());
target.reportAsync(true);
- verify(store, times(1)).pushAsync("broker-1", new TopBundlesLoadData());
+ verify(store, times(1)).pushAsync(broker, new TopBundlesLoadData());
}
public void testReport(){
- var target = new TopBundleLoadDataReporter(pulsar, "broker-1", store);
+ pulsar.getConfiguration().setLoadBalancerMaxNumberOfBundlesInBundleLoadReport(1);
+ var target = new TopBundleLoadDataReporter(pulsar, broker, store);
doReturn(1l).when(pulsarStats).getUpdatedAt();
var expected = new TopKBundles(pulsar);
expected.update(bundleStats, 1);
target.reportAsync(false);
- verify(store, times(1)).pushAsync("broker-1", expected.getLoadData());
+ verify(store, times(1)).pushAsync(broker, expected.getLoadData());
}
+ @Test
+ public void testTombstone() throws IllegalAccessException {
+
+ var target = spy(new TopBundleLoadDataReporter(pulsar, broker, store));
+
+ target.handleEvent(bundle,
+ new ServiceUnitStateData(ServiceUnitState.Assigning, broker, VERSION_ID_INIT), null);
+ verify(store, times(0)).removeAsync(eq(broker));
+ verify(target, times(0)).tombstone();
+
+ target.handleEvent(bundle,
+ new ServiceUnitStateData(ServiceUnitState.Deleted, broker, VERSION_ID_INIT), null);
+ verify(store, times(0)).removeAsync(eq(broker));
+ verify(target, times(0)).tombstone();
+
+
+ target.handleEvent(bundle,
+ new ServiceUnitStateData(ServiceUnitState.Init, broker, VERSION_ID_INIT), null);
+ verify(store, times(0)).removeAsync(eq(broker));
+ verify(target, times(0)).tombstone();
+
+ target.handleEvent(bundle,
+ new ServiceUnitStateData(ServiceUnitState.Free, broker, VERSION_ID_INIT), null);
+ verify(store, times(0)).removeAsync(eq(broker));
+ verify(target, times(0)).tombstone();
+
+ target.handleEvent(bundle,
+ new ServiceUnitStateData(ServiceUnitState.Releasing, "broker-2", broker, VERSION_ID_INIT),
+ new RuntimeException());
+ verify(store, times(0)).removeAsync(eq(broker));
+ verify(target, times(0)).tombstone();
+
+ target.handleEvent(bundle,
+ new ServiceUnitStateData(ServiceUnitState.Releasing, "broker-2", broker, VERSION_ID_INIT), null);
+ Awaitility.waitAtMost(3, TimeUnit.SECONDS).untilAsserted(() -> {
+ verify(target, times(1)).tombstone();
+ verify(store, times(1)).removeAsync(eq(broker));
+ });
+
+ target.handleEvent(bundle,
+ new ServiceUnitStateData(ServiceUnitState.Releasing, "broker-2", broker, VERSION_ID_INIT), null);
+ Awaitility.waitAtMost(3, TimeUnit.SECONDS).untilAsserted(() -> {
+ verify(target, times(2)).tombstone();
+ verify(store, times(1)).removeAsync(eq(broker));
+ });
+
+ FieldUtils.writeDeclaredField(target, "tombstoneDelayInMillis", 0, true);
+ target.handleEvent(bundle,
+ new ServiceUnitStateData(ServiceUnitState.Splitting, "broker-2", broker, VERSION_ID_INIT), null);
+ Awaitility.waitAtMost(3, TimeUnit.SECONDS).untilAsserted(() -> {
+ verify(target, times(3)).tombstone();
+ verify(store, times(2)).removeAsync(eq(broker));
+ });
+
+ target.handleEvent(bundle,
+ new ServiceUnitStateData(ServiceUnitState.Owned, broker, VERSION_ID_INIT), null);
+ Awaitility.waitAtMost(3, TimeUnit.SECONDS).untilAsserted(() -> {
+ verify(target, times(4)).tombstone();
+ verify(store, times(3)).removeAsync(eq(broker));
+ });
+ }
}
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/extensions/scheduler/TransferShedderTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/extensions/scheduler/TransferShedderTest.java
index 5ca9345d6de..8c8d18e202a 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/extensions/scheduler/TransferShedderTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/extensions/scheduler/TransferShedderTest.java
@@ -18,10 +18,11 @@
*/
package org.apache.pulsar.broker.loadbalance.extensions.scheduler;
+import static org.apache.pulsar.broker.loadbalance.extensions.models.UnloadDecision.Label.Failure;
import static org.apache.pulsar.broker.loadbalance.extensions.models.UnloadDecision.Label.Skip;
import static org.apache.pulsar.broker.loadbalance.extensions.models.UnloadDecision.Label.Success;
-import static org.apache.pulsar.broker.loadbalance.extensions.models.UnloadDecision.Reason.Balanced;
import static org.apache.pulsar.broker.loadbalance.extensions.models.UnloadDecision.Reason.CoolDown;
+import static org.apache.pulsar.broker.loadbalance.extensions.models.UnloadDecision.Reason.HitCount;
import static org.apache.pulsar.broker.loadbalance.extensions.models.UnloadDecision.Reason.NoBrokers;
import static org.apache.pulsar.broker.loadbalance.extensions.models.UnloadDecision.Reason.NoBundles;
import static org.apache.pulsar.broker.loadbalance.extensions.models.UnloadDecision.Reason.NoLoadData;
@@ -52,6 +53,7 @@ import java.util.Random;
import java.util.Set;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.TimeoutException;
+import java.util.concurrent.atomic.AtomicReference;
import java.util.function.BiConsumer;
import org.apache.commons.lang.reflect.FieldUtils;
import org.apache.commons.math3.stat.descriptive.moment.Mean;
@@ -59,7 +61,11 @@ import org.apache.commons.math3.stat.descriptive.moment.StandardDeviation;
import org.apache.pulsar.broker.PulsarService;
import org.apache.pulsar.broker.ServiceConfiguration;
import org.apache.pulsar.broker.loadbalance.extensions.BrokerRegistry;
+import org.apache.pulsar.broker.loadbalance.extensions.ExtensibleLoadManagerImpl;
+import org.apache.pulsar.broker.loadbalance.extensions.ExtensibleLoadManagerWrapper;
import org.apache.pulsar.broker.loadbalance.extensions.LoadManagerContext;
+import org.apache.pulsar.broker.loadbalance.extensions.channel.ServiceUnitStateChannel;
+import org.apache.pulsar.broker.loadbalance.extensions.channel.ServiceUnitStateChannelImpl;
import org.apache.pulsar.broker.loadbalance.extensions.data.BrokerLoadData;
import org.apache.pulsar.broker.loadbalance.extensions.data.BrokerLookupData;
import org.apache.pulsar.broker.loadbalance.extensions.data.TopBundlesLoadData;
@@ -97,6 +103,10 @@ public class TransferShedderTest {
double setupLoadStd = 0.3982762860126121;
PulsarService pulsar;
+ NamespaceService namespaceService;
+ ExtensibleLoadManagerWrapper loadManagerWrapper;
+ ExtensibleLoadManagerImpl loadManager;
+ ServiceUnitStateChannel channel;
ServiceConfiguration conf;
AntiAffinityGroupPolicyHelper antiAffinityGroupPolicyHelper;
LocalPoliciesResources localPoliciesResources;
@@ -108,13 +118,16 @@ public class TransferShedderTest {
@BeforeMethod
public void init() throws MetadataStoreException {
pulsar = mock(PulsarService.class);
+ loadManagerWrapper = mock(ExtensibleLoadManagerWrapper.class);
+ loadManager = mock(ExtensibleLoadManagerImpl.class);
+ channel = mock(ServiceUnitStateChannelImpl.class);
conf = new ServiceConfiguration();
conf.setLoadBalancerSheddingBundlesWithPoliciesEnabled(true);
var pulsarResources = mock(PulsarResources.class);
var namespaceResources = mock(NamespaceResources.class);
var isolationPolicyResources = mock(NamespaceResources.IsolationPolicyResources.class);
var factory = mock(NamespaceBundleFactory.class);
- var namespaceService = mock(NamespaceService.class);
+ namespaceService = mock(NamespaceService.class);
localPoliciesResources = mock(LocalPoliciesResources.class);
antiAffinityGroupPolicyHelper = mock(AntiAffinityGroupPolicyHelper.class);
doReturn(conf).when(pulsar).getConfiguration();
@@ -136,25 +149,30 @@ public class TransferShedderTest {
(upperEndpoint.equals(NamespaceBundles.FULL_UPPER_BOUND)) ? BoundType.CLOSED : BoundType.OPEN);
return new NamespaceBundle(NamespaceName.get(namespace), hashRange, factory);
}).when(factory).getBundle(anyString(), anyString());
- doReturn(true).when(antiAffinityGroupPolicyHelper).canUnload(any(), any(), any(), any());
+ doReturn(new AtomicReference<>(loadManagerWrapper)).when(pulsar).getLoadManager();
+ doReturn(loadManager).when(loadManagerWrapper).get();
+ doReturn(channel).when(loadManager).getServiceUnitStateChannel();
+ doReturn(true).when(channel).isOwner(any(), any());
}
public LoadManagerContext setupContext(){
var ctx = getContext();
- var brokerLoadDataStore = ctx.brokerLoadDataStore();
- brokerLoadDataStore.pushAsync("broker1", getCpuLoad(ctx, 2));
- brokerLoadDataStore.pushAsync("broker2", getCpuLoad(ctx, 4));
- brokerLoadDataStore.pushAsync("broker3", getCpuLoad(ctx, 6));
- brokerLoadDataStore.pushAsync("broker4", getCpuLoad(ctx, 80));
- brokerLoadDataStore.pushAsync("broker5", getCpuLoad(ctx, 90));
+
var topBundlesLoadDataStore = ctx.topBundleLoadDataStore();
- topBundlesLoadDataStore.pushAsync("broker1", getTopBundlesLoad("my-tenant/my-namespaceA", 2000000, 1000000));
- topBundlesLoadDataStore.pushAsync("broker2", getTopBundlesLoad("my-tenant/my-namespaceB", 3000000, 1000000));
- topBundlesLoadDataStore.pushAsync("broker3", getTopBundlesLoad("my-tenant/my-namespaceC", 4000000, 2000000));
- topBundlesLoadDataStore.pushAsync("broker4", getTopBundlesLoad("my-tenant/my-namespaceD", 6000000, 2000000));
- topBundlesLoadDataStore.pushAsync("broker5", getTopBundlesLoad("my-tenant/my-namespaceE", 7000000, 2000000));
+ topBundlesLoadDataStore.pushAsync("broker1", getTopBundlesLoad("my-tenant/my-namespaceA", 1000000, 2000000));
+ topBundlesLoadDataStore.pushAsync("broker2", getTopBundlesLoad("my-tenant/my-namespaceB", 1000000, 3000000));
+ topBundlesLoadDataStore.pushAsync("broker3", getTopBundlesLoad("my-tenant/my-namespaceC", 2000000, 4000000));
+ topBundlesLoadDataStore.pushAsync("broker4", getTopBundlesLoad("my-tenant/my-namespaceD", 2000000, 6000000));
+ topBundlesLoadDataStore.pushAsync("broker5", getTopBundlesLoad("my-tenant/my-namespaceE", 2000000, 7000000));
+
+ var brokerLoadDataStore = ctx.brokerLoadDataStore();
+ brokerLoadDataStore.pushAsync("broker1", getCpuLoad(ctx, 2, "broker1"));
+ brokerLoadDataStore.pushAsync("broker2", getCpuLoad(ctx, 4, "broker2"));
+ brokerLoadDataStore.pushAsync("broker3", getCpuLoad(ctx, 6, "broker3"));
+ brokerLoadDataStore.pushAsync("broker4", getCpuLoad(ctx, 80, "broker4"));
+ brokerLoadDataStore.pushAsync("broker5", getCpuLoad(ctx, 90, "broker5"));
return ctx;
}
@@ -166,8 +184,8 @@ public class TransferShedderTest {
Random rand = new Random();
for (int i = 0; i < clusterSize; i++) {
- int brokerLoad = rand.nextInt(100);
- brokerLoadDataStore.pushAsync("broker" + i, getCpuLoad(ctx, brokerLoad));
+ int brokerLoad = rand.nextInt(1000);
+ brokerLoadDataStore.pushAsync("broker" + i, getCpuLoad(ctx, brokerLoad, "broker" + i));
int bundleLoad = rand.nextInt(brokerLoad + 1);
topBundlesLoadDataStore.pushAsync("broker" + i, getTopBundlesLoad("my-tenant/my-namespace" + i,
bundleLoad, brokerLoad - bundleLoad));
@@ -175,7 +193,7 @@ public class TransferShedderTest {
return ctx;
}
- public BrokerLoadData getCpuLoad(LoadManagerContext ctx, int load) {
+ public BrokerLoadData getCpuLoad(LoadManagerContext ctx, int load, String broker) {
var loadData = new BrokerLoadData();
SystemResourceUsage usage1 = new SystemResourceUsage();
var cpu = new ResourceUsage(load, 100.0);
@@ -188,8 +206,17 @@ public class TransferShedderTest {
usage1.setDirectMemory(directMemory);
usage1.setBandwidthIn(bandwidthIn);
usage1.setBandwidthOut(bandwidthOut);
- loadData.update(usage1, 1,2,3,4,5,6,
- ctx.brokerConfiguration());
+ if (ctx.topBundleLoadDataStore()
+ .get(broker).isPresent()) {
+ var throughputOut = ctx.topBundleLoadDataStore()
+ .get(broker).get()
+ .getTopBundlesLoadData().stream().mapToDouble(v -> v.stats().msgThroughputOut).sum();
+ loadData.update(usage1, 1, throughputOut, 3, 4, 5, 6,
+ ctx.brokerConfiguration());
+ } else {
+ loadData.update(usage1, 1, 2, 3, 4, 5, 6,
+ ctx.brokerConfiguration());
+ }
return loadData;
}
@@ -204,6 +231,29 @@ public class TransferShedderTest {
return topKBundles.getLoadData();
}
+ public TopBundlesLoadData getTopBundlesLoad(String bundlePrefix,
+ int load1, int load2, int load3, int load4, int load5) {
+ var namespaceBundleStats1 = new NamespaceBundleStats();
+ namespaceBundleStats1.msgThroughputOut = load1;
+ var namespaceBundleStats2 = new NamespaceBundleStats();
+ namespaceBundleStats2.msgThroughputOut = load2;
+ var namespaceBundleStats3 = new NamespaceBundleStats();
+ namespaceBundleStats3.msgThroughputOut = load3;
+ var namespaceBundleStats4 = new NamespaceBundleStats();
+ namespaceBundleStats4.msgThroughputOut = load4;
+ var namespaceBundleStats5 = new NamespaceBundleStats();
+ namespaceBundleStats5.msgThroughputOut = load5;
+ var topKBundles = new TopKBundles(pulsar);
+ topKBundles.update(Map.of(
+ bundlePrefix + "/0x00000000_0x1FFFFFFF", namespaceBundleStats1,
+ bundlePrefix + "/0x1FFFFFFF_0x2FFFFFFF", namespaceBundleStats2,
+ bundlePrefix + "/0x2FFFFFFF_0x3FFFFFFF", namespaceBundleStats3,
+ bundlePrefix + "/0x3FFFFFFF_0x4FFFFFFF", namespaceBundleStats4,
+ bundlePrefix + "/0x4FFFFFFF_0x5FFFFFFF", namespaceBundleStats5
+ ), 5);
+ return topKBundles.getLoadData();
+ }
+
public TopBundlesLoadData getTopBundlesLoad(String bundlePrefix, int load1) {
var namespaceBundleStats1 = new NamespaceBundleStats();
namespaceBundleStats1.msgThroughputOut = load1;
@@ -230,6 +280,7 @@ public class TransferShedderTest {
var conf = new ServiceConfiguration();
conf.setLoadBalancerDebugModeEnabled(true);
conf.setLoadBalancerSheddingBundlesWithPoliciesEnabled(false);
+ conf.setLoadBalancerSheddingConditionHitCountThreshold(0);
var brokerLoadDataStore = new LoadDataStore<BrokerLoadData>() {
Map<String, BrokerLoadData> map = new HashMap<>();
@Override
@@ -357,29 +408,44 @@ public class TransferShedderTest {
UnloadCounter counter = new UnloadCounter();
TransferShedder transferShedder = new TransferShedder(counter);
var ctx = getContext();
- ctx.brokerConfiguration().setLoadBalancerDebugModeEnabled(true);
var res = transferShedder.findBundlesForUnloading(ctx, Map.of(), Map.of());
assertTrue(res.isEmpty());
assertEquals(counter.getBreakdownCounters().get(Skip).get(NoBrokers).get(), 1);
}
+ @Test
+ public void testNoOwnerLoadData() throws IllegalAccessException {
+ UnloadCounter counter = new UnloadCounter();
+ TransferShedder transferShedder = new TransferShedder(counter);
+ FieldUtils.writeDeclaredField(transferShedder, "channel", channel, true);
+ var ctx = setupContext();
+ doReturn(false).when(channel).isOwner(any(), any());
+ var res = transferShedder.findBundlesForUnloading(ctx, Map.of(), Map.of());
+ assertTrue(res.isEmpty());
+ assertEquals(counter.getBreakdownCounters().get(Skip).get(NoBundles).get(), 1);
+ assertEquals(counter.getLoadAvg(), setupLoadAvg);
+ assertEquals(counter.getLoadStd(), setupLoadStd);
+ }
+
@Test
public void testEmptyTopBundlesLoadData() {
UnloadCounter counter = new UnloadCounter();
TransferShedder transferShedder = new TransferShedder(counter);
var ctx = getContext();
- ctx.brokerConfiguration().setLoadBalancerDebugModeEnabled(true);
var brokerLoadDataStore = ctx.brokerLoadDataStore();
- brokerLoadDataStore.pushAsync("broker1", getCpuLoad(ctx, 90));
- brokerLoadDataStore.pushAsync("broker2", getCpuLoad(ctx, 10));
- brokerLoadDataStore.pushAsync("broker3", getCpuLoad(ctx, 20));
+ brokerLoadDataStore.pushAsync("broker1", getCpuLoad(ctx, 2, "broker1"));
+ brokerLoadDataStore.pushAsync("broker2", getCpuLoad(ctx, 4, "broker2"));
+ brokerLoadDataStore.pushAsync("broker3", getCpuLoad(ctx, 6, "broker3"));
+ brokerLoadDataStore.pushAsync("broker4", getCpuLoad(ctx, 80, "broker4"));
+ brokerLoadDataStore.pushAsync("broker5", getCpuLoad(ctx, 90, "broker5"));
+
var res = transferShedder.findBundlesForUnloading(ctx, Map.of(), Map.of());
assertTrue(res.isEmpty());
assertEquals(counter.getBreakdownCounters().get(Skip).get(NoLoadData).get(), 1);
- assertEquals(counter.getLoadAvg(), 0.39999999999999997);
- assertEquals(counter.getLoadStd(), 0.35590260840104376);
+ assertEquals(counter.getLoadAvg(), setupLoadAvg);
+ assertEquals(counter.getLoadStd(), setupLoadStd);
}
@Test
@@ -445,16 +511,18 @@ public class TransferShedderTest {
recentlyUnloadedBundles.put(bundleD1, now);
recentlyUnloadedBundles.put(bundleD2, now);
var res = transferShedder.findBundlesForUnloading(ctx, recentlyUnloadedBundles, Map.of());
-
- assertTrue(res.isEmpty());
- assertEquals(counter.getBreakdownCounters().get(Skip).get(NoBundles).get(), 1);
+ var expected = new HashSet<UnloadDecision>();
+ expected.add(new UnloadDecision(new Unload("broker3",
+ "my-tenant/my-namespaceC/0x00000000_0x0FFFFFFF",
+ Optional.of("broker1")),
+ Success, Overloaded));
+ assertEquals(res, expected);
assertEquals(counter.getLoadAvg(), setupLoadAvg);
assertEquals(counter.getLoadStd(), setupLoadStd);
}
@Test
public void testGetAvailableBrokersFailed() {
- var pulsar = getMockPulsar();
UnloadCounter counter = new UnloadCounter();
AntiAffinityGroupPolicyHelper affinityGroupPolicyHelper = mock(AntiAffinityGroupPolicyHelper.class);
TransferShedder transferShedder = new TransferShedder(pulsar, counter, affinityGroupPolicyHelper);
@@ -462,17 +530,16 @@ public class TransferShedderTest {
BrokerRegistry registry = ctx.brokerRegistry();
doReturn(FutureUtil.failedFuture(new TimeoutException())).when(registry).getAvailableBrokerLookupDataAsync();
var res = transferShedder.findBundlesForUnloading(ctx, Map.of(), Map.of());
- assertEquals(counter.getBreakdownCounters().get(Skip).get(Unknown).get(), 1);
- assertEquals(counter.getLoadAvg(), setupLoadAvg);
- assertEquals(counter.getLoadStd(), setupLoadStd);
+ assertEquals(counter.getBreakdownCounters().get(Failure).get(Unknown).get(), 1);
+ assertEquals(counter.getLoadAvg(), 0.0);
+ assertEquals(counter.getLoadStd(), 0.0);
}
@Test(timeOut = 30 * 1000)
public void testBundlesWithIsolationPolicies() throws IllegalAccessException {
- var pulsar = getMockPulsar();
+ doReturn(true).when(antiAffinityGroupPolicyHelper).canUnload(any(), any(), any(), any());
UnloadCounter counter = new UnloadCounter();
TransferShedder transferShedder = spy(new TransferShedder(pulsar, counter, antiAffinityGroupPolicyHelper));
-
var allocationPoliciesSpy = (SimpleResourceAllocationPolicies)
spy(FieldUtils.readDeclaredField(transferShedder, "allocationPolicies", true));
FieldUtils.writeDeclaredField(transferShedder, "allocationPolicies", allocationPoliciesSpy, true);
@@ -484,7 +551,7 @@ public class TransferShedderTest {
var ctx = setupContext();
var res = transferShedder.findBundlesForUnloading(ctx, Map.of(), Map.of());
var expected = new HashSet<UnloadDecision>();
- expected.add(new UnloadDecision(new Unload("broker4", bundleD1, Optional.of("broker2")),
+ expected.add(new UnloadDecision(new Unload("broker4", bundleD1, Optional.of("broker1")),
Success, Overloaded));
assertEquals(res, expected);
assertEquals(counter.getLoadAvg(), setupLoadAvg);
@@ -572,29 +639,9 @@ public class TransferShedderTest {
}).when(policies).shouldFailoverToSecondaries(eq(namespaceName), anyInt());
}
- private PulsarService getMockPulsar() {
- var pulsar = mock(PulsarService.class);
- var namespaceService = mock(NamespaceService.class);
- doReturn(namespaceService).when(pulsar).getNamespaceService();
- NamespaceBundleFactory factory = mock(NamespaceBundleFactory.class);
- doReturn(factory).when(namespaceService).getNamespaceBundleFactory();
- doAnswer(answer -> {
- String namespace = answer.getArgument(0, String.class);
- String bundleRange = answer.getArgument(1, String.class);
- String[] boundaries = bundleRange.split("_");
- Long lowerEndpoint = Long.decode(boundaries[0]);
- Long upperEndpoint = Long.decode(boundaries[1]);
- Range<Long> hashRange = Range.range(lowerEndpoint, BoundType.CLOSED, upperEndpoint,
- (upperEndpoint.equals(NamespaceBundles.FULL_UPPER_BOUND)) ? BoundType.CLOSED : BoundType.OPEN);
- return new NamespaceBundle(NamespaceName.get(namespace), hashRange, factory);
- }).when(factory).getBundle(anyString(), anyString());
- return pulsar;
- }
-
@Test
public void testBundlesWithAntiAffinityGroup() throws IllegalAccessException, MetadataStoreException {
- var pulsar = getMockPulsar();
var counter = new UnloadCounter();
TransferShedder transferShedder = new TransferShedder(pulsar, counter, antiAffinityGroupPolicyHelper);
var allocationPoliciesSpy = (SimpleResourceAllocationPolicies)
@@ -606,7 +653,10 @@ public class TransferShedderTest {
doReturn(Optional.of(localPolicies)).when(localPoliciesResources).getLocalPolicies(any());
var ctx = setupContext();
- doReturn(false).when(antiAffinityGroupPolicyHelper).canUnload(any(), any(), any(), any());
+ var antiAffinityGroupPolicyHelperSpy = (AntiAffinityGroupPolicyHelper)
+ spy(FieldUtils.readDeclaredField(transferShedder, "antiAffinityGroupPolicyHelper", true));
+ doReturn(false).when(antiAffinityGroupPolicyHelperSpy).canUnload(any(), any(), any(), any());
+ FieldUtils.writeDeclaredField(transferShedder, "antiAffinityGroupPolicyHelper", antiAffinityGroupPolicyHelperSpy, true);
var res = transferShedder.findBundlesForUnloading(ctx, Map.of(), Map.of());
assertTrue(res.isEmpty());
@@ -615,7 +665,7 @@ public class TransferShedderTest {
assertEquals(counter.getLoadStd(), setupLoadStd);
- doReturn(true).when(antiAffinityGroupPolicyHelper).canUnload(any(), eq(bundleE1), any(), any());
+ doReturn(true).when(antiAffinityGroupPolicyHelperSpy).canUnload(any(), eq(bundleE1), any(), any());
var res2 = transferShedder.findBundlesForUnloading(ctx, Map.of(), Map.of());
var expected2 = new HashSet<>();
expected2.add(new UnloadDecision(new Unload("broker5", bundleE1, Optional.of("broker1")),
@@ -630,11 +680,18 @@ public class TransferShedderTest {
UnloadCounter counter = new UnloadCounter();
TransferShedder transferShedder = new TransferShedder(counter);
var ctx = getContext();
+ BrokerRegistry brokerRegistry = mock(BrokerRegistry.class);
+ doReturn(CompletableFuture.completedFuture(Map.of(
+ "broker1", mock(BrokerLookupData.class),
+ "broker2", mock(BrokerLookupData.class),
+ "broker3", mock(BrokerLookupData.class)
+ ))).when(brokerRegistry).getAvailableBrokerLookupDataAsync();
+ doReturn(brokerRegistry).when(ctx).brokerRegistry();
ctx.brokerConfiguration().setLoadBalancerDebugModeEnabled(true);
var brokerLoadDataStore = ctx.brokerLoadDataStore();
- brokerLoadDataStore.pushAsync("broker1", getCpuLoad(ctx, 10));
- brokerLoadDataStore.pushAsync("broker2", getCpuLoad(ctx, 20));
- brokerLoadDataStore.pushAsync("broker3", getCpuLoad(ctx, 30));
+ brokerLoadDataStore.pushAsync("broker1", getCpuLoad(ctx, 10, "broker1"));
+ brokerLoadDataStore.pushAsync("broker2", getCpuLoad(ctx, 20, "broker2"));
+ brokerLoadDataStore.pushAsync("broker3", getCpuLoad(ctx, 30, "broker3"));
var topBundlesLoadDataStore = ctx.topBundleLoadDataStore();
@@ -645,7 +702,7 @@ public class TransferShedderTest {
var res = transferShedder.findBundlesForUnloading(ctx, Map.of(), Map.of());
assertTrue(res.isEmpty());
- assertEquals(counter.getBreakdownCounters().get(Skip).get(Balanced).get(), 1);
+ assertEquals(counter.getBreakdownCounters().get(Skip).get(HitCount).get(), 1);
assertEquals(counter.getLoadAvg(), 0.2000000063578288);
assertEquals(counter.getLoadStd(), 0.08164966587949089);
}
@@ -669,6 +726,26 @@ public class TransferShedderTest {
assertEquals(counter.getLoadStd(), setupLoadStd);
}
+ @Test
+ public void testBundleThroughputLargerThanOffloadThreshold() {
+ UnloadCounter counter = new UnloadCounter();
+ TransferShedder transferShedder = new TransferShedder(counter);
+ var ctx = setupContext();
+ var topBundlesLoadDataStore = ctx.topBundleLoadDataStore();
+ topBundlesLoadDataStore.pushAsync("broker4", getTopBundlesLoad("my-tenant/my-namespaceD", 1000000000, 1000000000));
+ topBundlesLoadDataStore.pushAsync("broker5", getTopBundlesLoad("my-tenant/my-namespaceE", 1000000000, 1000000000));
+ var res = transferShedder.findBundlesForUnloading(ctx, Map.of(), Map.of());
+
+ var expected = new HashSet<UnloadDecision>();
+ expected.add(new UnloadDecision(new Unload("broker3",
+ "my-tenant/my-namespaceC/0x00000000_0x0FFFFFFF",
+ Optional.of("broker1")),
+ Success, Overloaded));
+ assertEquals(res, expected);
+ assertEquals(counter.getLoadAvg(), setupLoadAvg);
+ assertEquals(counter.getLoadStd(), setupLoadStd);
+ }
+
@Test
public void testTargetStdAfterTransfer() {
@@ -676,8 +753,8 @@ public class TransferShedderTest {
TransferShedder transferShedder = new TransferShedder(counter);
var ctx = setupContext();
var brokerLoadDataStore = ctx.brokerLoadDataStore();
- brokerLoadDataStore.pushAsync("broker4", getCpuLoad(ctx, 55));
- brokerLoadDataStore.pushAsync("broker5", getCpuLoad(ctx, 65));
+ brokerLoadDataStore.pushAsync("broker4", getCpuLoad(ctx, 55, "broker4"));
+ brokerLoadDataStore.pushAsync("broker5", getCpuLoad(ctx, 65, "broker5"));
var res = transferShedder.findBundlesForUnloading(ctx, Map.of(), Map.of());
var expected = new HashSet<UnloadDecision>();
@@ -688,19 +765,186 @@ public class TransferShedderTest {
assertEquals(counter.getLoadStd(), 0.27644891028904417);
}
+ @Test
+ public void testUnloadBundlesGreaterThanTargetThroughput() throws IllegalAccessException {
+ UnloadCounter counter = new UnloadCounter();
+ TransferShedder transferShedder = new TransferShedder(counter);
+ var ctx = getContext();
+
+ var brokerRegistry = mock(BrokerRegistry.class);
+ doReturn(brokerRegistry).when(ctx).brokerRegistry();
+ doReturn(CompletableFuture.completedFuture(Map.of(
+ "broker1", mock(BrokerLookupData.class),
+ "broker2", mock(BrokerLookupData.class)
+ ))).when(brokerRegistry).getAvailableBrokerLookupDataAsync();
+
+ var topBundlesLoadDataStore = ctx.topBundleLoadDataStore();
+ topBundlesLoadDataStore.pushAsync("broker1", getTopBundlesLoad("my-tenant/my-namespaceA", 1000000, 2000000, 3000000, 4000000, 5000000));
+ topBundlesLoadDataStore.pushAsync("broker2",
+ getTopBundlesLoad("my-tenant/my-namespaceB", 100000000, 180000000, 220000000, 250000000, 250000000));
+
+
+ var brokerLoadDataStore = ctx.brokerLoadDataStore();
+ brokerLoadDataStore.pushAsync("broker1", getCpuLoad(ctx, 10, "broker1"));
+ brokerLoadDataStore.pushAsync("broker2", getCpuLoad(ctx, 1000, "broker2"));
+
+
+ var res = transferShedder.findBundlesForUnloading(ctx, Map.of(), Map.of());
+ var expected = new HashSet<UnloadDecision>();
+ expected.add(new UnloadDecision(
+ new Unload("broker2", "my-tenant/my-namespaceB/0x00000000_0x1FFFFFFF", Optional.of("broker1")),
+ Success, Overloaded));
+ expected.add(new UnloadDecision(
+ new Unload("broker2", "my-tenant/my-namespaceB/0x1FFFFFFF_0x2FFFFFFF", Optional.of("broker1")),
+ Success, Overloaded));
+ expected.add(new UnloadDecision(
+ new Unload("broker2", "my-tenant/my-namespaceB/0x2FFFFFFF_0x3FFFFFFF", Optional.of("broker1")),
+ Success, Overloaded));
+ expected.add(new UnloadDecision(
+ new Unload("broker1", "my-tenant/my-namespaceA/0x00000000_0x1FFFFFFF", Optional.of("broker2")),
+ Success, Overloaded));
+ expected.add(new UnloadDecision(
+ new Unload("broker1", "my-tenant/my-namespaceA/0x1FFFFFFF_0x2FFFFFFF", Optional.of("broker2")),
+ Success, Overloaded));
+ expected.add(new UnloadDecision(
+ new Unload("broker1","my-tenant/my-namespaceA/0x2FFFFFFF_0x3FFFFFFF", Optional.of("broker2")),
+ Success, Overloaded));
+ expected.add(new UnloadDecision(
+ new Unload("broker1","my-tenant/my-namespaceA/0x3FFFFFFF_0x4FFFFFFF", Optional.of("broker2")),
+ Success, Overloaded));
+ assertEquals(counter.getLoadAvg(), 5.05);
+ assertEquals(counter.getLoadStd(), 4.95);
+ assertEquals(res, expected);
+ var stats = (TransferShedder.LoadStats)
+ FieldUtils.readDeclaredField(transferShedder, "stats", true);
+ assertEquals(stats.std(), 0.050000004900021836);
+ }
+
+ @Test
+ public void testSkipBundlesGreaterThanTargetThroughputAfterSplit() {
+ UnloadCounter counter = new UnloadCounter();
+ TransferShedder transferShedder = new TransferShedder(counter);
+ var ctx = getContext();
+ ctx.brokerConfiguration().setLoadBalancerBrokerLoadTargetStd(0.20);
+
+ var brokerRegistry = mock(BrokerRegistry.class);
+ doReturn(brokerRegistry).when(ctx).brokerRegistry();
+ doReturn(CompletableFuture.completedFuture(Map.of(
+ "broker1", mock(BrokerLookupData.class),
+ "broker2", mock(BrokerLookupData.class)
+ ))).when(brokerRegistry).getAvailableBrokerLookupDataAsync();
+
+ var topBundlesLoadDataStore = ctx.topBundleLoadDataStore();
+ topBundlesLoadDataStore.pushAsync("broker1",
+ getTopBundlesLoad("my-tenant/my-namespaceA", 1, 500000000));
+ topBundlesLoadDataStore.pushAsync("broker2",
+ getTopBundlesLoad("my-tenant/my-namespaceB", 500000000, 500000000));
+
+
+ var brokerLoadDataStore = ctx.brokerLoadDataStore();
+ brokerLoadDataStore.pushAsync("broker1", getCpuLoad(ctx, 50, "broker1"));
+ brokerLoadDataStore.pushAsync("broker2", getCpuLoad(ctx, 100, "broker2"));
+
+
+ var res = transferShedder.findBundlesForUnloading(ctx, Map.of(), Map.of());
+ assertTrue(res.isEmpty());
+ assertEquals(counter.getBreakdownCounters().get(Skip).get(NoBundles).get(), 1);
+ }
+
+
+ @Test
+ public void testUnloadBundlesLessThanTargetThroughputAfterSplit() throws IllegalAccessException {
+ UnloadCounter counter = new UnloadCounter();
+ TransferShedder transferShedder = new TransferShedder(counter);
+ var ctx = getContext();
+
+ var brokerRegistry = mock(BrokerRegistry.class);
+ doReturn(brokerRegistry).when(ctx).brokerRegistry();
+ doReturn(CompletableFuture.completedFuture(Map.of(
+ "broker1", mock(BrokerLookupData.class),
+ "broker2", mock(BrokerLookupData.class)
+ ))).when(brokerRegistry).getAvailableBrokerLookupDataAsync();
+
+ var topBundlesLoadDataStore = ctx.topBundleLoadDataStore();
+ topBundlesLoadDataStore.pushAsync("broker1", getTopBundlesLoad("my-tenant/my-namespaceA", 1000000, 2000000, 3000000, 4000000, 5000000));
+ topBundlesLoadDataStore.pushAsync("broker2", getTopBundlesLoad("my-tenant/my-namespaceB", 490000000, 510000000));
+
+
+ var brokerLoadDataStore = ctx.brokerLoadDataStore();
+ brokerLoadDataStore.pushAsync("broker1", getCpuLoad(ctx, 10, "broker1"));
+ brokerLoadDataStore.pushAsync("broker2", getCpuLoad(ctx, 1000, "broker2"));
+
+
+ var res = transferShedder.findBundlesForUnloading(ctx, Map.of(), Map.of());
+ var expected = new HashSet<UnloadDecision>();
+ expected.add(new UnloadDecision(
+ new Unload("broker2", "my-tenant/my-namespaceB/0x00000000_0x0FFFFFFF", Optional.of("broker1")),
+ Success, Overloaded));
+ assertEquals(counter.getLoadAvg(), 5.05);
+ assertEquals(counter.getLoadStd(), 4.95);
+ assertEquals(res, expected);
+ var stats = (TransferShedder.LoadStats)
+ FieldUtils.readDeclaredField(transferShedder, "stats", true);
+ assertEquals(stats.std(), 0.050000004900021836);
+
+ }
+
+
+ @Test
+ public void testUnloadBundlesGreaterThanTargetThroughputAfterSplit() throws IllegalAccessException {
+ UnloadCounter counter = new UnloadCounter();
+ TransferShedder transferShedder = new TransferShedder(counter);
+ var ctx = getContext();
+
+ var brokerRegistry = mock(BrokerRegistry.class);
+ doReturn(brokerRegistry).when(ctx).brokerRegistry();
+ doReturn(CompletableFuture.completedFuture(Map.of(
+ "broker1", mock(BrokerLookupData.class),
+ "broker2", mock(BrokerLookupData.class)
+ ))).when(brokerRegistry).getAvailableBrokerLookupDataAsync();
+
+ var topBundlesLoadDataStore = ctx.topBundleLoadDataStore();
+ topBundlesLoadDataStore.pushAsync("broker1", getTopBundlesLoad("my-tenant/my-namespaceA", 2400000, 2400000));
+ topBundlesLoadDataStore.pushAsync("broker2", getTopBundlesLoad("my-tenant/my-namespaceB", 5000000, 5000000));
+
+
+ var brokerLoadDataStore = ctx.brokerLoadDataStore();
+ brokerLoadDataStore.pushAsync("broker1", getCpuLoad(ctx, 48, "broker1"));
+ brokerLoadDataStore.pushAsync("broker2", getCpuLoad(ctx, 100, "broker2"));
+ var res = transferShedder.findBundlesForUnloading(ctx, Map.of(), Map.of());
+
+ var expected = new HashSet<UnloadDecision>();
+ expected.add(new UnloadDecision(
+ new Unload("broker1",
+ res.stream().filter(x -> x.getUnload().sourceBroker().equals("broker1")).findFirst().get()
+ .getUnload().serviceUnit(), Optional.of("broker2")),
+ Success, Overloaded));
+ expected.add(new UnloadDecision(
+ new Unload("broker2",
+ res.stream().filter(x -> x.getUnload().sourceBroker().equals("broker2")).findFirst().get()
+ .getUnload().serviceUnit(), Optional.of("broker1")),
+ Success, Overloaded));
+ assertEquals(counter.getLoadAvg(), 0.74);
+ assertEquals(counter.getLoadStd(), 0.26);
+ assertEquals(res, expected);
+ var stats = (TransferShedder.LoadStats)
+ FieldUtils.readDeclaredField(transferShedder, "stats", true);
+ assertEquals(stats.std(), 2.5809568279517847E-8);
+ }
+
+
@Test
public void testMinBrokerWithZeroTraffic() throws IllegalAccessException {
UnloadCounter counter = new UnloadCounter();
TransferShedder transferShedder = new TransferShedder(counter);
var ctx = setupContext();
var brokerLoadDataStore = ctx.brokerLoadDataStore();
- brokerLoadDataStore.pushAsync("broker4", getCpuLoad(ctx, 55));
- brokerLoadDataStore.pushAsync("broker5", getCpuLoad(ctx, 65));
- var load = getCpuLoad(ctx, 4);
- FieldUtils.writeDeclaredField(load,"msgThroughputIn", 0, true);
- FieldUtils.writeDeclaredField(load,"msgThroughputOut", 0, true);
+ var load = getCpuLoad(ctx, 4, "broker2");
+ FieldUtils.writeDeclaredField(load,"msgThroughputEMA", 0, true);
brokerLoadDataStore.pushAsync("broker2", load);
+ brokerLoadDataStore.pushAsync("broker4", getCpuLoad(ctx, 55, "broker4"));
+ brokerLoadDataStore.pushAsync("broker5", getCpuLoad(ctx, 65, "broker5"));
var res = transferShedder.findBundlesForUnloading(ctx, Map.of(), Map.of());
@@ -714,13 +958,63 @@ public class TransferShedderTest {
assertEquals(counter.getLoadStd(), 0.27644891028904417);
}
+ @Test
+ public void testMinBrokerWithLowerLoadThanAvg() throws IllegalAccessException {
+ UnloadCounter counter = new UnloadCounter();
+ TransferShedder transferShedder = new TransferShedder(counter);
+ var ctx = setupContext();
+ var brokerLoadDataStore = ctx.brokerLoadDataStore();
+
+ var load = getCpuLoad(ctx, 3 , "broker2");
+ brokerLoadDataStore.pushAsync("broker2", load);
+ brokerLoadDataStore.pushAsync("broker4", getCpuLoad(ctx, 55, "broker4"));
+ brokerLoadDataStore.pushAsync("broker5", getCpuLoad(ctx, 65, "broker5"));
+
+ var res = transferShedder.findBundlesForUnloading(ctx, Map.of(), Map.of());
+
+ var expected = new HashSet<UnloadDecision>();
+ expected.add(new UnloadDecision(new Unload("broker5", bundleE1, Optional.of("broker1")),
+ Success, Overloaded));
+ expected.add(new UnloadDecision(new Unload("broker4", bundleD1, Optional.of("broker2")),
+ Success, Underloaded));
+ assertEquals(res, expected);
+ assertEquals(counter.getLoadAvg(), 0.262);
+ assertEquals(counter.getLoadStd(), 0.2780935094532054);
+ }
+
@Test
public void testMaxNumberOfTransfersPerShedderCycle() {
UnloadCounter counter = new UnloadCounter();
TransferShedder transferShedder = new TransferShedder(counter);
var ctx = setupContext();
ctx.brokerConfiguration()
- .setLoadBalancerMaxNumberOfBrokerTransfersPerCycle(10);
+ .setLoadBalancerMaxNumberOfBrokerSheddingPerCycle(10);
+ var res = transferShedder.findBundlesForUnloading(ctx, Map.of(), Map.of());
+ var expected = new HashSet<UnloadDecision>();
+ expected.add(new UnloadDecision(new Unload("broker5", bundleE1, Optional.of("broker1")),
+ Success, Overloaded));
+ expected.add(new UnloadDecision(new Unload("broker4", bundleD1, Optional.of("broker2")),
+ Success, Overloaded));
+ assertEquals(res, expected);
+ assertEquals(counter.getLoadAvg(), setupLoadAvg);
+ assertEquals(counter.getLoadStd(), setupLoadStd);
+ }
+
+ @Test
+ public void testLoadBalancerSheddingConditionHitCountThreshold() {
+ UnloadCounter counter = new UnloadCounter();
+ TransferShedder transferShedder = new TransferShedder(counter);
+ var ctx = setupContext();
+ int max = 3;
+ ctx.brokerConfiguration()
+ .setLoadBalancerSheddingConditionHitCountThreshold(max);
+ for (int i = 0; i < max; i++) {
+ var res = transferShedder.findBundlesForUnloading(ctx, Map.of(), Map.of());
+ assertTrue(res.isEmpty());
+ assertEquals(counter.getBreakdownCounters().get(Skip).get(HitCount).get(), i+1);
+ assertEquals(counter.getLoadAvg(), setupLoadAvg);
+ assertEquals(counter.getLoadStd(), setupLoadStd);
+ }
var res = transferShedder.findBundlesForUnloading(ctx, Map.of(), Map.of());
var expected = new HashSet<UnloadDecision>();
expected.add(new UnloadDecision(new Unload("broker5", bundleE1, Optional.of("broker1")),
@@ -738,7 +1032,7 @@ public class TransferShedderTest {
TransferShedder transferShedder = new TransferShedder(counter);
var ctx = setupContext();
var topBundlesLoadDataStore = ctx.topBundleLoadDataStore();
- topBundlesLoadDataStore.pushAsync("broker5", getTopBundlesLoad("my-tenant/my-namespaceE", 3000000, 2000000));
+ topBundlesLoadDataStore.pushAsync("broker5", getTopBundlesLoad("my-tenant/my-namespaceE", 2000000, 3000000));
var res = transferShedder.findBundlesForUnloading(ctx, Map.of(), Map.of());
var expected = new HashSet<UnloadDecision>();
@@ -751,6 +1045,33 @@ public class TransferShedderTest {
assertEquals(counter.getLoadStd(), setupLoadStd);
}
+ @Test
+ public void testLoadMoreThan100() throws IllegalAccessException {
+ UnloadCounter counter = new UnloadCounter();
+ TransferShedder transferShedder = new TransferShedder(counter);
+ var ctx = setupContext();
+
+ var brokerLoadDataStore = ctx.brokerLoadDataStore();
+ brokerLoadDataStore.pushAsync("broker4", getCpuLoad(ctx, 200, "broker4"));
+ brokerLoadDataStore.pushAsync("broker5", getCpuLoad(ctx, 1000, "broker5"));
+ var res = transferShedder.findBundlesForUnloading(ctx, Map.of(), Map.of());
+
+ var expected = new HashSet<UnloadDecision>();
+ expected.add(new UnloadDecision(new Unload("broker5", bundleE1, Optional.of("broker1")),
+ Success, Overloaded));
+ expected.add(new UnloadDecision(new Unload("broker4", bundleD1, Optional.of("broker2")),
+ Success, Overloaded));
+ assertEquals(res, expected);
+ assertEquals(counter.getLoadAvg(), 2.4240000000000004);
+ assertEquals(counter.getLoadStd(), 3.8633332758124816);
+
+
+ var stats = (TransferShedder.LoadStats)
+ FieldUtils.readDeclaredField(transferShedder, "stats", true);
+ assertEquals(stats.avg(), 2.4240000000000004);
+ assertEquals(stats.std(), 2.781643776903451);
+ }
+
@Test
public void testRandomLoad() throws IllegalAccessException {
UnloadCounter counter = new UnloadCounter();
@@ -767,7 +1088,7 @@ public class TransferShedderTest {
}
@Test
- public void testLoadStats() {
+ public void testRandomLoadStats() {
int numBrokers = 10;
double delta = 0.0001;
for (int t = 0; t < 5; t++) {
@@ -776,8 +1097,13 @@ public class TransferShedderTest {
var loadStore = ctx.brokerLoadDataStore();
stats.setLoadDataStore(loadStore);
var conf = ctx.brokerConfiguration();
- stats.update(loadStore, Map.of(), conf);
double[] loads = new double[numBrokers];
+ final Map<String, BrokerLookupData> availableBrokers = new HashMap<>();
+ for (int i = 0; i < loads.length; i++) {
+ availableBrokers.put("broker" + i, mock(BrokerLookupData.class));
+ }
+ stats.update(loadStore, availableBrokers, Map.of(), conf);
+
var brokerLoadDataStore = ctx.brokerLoadDataStore();
for (int i = 0; i < loads.length; i++) {
loads[i] = loadStore.get("broker" + i).get().getWeightedMaxEMA();
@@ -785,16 +1111,16 @@ public class TransferShedderTest {
int i = 0;
int j = loads.length - 1;
Arrays.sort(loads);
- for (int k = 0; k < conf.getLoadBalancerMaxNumberOfBrokerTransfersPerCycle(); k++) {
+ for (int k = 0; k < conf.getLoadBalancerMaxNumberOfBrokerSheddingPerCycle(); k++) {
double minLoad = loads[i];
double maxLoad = loads[j];
double offload = (maxLoad - minLoad) / 2;
Mean mean = new Mean();
StandardDeviation std = new StandardDeviation(false);
assertEquals(minLoad,
- loadStore.get(stats.minBrokers().pollLast()).get().getWeightedMaxEMA());
+ loadStore.get(stats.peekMinBroker()).get().getWeightedMaxEMA());
assertEquals(maxLoad,
- loadStore.get(stats.maxBrokers().pollLast()).get().getWeightedMaxEMA());
+ loadStore.get(stats.pollMaxBroker()).get().getWeightedMaxEMA());
assertEquals(stats.totalBrokers(), numBrokers);
assertEquals(stats.avg(), mean.evaluate(loads), delta);
assertEquals(stats.std(), std.evaluate(loads), delta);
@@ -804,4 +1130,41 @@ public class TransferShedderTest {
}
}
}
+
+ @Test
+ public void testHighVarianceLoadStats() {
+ int[] loads = {1, 100, 100, 100, 100, 100, 100, 100, 100, 100, 100, 100, 100, 100, 100, 100, 100};
+ var ctx = getContext();
+ TransferShedder.LoadStats stats = new TransferShedder.LoadStats();
+ var loadStore = ctx.brokerLoadDataStore();
+ stats.setLoadDataStore(loadStore);
+ var conf = ctx.brokerConfiguration();
+ final Map<String, BrokerLookupData> availableBrokers = new HashMap<>();
+ for (int i = 0; i < loads.length; i++) {
+ availableBrokers.put("broker" + i, mock(BrokerLookupData.class));
+ loadStore.pushAsync("broker" + i, getCpuLoad(ctx, loads[i], "broker" + i));
+ }
+ stats.update(loadStore, availableBrokers, Map.of(), conf);
+
+ assertEquals(stats.avg(), 0.9417647058823528);
+ assertEquals(stats.std(), 0.23294117647058868);
+ }
+
+ @Test
+ public void testLowVarianceLoadStats() {
+ int[] loads = {390, 391, 392, 393, 394, 395, 396, 397, 398, 399};
+ var ctx = getContext();
+ TransferShedder.LoadStats stats = new TransferShedder.LoadStats();
+ var loadStore = ctx.brokerLoadDataStore();
+ stats.setLoadDataStore(loadStore);
+ var conf = ctx.brokerConfiguration();
+ final Map<String, BrokerLookupData> availableBrokers = new HashMap<>();
+ for (int i = 0; i < loads.length; i++) {
+ availableBrokers.put("broker" + i, mock(BrokerLookupData.class));
+ loadStore.pushAsync("broker" + i, getCpuLoad(ctx, loads[i], "broker" + i));
+ }
+ stats.update(loadStore, availableBrokers, Map.of(), conf);
+ assertEquals(stats.avg(), 3.9449999999999994);
+ assertEquals(stats.std(), 0.028722813232795824);
+ }
}
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/extensions/scheduler/UnloadSchedulerTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/extensions/scheduler/UnloadSchedulerTest.java
index 7d9cf556360..38d4e9904e6 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/extensions/scheduler/UnloadSchedulerTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/extensions/scheduler/UnloadSchedulerTest.java
@@ -54,8 +54,8 @@ import java.util.concurrent.atomic.AtomicReference;
@Test(groups = "broker")
public class UnloadSchedulerTest {
-
- private ScheduledExecutorService loadManagerExecutor;
+ private PulsarService pulsar;
+ private ScheduledExecutorService loadManagerExecutor;
public LoadManagerContext setupContext(){
var ctx = getContext();
@@ -65,8 +65,12 @@ public class UnloadSchedulerTest {
@BeforeMethod
public void setUp() {
- this.loadManagerExecutor = Executors
- .newSingleThreadScheduledExecutor(new ExecutorProvider.ExtendedThreadFactory("pulsar-load-manager"));
+ this.pulsar = mock(PulsarService.class);
+ loadManagerExecutor = Executors
+ .newSingleThreadScheduledExecutor(new
+ ExecutorProvider.ExtendedThreadFactory("pulsar-load-manager"));
+ doReturn(loadManagerExecutor)
+ .when(pulsar).getLoadManagerExecutor();
}
@AfterMethod
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/extensions/strategy/DefaultNamespaceBundleSplitStrategyTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/extensions/strategy/DefaultNamespaceBundleSplitStrategyTest.java
index deff6fb00fa..0aa055b58ac 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/extensions/strategy/DefaultNamespaceBundleSplitStrategyTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/extensions/strategy/DefaultNamespaceBundleSplitStrategyTest.java
@@ -31,10 +31,15 @@ import static org.testng.Assert.assertEquals;
import java.util.LinkedHashMap;
import java.util.Map;
import java.util.Set;
+import java.util.concurrent.atomic.AtomicReference;
import org.apache.pulsar.broker.PulsarService;
import org.apache.pulsar.broker.ServiceConfiguration;
import org.apache.pulsar.broker.loadbalance.extensions.BrokerRegistry;
+import org.apache.pulsar.broker.loadbalance.extensions.ExtensibleLoadManagerImpl;
+import org.apache.pulsar.broker.loadbalance.extensions.ExtensibleLoadManagerWrapper;
import org.apache.pulsar.broker.loadbalance.extensions.LoadManagerContext;
+import org.apache.pulsar.broker.loadbalance.extensions.channel.ServiceUnitStateChannel;
+import org.apache.pulsar.broker.loadbalance.extensions.channel.ServiceUnitStateChannelImpl;
import org.apache.pulsar.broker.loadbalance.extensions.models.Split;
import org.apache.pulsar.broker.loadbalance.extensions.models.SplitDecision;
import org.apache.pulsar.broker.loadbalance.extensions.models.SplitCounter;
@@ -50,6 +55,9 @@ import org.testng.annotations.Test;
public class DefaultNamespaceBundleSplitStrategyTest {
PulsarService pulsar;
+ ExtensibleLoadManagerWrapper loadManagerWrapper;
+ ExtensibleLoadManagerImpl loadManager;
+ ServiceUnitStateChannel channel;
BrokerService brokerService;
PulsarStats pulsarStats;
Map<String, NamespaceBundleStats> bundleStats;
@@ -76,7 +84,7 @@ public class DefaultNamespaceBundleSplitStrategyTest {
config.setLoadBalancerNamespaceBundleMaxMsgRate(100);
config.setLoadBalancerNamespaceBundleMaxBandwidthMbytes(100);
config.setLoadBalancerMaxNumberOfBundlesToSplitPerCycle(1);
- config.setLoadBalancerNamespaceBundleSplitConditionThreshold(3);
+ config.setLoadBalancerNamespaceBundleSplitConditionHitCountThreshold(3);
pulsar = mock(PulsarService.class);
brokerService = mock(BrokerService.class);
@@ -85,8 +93,9 @@ public class DefaultNamespaceBundleSplitStrategyTest {
namespaceBundleFactory = mock(NamespaceBundleFactory.class);
loadManagerContext = mock(LoadManagerContext.class);
brokerRegistry = mock(BrokerRegistry.class);
-
-
+ loadManagerWrapper = mock(ExtensibleLoadManagerWrapper.class);
+ loadManager = mock(ExtensibleLoadManagerImpl.class);
+ channel = mock(ServiceUnitStateChannelImpl.class);
doReturn(brokerService).when(pulsar).getBrokerService();
doReturn(config).when(pulsar).getConfiguration();
@@ -96,6 +105,10 @@ public class DefaultNamespaceBundleSplitStrategyTest {
doReturn(true).when(namespaceBundleFactory).canSplitBundle(any());
doReturn(brokerRegistry).when(loadManagerContext).brokerRegistry();
doReturn(broker).when(brokerRegistry).getBrokerId();
+ doReturn(new AtomicReference(loadManagerWrapper)).when(pulsar).getLoadManager();
+ doReturn(loadManager).when(loadManagerWrapper).get();
+ doReturn(channel).when(loadManager).getServiceUnitStateChannel();
+ doReturn(true).when(channel).isOwner(any());
bundleStats = new LinkedHashMap<>();
@@ -109,7 +122,7 @@ public class DefaultNamespaceBundleSplitStrategyTest {
}
public void testNamespaceBundleSplitConditionThreshold() {
- config.setLoadBalancerNamespaceBundleSplitConditionThreshold(0);
+ config.setLoadBalancerNamespaceBundleSplitConditionHitCountThreshold(0);
bundleStats.values().forEach(v -> v.msgRateIn = config.getLoadBalancerNamespaceBundleMaxMsgRate() + 1);
var strategy = new DefaultNamespaceBundleSplitStrategyImpl(new SplitCounter());
var actual = strategy.findBundlesToSplit(loadManagerContext, pulsar);
@@ -118,7 +131,7 @@ public class DefaultNamespaceBundleSplitStrategyTest {
public void testNotEnoughTopics() {
- config.setLoadBalancerNamespaceBundleSplitConditionThreshold(0);
+ config.setLoadBalancerNamespaceBundleSplitConditionHitCountThreshold(0);
bundleStats.values().forEach(v -> v.msgRateIn = config.getLoadBalancerNamespaceBundleMaxMsgRate() + 1);
var strategy = new DefaultNamespaceBundleSplitStrategyImpl(new SplitCounter());
bundleStats.values().forEach(v -> v.topics = 1);
@@ -128,7 +141,7 @@ public class DefaultNamespaceBundleSplitStrategyTest {
}
public void testNamespaceMaximumBundles() throws Exception {
- config.setLoadBalancerNamespaceBundleSplitConditionThreshold(0);
+ config.setLoadBalancerNamespaceBundleSplitConditionHitCountThreshold(0);
bundleStats.values().forEach(v -> v.msgRateIn = config.getLoadBalancerNamespaceBundleMaxMsgRate() + 1);
var strategy = new DefaultNamespaceBundleSplitStrategyImpl(new SplitCounter());
doReturn(config.getLoadBalancerNamespaceMaximumBundles()).when(namespaceService).getBundleCount(any());
@@ -138,7 +151,7 @@ public class DefaultNamespaceBundleSplitStrategyTest {
}
public void testEmptyBundleStats() {
- config.setLoadBalancerNamespaceBundleSplitConditionThreshold(0);
+ config.setLoadBalancerNamespaceBundleSplitConditionHitCountThreshold(0);
bundleStats.values().forEach(v -> v.msgRateIn = config.getLoadBalancerNamespaceBundleMaxMsgRate() + 1);
var strategy = new DefaultNamespaceBundleSplitStrategyImpl(new SplitCounter());
bundleStats.clear();
@@ -147,9 +160,21 @@ public class DefaultNamespaceBundleSplitStrategyTest {
assertEquals(actual, expected);
}
+ public void testNoBundleOwner() {
+ var counter = spy(new SplitCounter());
+ config.setLoadBalancerNamespaceBundleSplitConditionHitCountThreshold(0);
+ bundleStats.values().forEach(v -> v.msgRateIn = config.getLoadBalancerNamespaceBundleMaxMsgRate() + 1);
+ doReturn(false).when(channel).isOwner(any());
+ var strategy = new DefaultNamespaceBundleSplitStrategyImpl(counter);
+ var actual = strategy.findBundlesToSplit(loadManagerContext, pulsar);
+ var expected = Set.of();
+ assertEquals(actual, expected);
+ verify(counter, times(2)).update(eq(SplitDecision.Label.Failure), eq(Unknown));
+ }
+
public void testError() throws Exception {
var counter = spy(new SplitCounter());
- config.setLoadBalancerNamespaceBundleSplitConditionThreshold(0);
+ config.setLoadBalancerNamespaceBundleSplitConditionHitCountThreshold(0);
bundleStats.values().forEach(v -> v.msgRateIn = config.getLoadBalancerNamespaceBundleMaxMsgRate() + 1);
var strategy = new DefaultNamespaceBundleSplitStrategyImpl(counter);
doThrow(new RuntimeException()).when(namespaceService).getBundleCount(any());
@@ -162,7 +187,7 @@ public class DefaultNamespaceBundleSplitStrategyTest {
public void testMaxMsgRate() {
var counter = spy(new SplitCounter());
var strategy = new DefaultNamespaceBundleSplitStrategyImpl(counter);
- int threshold = config.getLoadBalancerNamespaceBundleSplitConditionThreshold();
+ int threshold = config.getLoadBalancerNamespaceBundleSplitConditionHitCountThreshold();
bundleStats.values().forEach(v -> {
v.msgRateOut = config.getLoadBalancerNamespaceBundleMaxMsgRate() / 2 + 1;
v.msgRateIn = config.getLoadBalancerNamespaceBundleMaxMsgRate() / 2 + 1;
@@ -193,7 +218,7 @@ public class DefaultNamespaceBundleSplitStrategyTest {
public void testMaxTopics() {
var counter = spy(new SplitCounter());
var strategy = new DefaultNamespaceBundleSplitStrategyImpl(counter);
- int threshold = config.getLoadBalancerNamespaceBundleSplitConditionThreshold();
+ int threshold = config.getLoadBalancerNamespaceBundleSplitConditionHitCountThreshold();
bundleStats.values().forEach(v -> v.topics = config.getLoadBalancerNamespaceBundleMaxTopics() + 1);
for (int i = 0; i < threshold + 2; i++) {
var actual = strategy.findBundlesToSplit(loadManagerContext, pulsar);
@@ -221,7 +246,7 @@ public class DefaultNamespaceBundleSplitStrategyTest {
public void testMaxSessions() {
var counter = spy(new SplitCounter());
var strategy = new DefaultNamespaceBundleSplitStrategyImpl(counter);
- int threshold = config.getLoadBalancerNamespaceBundleSplitConditionThreshold();
+ int threshold = config.getLoadBalancerNamespaceBundleSplitConditionHitCountThreshold();
bundleStats.values().forEach(v -> {
v.producerCount = config.getLoadBalancerNamespaceBundleMaxSessions() / 2 + 1;
v.consumerCount = config.getLoadBalancerNamespaceBundleMaxSessions() / 2 + 1;
@@ -252,7 +277,7 @@ public class DefaultNamespaceBundleSplitStrategyTest {
public void testMaxBandwidthMbytes() {
var counter = spy(new SplitCounter());
var strategy = new DefaultNamespaceBundleSplitStrategyImpl(counter);
- int threshold = config.getLoadBalancerNamespaceBundleSplitConditionThreshold();
+ int threshold = config.getLoadBalancerNamespaceBundleSplitConditionHitCountThreshold();
bundleStats.values().forEach(v -> {
v.msgThroughputOut = config.getLoadBalancerNamespaceBundleMaxBandwidthMbytes() * 1024 * 1024 / 2 + 1;
v.msgThroughputIn = config.getLoadBalancerNamespaceBundleMaxBandwidthMbytes() * 1024 * 1024 / 2 + 1;
diff --git a/pulsar-client-admin-api/src/main/java/org/apache/pulsar/policies/data/loadbalancer/NamespaceBundleStats.java b/pulsar-client-admin-api/src/main/java/org/apache/pulsar/policies/data/loadbalancer/NamespaceBundleStats.java
index dc1cdd3d5cb..cd364990842 100644
--- a/pulsar-client-admin-api/src/main/java/org/apache/pulsar/policies/data/loadbalancer/NamespaceBundleStats.java
+++ b/pulsar-client-admin-api/src/main/java/org/apache/pulsar/policies/data/loadbalancer/NamespaceBundleStats.java
@@ -20,10 +20,12 @@ package org.apache.pulsar.policies.data.loadbalancer;
import java.io.Serializable;
import lombok.EqualsAndHashCode;
+import lombok.ToString;
/**
*/
@EqualsAndHashCode
+@ToString
public class NamespaceBundleStats implements Comparable<NamespaceBundleStats>, Serializable {
public double msgRateIn;
diff --git a/pulsar-client-admin-api/src/main/java/org/apache/pulsar/policies/data/loadbalancer/ResourceUsage.java b/pulsar-client-admin-api/src/main/java/org/apache/pulsar/policies/data/loadbalancer/ResourceUsage.java
index eed8b33b391..ae12e931071 100644
--- a/pulsar-client-admin-api/src/main/java/org/apache/pulsar/policies/data/loadbalancer/ResourceUsage.java
+++ b/pulsar-client-admin-api/src/main/java/org/apache/pulsar/policies/data/loadbalancer/ResourceUsage.java
@@ -19,11 +19,13 @@
package org.apache.pulsar.policies.data.loadbalancer;
import lombok.EqualsAndHashCode;
+import lombok.ToString;
/**
* POJO used to represent any system specific resource usage this is the format that load manager expects it in.
*/
@EqualsAndHashCode
+@ToString
public class ResourceUsage {
public final double usage;
public final double limit;