[Bug] Pulsar broker CPU stratification problem about pulsar

[lhotari]

@AuroraTwinkle In the future, please create such questions in GitHub Discussions at https://github.com/apache/pulsar/discussions/categories/q-a . It's possible to migrate a question to an issue if that's needed.

[AuroraTwinkle]

@AuroraTwinkle In the future, please create such questions in GitHub Discussions at https://github.com/apache/pulsar/discussions/categories/q-a . It's possible to migrate a question to an issue if that's needed.

Okay, got it!

[AuroraTwinkle]

I'm not sure if this is a bug, so just close it first.

[lhotari]

I'm not sure if this is a bug, so just close it first.

Most likely this could be considered a "feature" and not a bug. There are a lot of implementation level details that are undocumented. For example in this case, looking at the source code of ThresholdShedder could help:

pulsar/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/impl/ThresholdShedder.java

Lines 77 to 120 in 9692b67

	loadData.getBrokerData().forEach((broker, brokerData) -> {
	final LocalBrokerData localData = brokerData.getLocalData();
	final double currentUsage = brokerAvgResourceUsage.getOrDefault(broker, 0.0);
	if (currentUsage < avgUsage + threshold) {
	if (log.isDebugEnabled()) {
	log.debug("[{}] broker is not overloaded, ignoring at this point ({})", broker,
	localData.printResourceUsage());
	}
	return;
	}
	double percentOfTrafficToOffload =
	currentUsage - avgUsage - threshold + ADDITIONAL_THRESHOLD_PERCENT_MARGIN;
	double brokerCurrentThroughput = localData.getMsgThroughputIn() + localData.getMsgThroughputOut();
	double minimumThroughputToOffload = brokerCurrentThroughput * percentOfTrafficToOffload;
	if (minimumThroughputToOffload < minThroughputThreshold) {
	if (log.isDebugEnabled()) {
	log.debug("[{}] broker is planning to shed throughput {} MByte/s less than "
	+ "minimumThroughputThreshold {} MByte/s, skipping bundle unload ({})",
	broker, minimumThroughputToOffload / MB, minThroughputThreshold / MB,
	localData.printResourceUsage());
	}
	return;
	}
	log.info(
	"Attempting to shed load on {}, which has max resource usage above avgUsage and threshold {}%"
	+ " > {}% + {}% -- Offloading at least {} MByte/s of traffic,"
	+ " left throughput {} MByte/s ({})",
	broker, 100 * currentUsage, 100 * avgUsage, 100 * threshold, minimumThroughputToOffload / MB,
	(brokerCurrentThroughput - minimumThroughputToOffload) / MB, localData.printResourceUsage());
	if (localData.getBundles().size() > 1) {
	filterAndSelectBundle(loadData, recentlyUnloadedBundles, broker, localData, minimumThroughputToOffload);
	} else if (localData.getBundles().size() == 1) {
	log.warn(
	"HIGH USAGE WARNING : Sole namespace bundle {} is overloading broker {}. "
	+ "No Load Shedding will be done on this broker",
	localData.getBundles().iterator().next(), broker);
	} else {
	log.warn("Broker {} is overloaded despite having no bundles", broker);
	}
	});

Enabling debug logging will be useful in following the decisions that were made.

[AuroraTwinkle]

I'm not sure if this is a bug, so just close it first.

Most likely this could be considered a "feature" and not a bug. There are a lot of implementation level details that are undocumented. For example in this case, looking at the source code of ThresholdShedder could help:

pulsar/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/impl/ThresholdShedder.java

Lines 77 to 120 in 9692b67

	loadData.getBrokerData().forEach((broker, brokerData) -> {
	final LocalBrokerData localData = brokerData.getLocalData();
	final double currentUsage = brokerAvgResourceUsage.getOrDefault(broker, 0.0);
	if (currentUsage < avgUsage + threshold) {
	if (log.isDebugEnabled()) {
	log.debug("[{}] broker is not overloaded, ignoring at this point ({})", broker,
	localData.printResourceUsage());
	}
	return;
	}
	double percentOfTrafficToOffload =
	currentUsage - avgUsage - threshold + ADDITIONAL_THRESHOLD_PERCENT_MARGIN;
	double brokerCurrentThroughput = localData.getMsgThroughputIn() + localData.getMsgThroughputOut();
	double minimumThroughputToOffload = brokerCurrentThroughput * percentOfTrafficToOffload;
	if (minimumThroughputToOffload < minThroughputThreshold) {
	if (log.isDebugEnabled()) {
	log.debug("[{}] broker is planning to shed throughput {} MByte/s less than "
	+ "minimumThroughputThreshold {} MByte/s, skipping bundle unload ({})",
	broker, minimumThroughputToOffload / MB, minThroughputThreshold / MB,
	localData.printResourceUsage());
	}
	return;
	}
	log.info(
	"Attempting to shed load on {}, which has max resource usage above avgUsage and threshold {}%"
	+ " > {}% + {}% -- Offloading at least {} MByte/s of traffic,"
	+ " left throughput {} MByte/s ({})",
	broker, 100 * currentUsage, 100 * avgUsage, 100 * threshold, minimumThroughputToOffload / MB,
	(brokerCurrentThroughput - minimumThroughputToOffload) / MB, localData.printResourceUsage());
	if (localData.getBundles().size() > 1) {
	filterAndSelectBundle(loadData, recentlyUnloadedBundles, broker, localData, minimumThroughputToOffload);
	} else if (localData.getBundles().size() == 1) {
	log.warn(
	"HIGH USAGE WARNING : Sole namespace bundle {} is overloading broker {}. "
	+ "No Load Shedding will be done on this broker",
	localData.getBundles().iterator().next(), broker);
	} else {
	log.warn("Broker {} is overloaded despite having no bundles", broker);
	}
	});

Enabling debug logging will be useful in following the decisions that were made.

Okay, thanks!