[Bug] Consumption unevenness arises in the consumption performance pressure test.

[g0715158]

Search before reporting

• I searched in the issues and found nothing similar.

Read release policy

• I understand that unsupported versions don't get bug fixes. I will attempt to reproduce the issue on a supported version of Pulsar client and Pulsar broker.

User environment

3.0.10
jdk17
client 3.0.10

Issue Description

When conducting consumption performance pressure tests with a relatively large number of consumers per partition (over 1000), consumption imbalance will occur: the consumption speed of some partitions drops significantly, the overall throughput declines, and message backlog eventually arises in certain partitions.

Error messages

no error

Reproducing the issue

When conducting consumption performance pressure tests with a relatively large number of consumers per partition (over 1000), consumption imbalance will occur: the consumption speed of some partitions drops significantly, the overall throughput declines, and message backlog eventually arises in certain partitions.

Additional information

No response

Are you willing to submit a PR?

• I'm willing to submit a PR!

[lhotari]

Reproducing the issue

When conducting consumption performance pressure tests with a relatively large number of consumers per partition (over 1000), consumption imbalance will occur: the consumption speed of some partitions drops significantly, the overall throughput declines, and message backlog eventually arises in certain partitions.

It would be useful to provide more practical details.

[lhotari]

3.0.10
jdk17
client 3.0.10

@g0715158 Please retest with Pulsar 4.1.2 version, running on Java 21.0.9 (or newer). Pulsar 4.1.x comes with "PIP-430: Pulsar Broker Cache Improvements: Refactoring Eviction and Adding a New Cache Strategy Based on Expected Read Count" which improves performance significantly in the scenario you have described.

Before PIP-430, when consumers "drop off the tail", the entries are no longer retrieved from cache, but have to be retrieved all the way from BookKeeper. This adds additional load and latency to the system. In PIP-430, you would still need to ensure that the broker cache is large enough to hold the in progress entries (batch messages). For a large number of partitions, you need a large broker cache size (configured with managedLedgerCacheSizeMB). You should also set managedLedgerMaxReadsInFlightSizeInMB to a reasonable amount.

You didn't mention how many subscriptions you have. Are there a lot of consumers on a single shared subscription, so that each partition has one shared subscription?

[g0715158]

@lhotari Currently, my goal is to identify the root cause of partition imbalance in version 3.0.10. Approximately 1000 consumers per partition experience performance degradation after consuming for about 9 minutes, and partition imbalance can be observed from the console. I would like to ask if such a situation has occurred in any existing issues?

[lhotari]

Currently, my goal is to identify the root cause of partition imbalance in version 3.0.10.

@g0715158 In the OSS project, we don't maintain specific versions such as 3.0.10. 3.0.x continues to be maintained, but the latest released version is 3.0.15 .
For this case, please attempt to reproduce with 4.1.2 as I suggested before. If you cannot reproduce, that's a lot more information for you to identify the root cause in version 3.0.10.

Approximately 1000 consumers per partition experience performance degradation after consuming for about 9 minutes, and partition imbalance can be observed from the console. I would like to ask if such a situation has occurred in any existing issues?

Yes, I've seen that happen. The imbalance is common, but the case where consuming stops completely might be a different issue such as #24926 .

However, in the stats that you shared, there are many cases where the backlog is 0 for the partitions that have out rate of 0.
Is this test case of a scenario where producers are producing actively and consumers are following? Or is it a "catch-up scenario" where there's existing backlog which consumers consume.

In your test scenario, you didn't mention anything about the client side. How many separate client instances and/or client connections do you have? How well is the client side tuned? For example, https://pulsar.apache.org/docs/next/client-libraries-java-setup/#java-client-performance ?

When you are creating a large number of Java client instances in a single JVM, it's necessary to share resources. There's an example in branch-4.0 in this test: https://github.com/apache/pulsar/blob/branch-4.0/pulsar-broker/src/test/java/org/apache/pulsar/client/api/PatternConsumerBackPressureMultipleConsumersTest.java#L238-L280 .
For 4.1+, there's PIP-234 PulsarClientSharedResources:

