From 044386a160aa097a408468d896165ffefd1a9212 Mon Sep 17 00:00:00 2001
From: Nikita Shupletsov <nshuplet@amazon.com>
Date: Mon, 26 Jun 2023 16:13:42 -0700
Subject: [PATCH] Fix slow ack replication. Nikita Shupletsov 13 minutes ago

---
 .../apache/activemq/broker/region/Queue.java  | 23 ++++++++++---------
 1 file changed, 12 insertions(+), 11 deletions(-)

diff --git a/activemq-broker/src/main/java/org/apache/activemq/broker/region/Queue.java b/activemq-broker/src/main/java/org/apache/activemq/broker/region/Queue.java
index f1508f886be..2f0c98cb7e4 100644
--- a/activemq-broker/src/main/java/org/apache/activemq/broker/region/Queue.java
+++ b/activemq-broker/src/main/java/org/apache/activemq/broker/region/Queue.java
@@ -24,6 +24,7 @@
 import java.util.Collection;
 import java.util.Collections;
 import java.util.Comparator;
+import java.util.HashMap;
 import java.util.HashSet;
 import java.util.Iterator;
 import java.util.LinkedHashMap;
@@ -2504,26 +2505,26 @@ private QueueMessageReference getMatchingMessage(MessageDispatchNotification mes
         QueueMessageReference message = null;
         MessageId messageId = messageDispatchNotification.getMessageId();
 
-        Set<MessageReference> set = new LinkedHashSet<MessageReference>();
+        int size = 0;
         do {
             doPageIn(true, false, getMaxPageSize());
             pagedInMessagesLock.readLock().lock();
             try {
-                if (!set.addAll(pagedInMessages.values())) {
+                if (pagedInMessages.size() == size) {
                     // nothing new to check - mem constraint on page in
                     break;
-                };
+                }
+                size = pagedInMessages.size();
+                for (MessageReference ref : pagedInMessages) {
+                    if (ref.getMessageId().equals(messageId)) {
+                        message = (QueueMessageReference) ref;
+                        break;
+                    }
+                }
             } finally {
                 pagedInMessagesLock.readLock().unlock();
             }
-            List<MessageReference> list = new ArrayList<MessageReference>(set);
-            for (MessageReference ref : list) {
-                if (ref.getMessageId().equals(messageId)) {
-                    message = (QueueMessageReference) ref;
-                    break;
-                }
-            }
-        } while (set.size() < this.destinationStatistics.getMessages().getCount());
+        } while (size < this.destinationStatistics.getMessages().getCount());
 
         if (message == null) {
             throw new JMSException("Slave broker out of sync with master - Message: "