[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 0844f3153d2 [improve][broker] Avoid go through all the consumers to 
get the message ack owner (#16245)
0844f3153d2 is described below

commit 0844f3153d22546780cd36a5c6610b7a382bdcbc
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;
+                    }
                 }
             }
         }