pulsar/pulsar-broker/src/test/java/org/apache/pulsar/client/api/PatternConsumerBackPressureMultipleConsumersTest.java

Lines 103 to 123 in 270120c

	@Cleanup
	PulsarClientSharedResources sharedResources =
	PulsarClientSharedResources.builder().build();
	List<PulsarClientImpl> clients = new ArrayList<>(numberOfClients);
	@Cleanup
	Closeable closeClients = () -> {
	for (PulsarClient client : clients) {
	try {
	client.close();
	} catch (PulsarClientException e) {
	log.error("Failed to close client {}", client, e);
	}
	}
	};
	for (int i = 0; i < numberOfClients; i++) {
	PulsarClientImpl client = (PulsarClientImpl) PulsarClient.builder()
	.serviceUrl(getClientServiceUrl())
	.sharedResources(sharedResources)
	.build();
	clients.add(client);
	}

If you are actively producing to partitions in a test case, one common issue for test scenarios is the producing side. It's also possible that the producing side doesn't produce evenly across partitions. One way to solve this is to produce individually to specific partitions (*-partition-0, *-partition-1, ...) in the load generator and have a sufficient amount of separate nodes for producing the messages so that the bottleneck isn't in producing clients.
On the producer side, using a multi-topic (partitioned) producer will also have more variance across partitions due to the default use of https://github.com/apache/pulsar/blob/master/pulsar-client/src/main/java/org/apache/pulsar/client/impl/RoundRobinPartitionMessageRouterImpl.java .
The setting can be controlled with

pulsar/pulsar-client-api/src/main/java/org/apache/pulsar/client/api/ProducerBuilder.java

Lines 462 to 474 in cc5e479

	/**
	* Set the partition switch frequency while batching of messages is enabled and
	* using round-robin routing mode for non-keyed messages. <i>Default: 10</i>.
	*
	* <p>The time period of partition switch is frequency * batchingMaxPublishDelay. During this period,
	* all arriving messages will be routed to the same partition.
	*
	* @param frequency the frequency of partition switch
	* @return the producer builder instance
	* @see #messageRoutingMode(MessageRoutingMode)
	* @see #batchingMaxPublishDelay(long, TimeUnit)
	*/
	ProducerBuilder<T> roundRobinRouterBatchingPartitionSwitchFrequency(int frequency);

.
The defaults are time based for both routing and batching. For testing purposes, it could be better to use a count based routing if partitioned producer is used and configure batching with a long batchingMaxPublishDelay and use batchingMaxMessages to achieve similar sized batches each time.

[lhotari]

In dispatching, I have had some doubts of the source of imbalance in this logic:

pulsar/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractDispatcherMultipleConsumers.java

