You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by pe...@apache.org on 2022/06/28 15:13:30 UTC
[pulsar] 25/29: [improve][broker] Avoid go through all the consumers to get the message ack owner (#16245)
This is an automated email from the ASF dual-hosted git repository.
penghui pushed a commit to branch branch-2.10
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit b75af1772c1c0c38447169f7cff02b089a0ac11d
Author: lipenghui <pe...@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 9b0d678ffc9..627c06b3976 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
@@ -627,10 +627,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;
+ }
}
}
}