https://issues.apache.org/jira/browse/AMQ-5837

Switching to a List to track dispatched messages in a TopicSubscription
to be consistent with a PrefetchSubscription and to prevent an error
in case acks come back out of order.

activemq-broker/src/main/java/org/apache/activemq/broker/region/TopicSubscription.java

@@ -17,11 +17,9 @@
 package org.apache.activemq.broker.region;
 
 import java.io.IOException;
-import java.util.Comparator;
+import java.util.ArrayList;
-import java.util.Iterator;
 import java.util.LinkedList;
-import java.util.NavigableMap;
+import java.util.List;
-import java.util.concurrent.ConcurrentSkipListMap;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.concurrent.atomic.AtomicLong;
 
@@ -76,19 +74,9 @@ public class TopicSubscription extends AbstractSubscription {
     protected boolean active = false;
     protected boolean discarding = false;
 
-
+    //Used for inflight message size calculations
-    /**
+    protected final Object dispatchLock = new Object();
-     * This Map is used to keep track of messages that have been dispatched in sorted order to
+    protected final List<MessageReference> dispatched = new ArrayList<MessageReference>();
-     * optimize message acknowledgement
-     */
-    private NavigableMap<MessageId, MessageReference> dispatched = new ConcurrentSkipListMap<>(
-            new Comparator<MessageId>() {
-                @Override
-                public int compare(MessageId m1, MessageId m2) {
-                    return m1 == null ? (m2 == null ? 0 : -1) : (m2 == null ? 1
-                            : Long.compare(m1.getBrokerSequenceId(), m2.getBrokerSequenceId()));
-                }
-            });
 
     public TopicSubscription(Broker broker,ConnectionContext context, ConsumerInfo info, SystemUsage usageManager) throws Exception {
         super(broker, context, info);
@@ -267,11 +255,13 @@ public class TopicSubscription extends AbstractSubscription {
                     MessageReference node = matched.next();
                     node.decrementReferenceCount();
                     if (node.getMessageId().equals(mdn.getMessageId())) {
-                        matched.remove();
+                        synchronized(dispatchLock) {
-                        getSubscriptionStatistics().getDispatched().increment();
+                            matched.remove();
-                        dispatched.put(node.getMessageId(), node);
+                            getSubscriptionStatistics().getDispatched().increment();
-                        getSubscriptionStatistics().getInflightMessageSize().addSize(node.getSize());
+                            dispatched.add(node);
-                        node.decrementReferenceCount();
+                            getSubscriptionStatistics().getInflightMessageSize().addSize(node.getSize());
+                            node.decrementReferenceCount();
+                        }
                         break;
                     }
                 }
@@ -403,23 +393,31 @@ public class TopicSubscription extends AbstractSubscription {
     }
 
     /**
-     * Update the inflight statistics on message ack.  Since a message ack could be a range,
+     * Update the inflight statistics on message ack.
-     * we need to grab a subtree of the dispatched map to acknowledge messages.  Finding the
-     * subMap is an O(log n) operation.
      * @param ack
      */
     private void updateInflightMessageSizeOnAck(final MessageAck ack) {
-        if (ack.getFirstMessageId() != null) {
+        synchronized(dispatchLock) {
-            NavigableMap<MessageId, MessageReference> acked = dispatched
+            boolean inAckRange = false;
-                    .subMap(ack.getFirstMessageId(), true, ack.getLastMessageId(), true);
+            List<MessageReference> removeList = new ArrayList<MessageReference>();
-            Iterator<MessageId> i = acked.keySet().iterator();
+            for (final MessageReference node : dispatched) {
-            while (i.hasNext()) {
+                MessageId messageId = node.getMessageId();
-                getSubscriptionStatistics().getInflightMessageSize().addSize(-acked.get(i.next()).getSize());
+                if (ack.getFirstMessageId() == null
-                i.remove();
+                        || ack.getFirstMessageId().equals(messageId)) {
+                    inAckRange = true;
+                }
+                if (inAckRange) {
+                    removeList.add(node);
+                    if (ack.getLastMessageId().equals(messageId)) {
+                        break;
+                    }
+                }
+            }
+
+            for (final MessageReference node : removeList) {
+                dispatched.remove(node);
+                getSubscriptionStatistics().getInflightMessageSize().addSize(-node.getSize());
             }
-        } else {
-            getSubscriptionStatistics().getInflightMessageSize().addSize(-dispatched.get(ack.getLastMessageId()).getSize());
-            dispatched.remove(ack.getLastMessageId());
         }
     }
 
@@ -645,9 +643,12 @@ public class TopicSubscription extends AbstractSubscription {
         md.setConsumerId(info.getConsumerId());
         if (node != null) {
             md.setDestination(((Destination)node.getRegionDestination()).getActiveMQDestination());
-            getSubscriptionStatistics().getDispatched().increment();
+            synchronized(dispatchLock) {
-            dispatched.put(node.getMessageId(), node);
+                getSubscriptionStatistics().getDispatched().increment();
-            getSubscriptionStatistics().getInflightMessageSize().addSize(node.getSize());
+                dispatched.add(node);
+                getSubscriptionStatistics().getInflightMessageSize().addSize(node.getSize());
+            }
+
             // Keep track if this subscription is receiving messages from a single destination.
             if (singleDestination) {
                 if (destination == null) {
@@ -729,7 +730,9 @@ public class TopicSubscription extends AbstractSubscription {
             }
         }
         setSlowConsumer(false);
-        dispatched.clear();
+        synchronized(dispatchLock) {
+            dispatched.clear();
+        }
     }
 
     @Override