From 4daa5cf1b1d681257cefba9458a4b6458765ef24 Mon Sep 17 00:00:00 2001
From: penghui <penghui@apache.org>
Date: Mon, 27 Jun 2022 22:48:00 +0800
Subject: [PATCH] [improve][broker] Avoid go through all the consumers to get
 the message ack owner

### 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.
---
 .../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 cce8fbee88509..bedaabf28d1ee 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
@@ -624,10 +624,12 @@ private void checkCanRemovePendingAcksAndHandle(PositionImpl position, MessageId
     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;
+                    }
                 }
             }
         }