Lines 77 to 226 in ea56ada

	/**
	* <pre>
	* Broker gives more priority while dispatching messages. Here, broker follows descending priorities. (eg:
	* 0=max-priority, 1, 2,..)
	* <p>
	* Broker will first dispatch messages to max priority-level consumers if they
	* have permits, else broker will consider next priority level consumers.
	* Also on the same priority-level, it selects consumer in round-robin manner.
	* <p>
	* If subscription has consumer-A with priorityLevel 1 and Consumer-B with priorityLevel 2
	* then broker will dispatch
	* messages to only consumer-A until it runs out permit and then broker starts dispatching messages to Consumer-B.
	* <p>
	* Consumer PriorityLevel Permits
	* C1 0 2
	* C2 0 1
	* C3 0 1
	* C4 1 2
	* C5 1 1
	* Result of getNextConsumer(): C1, C2, C3, C1, C4, C5, C4
	* </pre>
	*
	* <pre>
	* <b>Algorithm:</b>
	* 1. consumerList: it stores consumers in sorted-list: max-priority stored first
	* 2. currentConsumerRoundRobinIndex: it always stores last served consumer-index
	*
	* Each time getNextConsumer() is called:<p>
	* 1. It always starts to traverse from the max-priority consumer (first element) from sorted-list
	* 2. Consumers on same priority-level will be treated equally and it tries to pick one of them in
	* round-robin manner
	* 3. If consumer is not available on given priority-level then only it will go to the next lower priority-level
	* consumers
	* 4. Returns null in case it doesn't find any available consumer
	* </pre>
	*
	* @return nextAvailableConsumer
	*/
	public Consumer getNextConsumer() {
	if (consumerList.isEmpty() || IS_CLOSED_UPDATER.get(this) == TRUE) {
	// abort read if no consumers are connected or if disconnect is initiated
	return null;
	}
	if (currentConsumerRoundRobinIndex >= consumerList.size()) {
	currentConsumerRoundRobinIndex = 0;
	}
	int currentRoundRobinConsumerPriority = consumerList.get(currentConsumerRoundRobinIndex).getPriorityLevel();
	// first find available-consumer on higher level unless currentIndex is not on highest level which is 0
	if (currentRoundRobinConsumerPriority != 0) {
	int higherPriorityConsumerIndex = getConsumerFromHigherPriority(currentRoundRobinConsumerPriority);
	if (higherPriorityConsumerIndex != -1) {
	currentConsumerRoundRobinIndex = higherPriorityConsumerIndex + 1;
	return consumerList.get(higherPriorityConsumerIndex);
	}
	}
	// currentIndex is already on highest level or couldn't find consumer on higher level so, find consumer on same
	// or lower level
	int availableConsumerIndex = getNextConsumerFromSameOrLowerLevel(currentConsumerRoundRobinIndex);
	if (availableConsumerIndex != -1) {
	currentConsumerRoundRobinIndex = availableConsumerIndex + 1;
	return consumerList.get(availableConsumerIndex);
	}
	// couldn't find available consumer
	return null;
	}
	/**
	* Get random consumer from consumerList.
	*
	* @return null if no consumer available, else return random consumer from consumerList
	*/
	public Consumer getRandomConsumer() {
	if (consumerList.isEmpty() || IS_CLOSED_UPDATER.get(this) == TRUE) {
	// abort read if no consumers are connected of if disconnect is initiated
	return null;
	}
	return consumerList.get(ThreadLocalRandom.current().nextInt(consumerList.size()));
	}
	/**
	* Finds index of first available consumer which has higher priority then given targetPriority.
	*
	* @param targetPriority
	* @return -1 if couldn't find any available consumer
	*/
	private int getConsumerFromHigherPriority(int targetPriority) {
	for (int i = 0; i < currentConsumerRoundRobinIndex; i++) {
	Consumer consumer = consumerList.get(i);
	if (consumer.getPriorityLevel() < targetPriority) {
	if (isConsumerAvailable(consumerList.get(i))) {
	return i;
	}
	} else {
	break;
	}
	}
	return -1;
	}
	/**
	* Finds index of round-robin available consumer that present on same level as consumer on
	* currentRoundRobinIndex if doesn't find consumer on same level then it finds first available consumer on lower
	* priority level else returns
	* index=-1 if couldn't find any available consumer in the list.
	*
	* @param currentRoundRobinIndex
	* @return
	*/
	private int getNextConsumerFromSameOrLowerLevel(int currentRoundRobinIndex) {
	Consumer currentRRConsumer = consumerList.get(currentRoundRobinIndex);
	if (isConsumerAvailable(currentRRConsumer)) {
	return currentRoundRobinIndex;
	}
	// scan the consumerList, if consumer in currentRoundRobinIndex is unavailable
	int targetPriority = currentRRConsumer.getPriorityLevel();
	int scanIndex = currentRoundRobinIndex + 1;
	int endPriorityLevelIndex = currentRoundRobinIndex;
	do {
	Consumer scanConsumer = scanIndex < consumerList.size() ? consumerList.get(scanIndex)
	: null /* reached to last consumer of list */;
	// if reached to last consumer of list then check from beginning to currentRRIndex of the list
	if (scanConsumer == null || scanConsumer.getPriorityLevel() != targetPriority) {
	endPriorityLevelIndex = scanIndex; // last consumer on this level
	scanIndex = getFirstConsumerIndexOfPriority(targetPriority);
	} else {
	if (isConsumerAvailable(scanConsumer)) {
	return scanIndex;
	}
	scanIndex++;
	}
	} while (scanIndex != currentRoundRobinIndex);
	// it means: didn't find consumer in the same priority-level so, check available consumer lower than this level
	for (int i = endPriorityLevelIndex; i < consumerList.size(); i++) {
	if (isConsumerAvailable(consumerList.get(i))) {
	return i;
	}
	}
	return -1;
	}

