[pulsar] branch branch-2.9 updated: [improve][broker] Avoid go through all the consumers to get the message ack owner (#16245)

[bogong]

This is an automated email from the ASF dual-hosted git repository.

bogong pushed a commit to branch branch-2.9
in repository https://gitbox.apache.org/repos/asf/pulsar.git


The following commit(s) were added to refs/heads/branch-2.9 by this push:
     new 95f3b06cfac [improve][broker] Avoid go through all the consumers to 
get the message ack owner (#16245)
95f3b06cfac is described below

commit 95f3b06cfacae71f1d48b051e985ee14eb0cc7b6
Author: lipenghui <peng...@apache.org>
AuthorDate: Tue Jun 28 09:19:18 2022 +0800

    [improve][broker] Avoid go through all the consumers to get the message ack 
owner (#16245)
    
    ### Motivation
    
    The broker don't need to go through all the consumers to get the ack owner 
consumer.
    Instead, it should check the current consumer first. If the pending acks of 
current consumer
    don't have the ack position, go through all the consumers to find the owner 
consumer.
    
    (cherry picked from commit 68484f9162bc768816cfd039140fb78196485d19)
---
 .../main/java/org/apache/pulsar/broker/service/Consumer.java   | 10 ++++++----
 1 file changed, 6 insertions(+), 4 deletions(-)

diff --git 
a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Consumer.java 
b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Consumer.java
index 297aefee6e8..25dd3b908e5 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Consumer.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Consumer.java
@@ -612,10 +612,12 @@ public class Consumer {
     private Consumer getAckOwnerConsumer(long ledgerId, long entryId) {
         Consumer ackOwnerConsumer = this;
         if (Subscription.isIndividualAckMode(subType)) {
-            for (Consumer consumer : subscription.getConsumers()) {
-                if (consumer != this && 
consumer.getPendingAcks().containsKey(ledgerId, entryId)) {
-                    ackOwnerConsumer = consumer;
-                    break;
+            if (!getPendingAcks().containsKey(ledgerId, entryId)) {
+                for (Consumer consumer : subscription.getConsumers()) {
+                    if (consumer != this && 
consumer.getPendingAcks().containsKey(ledgerId, entryId)) {
+                        ackOwnerConsumer = consumer;
+                        break;
+                    }
                 }
             }
         }