[improve][broker] Optimize the performance of individual acknowledgments

[shibd]

Motivation

If a subscription has many consumers and each consumer individually acknowledges messages from other consumers, this scenario is very CPU resource-intensive.

Refer this:

In fact, these three methods achieve the same goal: get owning consumer by message ID
This operation is very CPU resource-intensive

getAckOwnerConsumer:

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

Lines 735 to 748 in d7e8ea1

	private Consumer getAckOwnerConsumer(long ledgerId, long entryId) {
	Consumer ackOwnerConsumer = this;
	if (Subscription.isIndividualAckMode(subType)) {
	if (!getPendingAcks().containsKey(ledgerId, entryId)) {
	for (Consumer consumer : subscription.getConsumers()) {
	if (consumer != this && consumer.getPendingAcks().containsKey(ledgerId, entryId)) {
	ackOwnerConsumer = consumer;
	break;
	}
	}
	}
	}
	return ackOwnerConsumer;
	}

getBatchSize:

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

Line 657 in d7e8ea1

Consumer ackOwnerConsumer = getAckOwnerConsumer(msgId.getLedgerId(), msgId.getEntryId());

removePendingAcks:

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

Lines 1024 to 1034 in d7e8ea1

	if (pendingAcks.get(position.getLedgerId(), position.getEntryId()) == null) {
	for (Consumer consumer : subscription.getConsumers()) {
	if (!consumer.equals(this) && consumer.getPendingAcks().containsKey(position.getLedgerId(),
	position.getEntryId())) {
	ackOwnedConsumer = consumer;
	break;
	}
	}
	} else {
	ackOwnedConsumer = this;
	}

Modifications

This PR ensures that the getAckOwnerConsumer method is called only once during the processing of a message ack. In the future, we may need to optimize this method further.

After this PR:

ATT: Client Side Test Code: https://gist.github.com/shibd/82b77eccc8145ac2c1784aca97a29c24

Verifying this change

All tests passed to make sure not have any break change.

Documentation

• doc
• doc-required
• doc-not-needed
• doc-complete

[codecov-commenter]

Codecov Report

Attention: Patch coverage is 91.30435% with 4 lines in your changes missing coverage. Please review.

Project coverage is 73.44%. Comparing base (bbc6224) to head (679ac08).
Report is 474 commits behind head on master.

Additional details and impacted files

@@             Coverage Diff              @@
##             master   #23072      +/-   ##
============================================
- Coverage     73.57%   73.44%   -0.13%     
- Complexity    32624    33503     +879     
============================================
  Files          1877     1917      +40     
  Lines        139502   144051    +4549     
  Branches      15299    15734     +435     
============================================
+ Hits         102638   105803    +3165     
- Misses        28908    30127    +1219     
- Partials       7956     8121     +165     

Flag	Coverage Δ
inttests	27.49% <67.39%> (+2.90%)	⬆️
systests	24.75% <41.30%> (+0.43%)	⬆️
unittests	72.50% <91.30%> (-0.34%)	⬇️

Flags with carried forward coverage won't be shown. Click here to find out more.

Files	Coverage Δ
...ava/org/apache/pulsar/broker/service/Consumer.java	86.65% <91.30%> (-0.02%)	⬇️

... and 511 files with indirect coverage changes

[lhotari]

Great catch @shibd . Added a suggestion about renaming a method & variable since the meaning has changed.