Fix slow ack replication. Nikita Shupletsov 13 minutes ago

amazon-mq · Jun 26, 2023 · 044386a · 044386a
1 parent 600f8b6
commit 044386a
Showing 1 changed file with 12 additions and 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
@@ -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: "