[g0715158]

@lhotari Thank you so much for your help! When I ran pressure tests with 1KB small messages, the performance was solid and stable—only partition imbalance occurred, with a TPS of around 250K. However, when consuming 500KB large messages, performance degrades after a few minutes, paired with severe partition consumption imbalance. The most critical issue is that the consumption rate metrics for partitions vanish entirely: neither pulsar-manager API queries nor the pulsar_rate_out metric returns any data (this affects all brokers). Other metrics appear normal, and no thread blockages were detected in the broker thread dumps. Additionally, no priority distinctions were configured for the consumers.

[lhotari]

@lhotari Thank you so much for your help! When I ran pressure tests with 1KB small messages, the performance was solid and stable—only partition imbalance occurred, with a TPS of around 250K. However, when consuming 500KB large messages, performance degrades after a few minutes, paired with severe partition consumption imbalance. The most critical issue is that the consumption rate metrics for partitions vanish entirely: neither pulsar-manager API queries nor the pulsar_rate_out metric returns any data (this affects all brokers). Other metrics appear normal, and no thread blockages were detected in the broker thread dumps. Additionally, no priority distinctions were configured for the consumers.

Have you configured managedLedgerMaxReadsInFlightSizeInMB to a reasonable value? The feature isn't enabled by default. It was added in 2.11 with #18245 and improved later in #23901 . There's some details about broker memory limits in PIP-442 https://github.com/apache/pulsar/blob/master/pip/pip-442.md#existing-broker-memory-management .
If you haven't configured managedLedgerMaxReadsInFlightSizeInMB, the broker can get overloaded with a large amount of consumers. You can try setting it to a value like 500 MB when you have at least 4GB of direct memory available.

[lhotari]

Other metrics appear normal, and no thread blockages were detected in the broker thread dumps.

What metrics are you checking? If you want to get proper performance, you should be optimizing broker cache hit rate.
There are dashboards such as https://github.com/lhotari/pulsar-grafana-dashboards/blob/master/pulsar/broker-cache.json and https://github.com/lhotari/pulsar-grafana-dashboards/blob/master/pulsar/broker-cache-by-broker.json for this purpose.

Before Pulsar 4.1 and PIP-430, there are workarounds for tuning the broker cache. #23466.

[lhotari]

Please also check #24695

[g0715158]

@lhotari We have not enabled this parameter as managedLedgerMaxReadsInFlightSizeInMB=0. Currently, the broker has 20GB of heap memory and 40GB of off-heap memory available. Would it be appropriate for us to set this parameter to 2GB?

[lhotari]

@lhotari We have not enabled this parameter as managedLedgerMaxReadsInFlightSizeInMB=0. Currently, the broker has 20GB of heap memory and 40GB of off-heap memory available. Would it be appropriate for us to set this parameter to 2GB?

Yes, I think around 2GB-4GB is reasonable in that case. (btw. The limit isn't an accurate limit and more direct memory will get used).