activemq-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From gtu...@apache.org
Subject [1/5] rework cursor store sync w.r.t to index order. resolve issues with skipped dispatch and duplicate dispatch. https://issues.apache.org/jira/browse/AMQ-4485 https://issues.apache.org/jira/browse/AMQ-5266
Date Sat, 30 Aug 2014 22:53:41 GMT
Repository: activemq
Updated Branches:
  refs/heads/trunk b2afb8c96 -> e8f815514


http://git-wip-us.apache.org/repos/asf/activemq/blob/54e2e3be/activemq-kahadb-store/src/main/java/org/apache/activemq/store/kahadb/MessageDatabase.java
----------------------------------------------------------------------
diff --git a/activemq-kahadb-store/src/main/java/org/apache/activemq/store/kahadb/MessageDatabase.java b/activemq-kahadb-store/src/main/java/org/apache/activemq/store/kahadb/MessageDatabase.java
index 755f214..776930b 100644
--- a/activemq-kahadb-store/src/main/java/org/apache/activemq/store/kahadb/MessageDatabase.java
+++ b/activemq-kahadb-store/src/main/java/org/apache/activemq/store/kahadb/MessageDatabase.java
@@ -38,6 +38,7 @@ import java.util.HashSet;
 import java.util.Iterator;
 import java.util.LinkedHashMap;
 import java.util.LinkedHashSet;
+import java.util.LinkedList;
 import java.util.List;
 import java.util.Map;
 import java.util.Map.Entry;
@@ -45,6 +46,7 @@ import java.util.Set;
 import java.util.SortedSet;
 import java.util.TreeMap;
 import java.util.TreeSet;
+import java.util.concurrent.Callable;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicLong;
 import java.util.concurrent.locks.ReentrantReadWriteLock;
@@ -254,8 +256,10 @@ public abstract class MessageDatabase extends ServiceSupport implements BrokerSe
     private boolean enableIndexPageCaching = true;
     ReentrantReadWriteLock checkpointLock = new ReentrantReadWriteLock();
 
-    public MessageDatabase() {
+    interface SerialExecution<V> {
+        public V execute(Callable<V> c) throws Exception;
     }
+    SerialExecution<Location> serialExecutor;
 
     @Override
     public void doStart() throws Exception {
@@ -517,12 +521,12 @@ public abstract class MessageDatabase extends ServiceSupport implements BrokerSe
             }
             KahaDestination destination;
             boolean isAdd = false;
-            if (operation instanceof AddOpperation) {
-                AddOpperation add = (AddOpperation) operation;
+            if (operation instanceof AddOperation) {
+                AddOperation add = (AddOperation) operation;
                 destination = add.getCommand().getDestination();
                 isAdd = true;
             } else {
-                RemoveOpperation removeOpperation = (RemoveOpperation) operation;
+                RemoveOperation removeOpperation = (RemoveOperation) operation;
                 destination = removeOpperation.getCommand().getDestination();
             }
             opCount opCount = destinationOpCount.get(destination);
@@ -884,7 +888,7 @@ public abstract class MessageDatabase extends ServiceSupport implements BrokerSe
                 lastRecoveryPosition = nextRecoveryPosition;
                 metadata.lastUpdate = lastRecoveryPosition;
                 JournalCommand<?> message = load(lastRecoveryPosition);
-                process(message, lastRecoveryPosition, (Runnable)null, (Runnable)null);
+                process(message, lastRecoveryPosition, (IndexAware) null);
                 nextRecoveryPosition = journal.getNextLocation(lastRecoveryPosition);
             }
         } finally {
@@ -951,20 +955,34 @@ public abstract class MessageDatabase extends ServiceSupport implements BrokerSe
     }
 
     public Location store(JournalCommand<?> data, Runnable onJournalStoreComplete) throws IOException {
-        return store(data, false, null,null, onJournalStoreComplete);
+        return store(data, false, null, null, onJournalStoreComplete);
     }
 
-    public Location store(JournalCommand<?> data, boolean sync, Runnable before,Runnable after) throws IOException {
+    public Location store(JournalCommand<?> data, boolean sync, IndexAware before,Runnable after) throws IOException {
         return store(data, sync, before, after, null);
     }
 
+    public Location store(final KahaCommitCommand data, final boolean sync, final IndexAware before, final Runnable after) throws IOException {
+        try {
+            return serialExecutor.execute(new Callable<Location>() {
+                @Override
+                public Location call() throws Exception {
+                    return store(data, sync, before, after, null);
+                }
+            });
+        } catch (Exception e) {
+            LOG.error("Failed to execute commit", e);
+            throw new IOException(e);
+        }
+    }
+
     /**
      * All updated are are funneled through this method. The updates are converted
      * to a JournalMessage which is logged to the journal and then the data from
      * the JournalMessage is used to update the index just like it would be done
      * during a recovery process.
      */
-    public Location store(JournalCommand<?> data, boolean sync, Runnable before, Runnable after, Runnable onJournalStoreComplete) throws IOException {
+    public Location store(JournalCommand<?> data, boolean sync, IndexAware before, Runnable after, Runnable onJournalStoreComplete) throws IOException {
         try {
             ByteSequence sequence = toByteSequence(data);
 
@@ -975,7 +993,7 @@ public abstract class MessageDatabase extends ServiceSupport implements BrokerSe
                 long start = System.currentTimeMillis();
                 location = onJournalStoreComplete == null ? journal.write(sequence, sync) :  journal.write(sequence, onJournalStoreComplete) ;
                 long start2 = System.currentTimeMillis();
-                process(data, location, before, after);
+                process(data, location, before);
 
                 long end = System.currentTimeMillis();
                 if( LOG_SLOW_ACCESS_TIME>0 && end-start > LOG_SLOW_ACCESS_TIME) {
@@ -1049,7 +1067,7 @@ public abstract class MessageDatabase extends ServiceSupport implements BrokerSe
                     return;
                 }
             }
-            process(data, location, (Runnable) null, (Runnable) null);
+            process(data, location, (IndexAware) null);
         } else {
             // just recover producer audit
             data.visit(new Visitor() {
@@ -1067,11 +1085,11 @@ public abstract class MessageDatabase extends ServiceSupport implements BrokerSe
     // from the recovery method too so they need to be idempotent
     // /////////////////////////////////////////////////////////////////
 
-    void process(JournalCommand<?> data, final Location location, final Runnable before, final Runnable after) throws IOException {
+    void process(JournalCommand<?> data, final Location location, final IndexAware onSequenceAssignedCallback) throws IOException {
         data.visit(new Visitor() {
             @Override
             public void visit(KahaAddMessageCommand command) throws IOException {
-                process(command, location);
+                process(command, location, onSequenceAssignedCallback);
             }
 
             @Override
@@ -1086,7 +1104,7 @@ public abstract class MessageDatabase extends ServiceSupport implements BrokerSe
 
             @Override
             public void visit(KahaCommitCommand command) throws IOException {
-                process(command, location, before, after);
+                process(command, location, onSequenceAssignedCallback);
             }
 
             @Override
@@ -1127,19 +1145,23 @@ public abstract class MessageDatabase extends ServiceSupport implements BrokerSe
     }
 
     @SuppressWarnings("rawtypes")
-    protected void process(final KahaAddMessageCommand command, final Location location) throws IOException {
+    protected void process(final KahaAddMessageCommand command, final Location location, final IndexAware runWithIndexLock) throws IOException {
         if (command.hasTransactionInfo()) {
-            List<Operation> inflightTx = getInflightTx(command.getTransactionInfo(), location);
-            inflightTx.add(new AddOpperation(command, location));
+            List<Operation> inflightTx = getInflightTx(command.getTransactionInfo());
+            inflightTx.add(new AddOperation(command, location, runWithIndexLock));
         } else {
             this.indexLock.writeLock().lock();
             try {
                 pageFile.tx().execute(new Transaction.Closure<IOException>() {
                     @Override
                     public void execute(Transaction tx) throws IOException {
-                        updateIndex(tx, command, location);
+                        long assignedIndex = updateIndex(tx, command, location);
+                        if (runWithIndexLock != null) {
+                            runWithIndexLock.sequenceAssignedWithIndexLocked(assignedIndex);
+                        }
                     }
                 });
+
             } finally {
                 this.indexLock.writeLock().unlock();
             }
@@ -1164,8 +1186,8 @@ public abstract class MessageDatabase extends ServiceSupport implements BrokerSe
     @SuppressWarnings("rawtypes")
     protected void process(final KahaRemoveMessageCommand command, final Location location) throws IOException {
         if (command.hasTransactionInfo()) {
-           List<Operation> inflightTx = getInflightTx(command.getTransactionInfo(), location);
-           inflightTx.add(new RemoveOpperation(command, location));
+           List<Operation> inflightTx = getInflightTx(command.getTransactionInfo());
+           inflightTx.add(new RemoveOperation(command, location));
         } else {
             this.indexLock.writeLock().lock();
             try {
@@ -1219,7 +1241,7 @@ public abstract class MessageDatabase extends ServiceSupport implements BrokerSe
     }
 
     @SuppressWarnings("rawtypes")
-    protected void process(KahaCommitCommand command, Location location, final Runnable before, final Runnable after) throws IOException {
+    protected void process(KahaCommitCommand command, final Location location, final IndexAware before) throws IOException {
         TransactionId key = TransactionIdConversion.convert(command.getTransactionInfo());
         List<Operation> inflightTx;
         synchronized (inflightTransactions) {
@@ -1231,29 +1253,30 @@ public abstract class MessageDatabase extends ServiceSupport implements BrokerSe
         if (inflightTx == null) {
             // only non persistent messages in this tx
             if (before != null) {
-                before.run();
+                before.sequenceAssignedWithIndexLocked(-1);
             }
             return;
         }
 
         final List<Operation> messagingTx = inflightTx;
-        this.indexLock.writeLock().lock();
         try {
-            // run before with the index lock so that queue can order cursor updates with index updates
-            if (before != null) {
-                before.run();
-            }
-            pageFile.tx().execute(new Transaction.Closure<IOException>() {
-                @Override
-                public void execute(Transaction tx) throws IOException {
-                    for (Operation op : messagingTx) {
-                        op.execute(tx);
+            indexLock.writeLock().lock();
+            try {
+                pageFile.tx().execute(new Transaction.Closure<IOException>() {
+                    @Override
+                    public void execute(Transaction tx) throws IOException {
+                        for (Operation op : messagingTx) {
+                            op.execute(tx);
+                        }
                     }
-                }
-            });
-            metadata.lastUpdate = location;
-        } finally {
-            this.indexLock.writeLock().unlock();
+                });
+                metadata.lastUpdate = location;
+            } finally {
+                indexLock.writeLock().unlock();
+            }
+        } catch (Exception e) {
+            LOG.error("serial execution of commit failed", e);
+            throw new IOException(e);
         }
     }
 
@@ -1287,13 +1310,13 @@ public abstract class MessageDatabase extends ServiceSupport implements BrokerSe
     protected final ReentrantReadWriteLock indexLock = new ReentrantReadWriteLock();
     private final HashSet<Integer> journalFilesBeingReplicated = new HashSet<Integer>();
 
-    void updateIndex(Transaction tx, KahaAddMessageCommand command, Location location) throws IOException {
+    long updateIndex(Transaction tx, KahaAddMessageCommand command, Location location) throws IOException {
         StoredDestination sd = getStoredDestination(command.getDestination(), tx);
 
         // Skip adding the message to the index if this is a topic and there are
         // no subscriptions.
         if (sd.subscriptions != null && sd.subscriptions.isEmpty(tx)) {
-            return;
+            return -1;
         }
 
         // Add the message.
@@ -1308,12 +1331,11 @@ public abstract class MessageDatabase extends ServiceSupport implements BrokerSe
                     addAckLocationForNewMessage(tx, sd, id);
                 }
             } else {
-                // If the message ID as indexed, then the broker asked us to
-                // store a DUP message. Bad BOY! Don't do it, and log a warning.
+                // If the message ID is indexed, then the broker asked us to store a duplicate before the message was dispatched and acked, we ignore this add attempt
                 LOG.warn("Duplicate message add attempt rejected. Destination: {}://{}, Message id: {}", command.getDestination().getType(), command.getDestination().getName(), command.getMessageId());
                 sd.messageIdIndex.put(tx, command.getMessageId(), previous);
                 sd.locationIndex.remove(tx, location);
-                rollbackStatsOnDuplicate(command.getDestination());
+                id = -1;
             }
         } else {
             // restore the previous value.. Looks like this was a redo of a previously
@@ -1324,6 +1346,21 @@ public abstract class MessageDatabase extends ServiceSupport implements BrokerSe
         // record this id in any event, initial send or recovery
         metadata.producerSequenceIdTracker.isDuplicate(command.getMessageId());
         metadata.lastUpdate = location;
+        return id;
+    }
+
+    void trackPendingAdd(KahaDestination destination, Long seq) {
+        StoredDestination sd = storedDestinations.get(key(destination));
+        if (sd != null) {
+            sd.trackPendingAdd(seq);
+        }
+    }
+
+    void trackPendingAddComplete(KahaDestination destination, Long seq) {
+        StoredDestination sd = storedDestinations.get(key(destination));
+        if (sd != null) {
+            sd.trackPendingAddComplete(seq);
+        }
     }
 
     void updateIndex(Transaction tx, KahaUpdateMessageCommand updateMessageCommand, Location location) throws IOException {
@@ -1345,8 +1382,6 @@ public abstract class MessageDatabase extends ServiceSupport implements BrokerSe
         metadata.lastUpdate = location;
     }
 
-    abstract void rollbackStatsOnDuplicate(KahaDestination commandDestination);
-
     void updateIndex(Transaction tx, KahaRemoveMessageCommand command, Location ackLocation) throws IOException {
         StoredDestination sd = getStoredDestination(command.getDestination(), tx);
         if (!command.hasSubscriptionKey()) {
@@ -1864,6 +1899,14 @@ public abstract class MessageDatabase extends ServiceSupport implements BrokerSe
         // Transient data used to track which Messages are no longer needed.
         final TreeMap<Long, Long> messageReferences = new TreeMap<Long, Long>();
         final HashSet<String> subscriptionCache = new LinkedHashSet<String>();
+
+        public void trackPendingAdd(Long seq) {
+            orderIndex.trackPendingAdd(seq);
+        }
+
+        public void trackPendingAddComplete(Long seq) {
+            orderIndex.trackPendingAddComplete(seq);
+        }
     }
 
     protected class StoredDestinationMarshaller extends VariableMarshaller<StoredDestination> {
@@ -2361,7 +2404,7 @@ public abstract class MessageDatabase extends ServiceSupport implements BrokerSe
     }
 
     @SuppressWarnings("rawtypes")
-    private List<Operation> getInflightTx(KahaTransactionInfo info, Location location) {
+    private List<Operation> getInflightTx(KahaTransactionInfo info) {
         TransactionId key = TransactionIdConversion.convert(info);
         List<Operation> tx;
         synchronized (inflightTransactions) {
@@ -2399,22 +2442,26 @@ public abstract class MessageDatabase extends ServiceSupport implements BrokerSe
         abstract public void execute(Transaction tx) throws IOException;
     }
 
-    class AddOpperation extends Operation<KahaAddMessageCommand> {
-
-        public AddOpperation(KahaAddMessageCommand command, Location location) {
+    class AddOperation extends Operation<KahaAddMessageCommand> {
+        final IndexAware runWithIndexLock;
+        public AddOperation(KahaAddMessageCommand command, Location location, IndexAware runWithIndexLock) {
             super(command, location);
+            this.runWithIndexLock = runWithIndexLock;
         }
 
         @Override
         public void execute(Transaction tx) throws IOException {
-            updateIndex(tx, command, location);
+            long seq = updateIndex(tx, command, location);
+            if (runWithIndexLock != null) {
+                runWithIndexLock.sequenceAssignedWithIndexLocked(seq);
+            }
         }
 
     }
 
-    class RemoveOpperation extends Operation<KahaRemoveMessageCommand> {
+    class RemoveOperation extends Operation<KahaRemoveMessageCommand> {
 
-        public RemoveOpperation(KahaRemoveMessageCommand command, Location location) {
+        public RemoveOperation(KahaRemoveMessageCommand command, Location location) {
             super(command, location);
         }
 
@@ -2765,6 +2812,7 @@ public abstract class MessageDatabase extends ServiceSupport implements BrokerSe
         Long lastHighKey;
         Long lastLowKey;
         byte lastGetPriority;
+        final List<Long> pendingAdditions = new LinkedList<Long>();
 
         MessageKeys remove(Transaction tx, Long key) throws IOException {
             MessageKeys result = defaultPriorityIndex.remove(tx, key);
@@ -2928,7 +2976,7 @@ public abstract class MessageDatabase extends ServiceSupport implements BrokerSe
         void getDeleteList(Transaction tx, ArrayList<Entry<Long, MessageKeys>> deletes,
                 BTreeIndex<Long, MessageKeys> index, Long sequenceId) throws IOException {
 
-            Iterator<Entry<Long, MessageKeys>> iterator = index.iterator(tx, sequenceId);
+            Iterator<Entry<Long, MessageKeys>> iterator = index.iterator(tx, sequenceId, null);
             deletes.add(iterator.next());
         }
 
@@ -2963,11 +3011,11 @@ public abstract class MessageDatabase extends ServiceSupport implements BrokerSe
         }
 
         Iterator<Entry<Long, MessageKeys>> iterator(Transaction tx) throws IOException{
-            return new MessageOrderIterator(tx,cursor);
+            return new MessageOrderIterator(tx,cursor,this);
         }
 
         Iterator<Entry<Long, MessageKeys>> iterator(Transaction tx, MessageOrderCursor m) throws IOException{
-            return new MessageOrderIterator(tx,m);
+            return new MessageOrderIterator(tx,m,this);
         }
 
         public byte lastGetPriority() {
@@ -2980,21 +3028,45 @@ public abstract class MessageDatabase extends ServiceSupport implements BrokerSe
                     (cursor.lowPriorityCursorPosition > 0 && cursor.lowPriorityCursorPosition >= sequence);
         }
 
+        public void trackPendingAdd(Long seq) {
+            synchronized (pendingAdditions) {
+                pendingAdditions.add(seq);
+            }
+        }
+
+        public void trackPendingAddComplete(Long seq) {
+            synchronized (pendingAdditions) {
+                pendingAdditions.remove(seq);
+            }
+        }
+
+        public Long minPendingAdd() {
+            synchronized (pendingAdditions) {
+                if (!pendingAdditions.isEmpty()) {
+                    return pendingAdditions.get(0);
+                } else {
+                    return null;
+                }
+            }
+        }
+
+
         class MessageOrderIterator implements Iterator<Entry<Long, MessageKeys>>{
             Iterator<Entry<Long, MessageKeys>>currentIterator;
             final Iterator<Entry<Long, MessageKeys>>highIterator;
             final Iterator<Entry<Long, MessageKeys>>defaultIterator;
             final Iterator<Entry<Long, MessageKeys>>lowIterator;
 
-            MessageOrderIterator(Transaction tx, MessageOrderCursor m) throws IOException {
-                this.defaultIterator = defaultPriorityIndex.iterator(tx, m.defaultCursorPosition);
+            MessageOrderIterator(Transaction tx, MessageOrderCursor m, MessageOrderIndex messageOrderIndex) throws IOException {
+                Long pendingAddLimiter = messageOrderIndex.minPendingAdd();
+                this.defaultIterator = defaultPriorityIndex.iterator(tx, m.defaultCursorPosition, pendingAddLimiter);
                 if (highPriorityIndex != null) {
-                    this.highIterator = highPriorityIndex.iterator(tx, m.highPriorityCursorPosition);
+                    this.highIterator = highPriorityIndex.iterator(tx, m.highPriorityCursorPosition, pendingAddLimiter);
                 } else {
                     this.highIterator = null;
                 }
                 if (lowPriorityIndex != null) {
-                    this.lowIterator = lowPriorityIndex.iterator(tx, m.lowPriorityCursorPosition);
+                    this.lowIterator = lowPriorityIndex.iterator(tx, m.lowPriorityCursorPosition, pendingAddLimiter);
                 } else {
                     this.lowIterator = null;
                 }
@@ -3117,4 +3189,8 @@ public abstract class MessageDatabase extends ServiceSupport implements BrokerSe
     public void setIndexDirectory(File indexDirectory) {
         this.indexDirectory = indexDirectory;
     }
+
+    interface IndexAware {
+        public void sequenceAssignedWithIndexLocked(long index);
+    }
 }

http://git-wip-us.apache.org/repos/asf/activemq/blob/54e2e3be/activemq-kahadb-store/src/main/java/org/apache/activemq/store/kahadb/disk/index/BTreeIndex.java
----------------------------------------------------------------------
diff --git a/activemq-kahadb-store/src/main/java/org/apache/activemq/store/kahadb/disk/index/BTreeIndex.java b/activemq-kahadb-store/src/main/java/org/apache/activemq/store/kahadb/disk/index/BTreeIndex.java
index 91b46ba..fca1ea0 100644
--- a/activemq-kahadb-store/src/main/java/org/apache/activemq/store/kahadb/disk/index/BTreeIndex.java
+++ b/activemq-kahadb-store/src/main/java/org/apache/activemq/store/kahadb/disk/index/BTreeIndex.java
@@ -229,7 +229,11 @@ public class BTreeIndex<Key,Value> implements Index<Key,Value> {
     }
     
     synchronized public Iterator<Map.Entry<Key,Value>> iterator(final Transaction tx, Key initialKey) throws IOException {
-        return getRoot(tx).iterator(tx, initialKey);
+        return getRoot(tx).iterator(tx, initialKey, null);
+    }
+
+    synchronized public Iterator<Map.Entry<Key,Value>> iterator(final Transaction tx, Key initialKey, Key maxKey) throws IOException {
+        return getRoot(tx).iterator(tx, initialKey, maxKey);
     }
     
     synchronized public void visit(Transaction tx, BTreeVisitor<Key, Value> visitor) throws IOException {

http://git-wip-us.apache.org/repos/asf/activemq/blob/54e2e3be/activemq-kahadb-store/src/main/java/org/apache/activemq/store/kahadb/disk/index/BTreeNode.java
----------------------------------------------------------------------
diff --git a/activemq-kahadb-store/src/main/java/org/apache/activemq/store/kahadb/disk/index/BTreeNode.java b/activemq-kahadb-store/src/main/java/org/apache/activemq/store/kahadb/disk/index/BTreeNode.java
index 7f5404b..a53d5bf 100644
--- a/activemq-kahadb-store/src/main/java/org/apache/activemq/store/kahadb/disk/index/BTreeNode.java
+++ b/activemq-kahadb-store/src/main/java/org/apache/activemq/store/kahadb/disk/index/BTreeNode.java
@@ -80,14 +80,19 @@ public final class BTreeNode<Key,Value> {
     private final class BTreeIterator implements Iterator<Map.Entry<Key, Value>> {
         
         private final Transaction tx;
+        private final Key endKey;
         BTreeNode<Key,Value> current;
         int nextIndex;
         Map.Entry<Key,Value> nextEntry;
 
-        private BTreeIterator(Transaction tx, BTreeNode<Key,Value> current, int nextIndex) {
+        private BTreeIterator(Transaction tx, BTreeNode<Key, Value> current, int nextIndex, Key endKey) {
             this.tx = tx;
             this.current = current;
             this.nextIndex=nextIndex;
+            this.endKey = endKey;
+            if (endKey != null && endKey.equals(0l)) {
+                Thread.dumpStack();
+            }
         }
 
         synchronized private void findNextPage() {
@@ -107,6 +112,10 @@ public final class BTreeNode<Key,Value> {
                             break;
                         }
                     }  else {
+                        if (endKey != null && current.keys[nextIndex].equals(endKey)) {
+                            System.err.println("Stopping iterator on reaching: " + endKey);
+                            break;
+                        }
                         nextEntry = new KeyValueEntry(current.keys[nextIndex], current.values[nextIndex]);
                         nextIndex++;
                         break;
@@ -631,23 +640,23 @@ public final class BTreeNode<Key,Value> {
         return node;
     }
     
-    public Iterator<Map.Entry<Key,Value>> iterator(final Transaction tx, Key startKey) throws IOException {
+    public Iterator<Map.Entry<Key,Value>> iterator(final Transaction tx, Key startKey, Key endKey) throws IOException {
         if (startKey == null) {
             return iterator(tx);
         }
         if( isBranch() ) {
-            return getLeafNode(tx, this, startKey).iterator(tx, startKey);
+            return getLeafNode(tx, this, startKey).iterator(tx, startKey, endKey);
         } else {
             int idx = Arrays.binarySearch(keys, startKey);
             if (idx < 0) {
                 idx = -(idx + 1);
             }
-            return new BTreeIterator(tx, this, idx);
+            return new BTreeIterator(tx, this, idx, endKey);
         }
     }
 
     public Iterator<Map.Entry<Key,Value>> iterator(final Transaction tx) throws IOException {
-        return new BTreeIterator(tx, getFirstLeafNode(tx), 0);
+        return new BTreeIterator(tx, getFirstLeafNode(tx), 0, null);
     }
     
     public void clear(Transaction tx) throws IOException {

http://git-wip-us.apache.org/repos/asf/activemq/blob/54e2e3be/activemq-kahadb-store/src/test/java/org/apache/activemq/store/kahadb/disk/index/BTreeIndexTest.java
----------------------------------------------------------------------
diff --git a/activemq-kahadb-store/src/test/java/org/apache/activemq/store/kahadb/disk/index/BTreeIndexTest.java b/activemq-kahadb-store/src/test/java/org/apache/activemq/store/kahadb/disk/index/BTreeIndexTest.java
index 188de40..ed0d12b 100644
--- a/activemq-kahadb-store/src/test/java/org/apache/activemq/store/kahadb/disk/index/BTreeIndexTest.java
+++ b/activemq-kahadb-store/src/test/java/org/apache/activemq/store/kahadb/disk/index/BTreeIndexTest.java
@@ -163,6 +163,35 @@ public class BTreeIndexTest extends IndexTestSupport {
     }
 
     @Test(timeout=60000)
+    public void testLimitedIteration() throws Exception {
+        createPageFileAndIndex(500);
+        BTreeIndex<String,Long> index = ((BTreeIndex<String,Long>)this.index);
+        this.index.load(tx);
+        tx.commit();
+
+        // Insert in reverse order..
+        doInsertReverse(1000);
+
+        this.index.unload(tx);
+        tx.commit();
+        this.index.load(tx);
+        tx.commit();
+
+        // BTree should iterate it in sorted order up to limit
+        int counter=0;
+        for (Iterator<Map.Entry<String,Long>> i = index.iterator(tx, key(0), key(500)); i.hasNext();) {
+            Map.Entry<String,Long> entry = i.next();
+            assertEquals(key(counter),entry.getKey());
+            assertEquals(counter,(long)entry.getValue());
+            counter++;
+        }
+
+        assertEquals("got to 500", 500, counter);
+        this.index.unload(tx);
+        tx.commit();
+    }
+
+    @Test(timeout=60000)
     public void testVisitor() throws Exception {
         createPageFileAndIndex(100);
         BTreeIndex<String,Long> index = ((BTreeIndex<String,Long>)this.index);

http://git-wip-us.apache.org/repos/asf/activemq/blob/54e2e3be/activemq-leveldb-store/src/main/scala/org/apache/activemq/leveldb/DBManager.scala
----------------------------------------------------------------------
diff --git a/activemq-leveldb-store/src/main/scala/org/apache/activemq/leveldb/DBManager.scala b/activemq-leveldb-store/src/main/scala/org/apache/activemq/leveldb/DBManager.scala
index 775b99a..34bcc6a 100644
--- a/activemq-leveldb-store/src/main/scala/org/apache/activemq/leveldb/DBManager.scala
+++ b/activemq-leveldb-store/src/main/scala/org/apache/activemq/leveldb/DBManager.scala
@@ -104,6 +104,7 @@ class CountDownFuture[T <: AnyRef]() extends ListenableFuture[T] {
   var value:T = _
   var error:Throwable = _
   var listener:Runnable = _
+  var id:MessageId = _
 
   def cancel(mayInterruptIfRunning: Boolean) = false
   def isCancelled = false
@@ -115,6 +116,9 @@ class CountDownFuture[T <: AnyRef]() extends ListenableFuture[T] {
 
   def set(v:T) = {
     value = v
+    if (id != null) {
+      id.setFutureOrSequenceLong(id.getEntryLocator.asInstanceOf[EntryLocator].seq)
+    }
     latch.countDown()
     fireListener
   }
@@ -326,6 +330,8 @@ class DelayableUOW(val manager:DBManager) extends BaseRetained {
     val entry = QueueEntryRecord(id, queueKey, queueSeq)
     assert(id.getEntryLocator == null)
     id.setEntryLocator(EntryLocator(queueKey, queueSeq))
+    id.setFutureOrSequenceLong(countDownFuture)
+    countDownFuture.id = id
 
     val a = this.synchronized {
       if( !delay )

http://git-wip-us.apache.org/repos/asf/activemq/blob/54e2e3be/activemq-leveldb-store/src/main/scala/org/apache/activemq/leveldb/LevelDBStore.scala
----------------------------------------------------------------------
diff --git a/activemq-leveldb-store/src/main/scala/org/apache/activemq/leveldb/LevelDBStore.scala b/activemq-leveldb-store/src/main/scala/org/apache/activemq/leveldb/LevelDBStore.scala
index 04eaf16..6e3faff 100644
--- a/activemq-leveldb-store/src/main/scala/org/apache/activemq/leveldb/LevelDBStore.scala
+++ b/activemq-leveldb-store/src/main/scala/org/apache/activemq/leveldb/LevelDBStore.scala
@@ -245,7 +245,7 @@ class LevelDBStore extends LockableServiceSupport with BrokerServiceAware with P
       val (msgs, acks) = db.getXAActions(transaction.xacontainer_id)
       transaction.xarecovery = (msgs, acks.map(_.ack))
       for ( msg <- msgs ) {
-        transaction.add(createMessageStore(msg.getDestination), msg, false);
+        transaction.add(createMessageStore(msg.getDestination), new IndexListener.MessageContext(null, msg, null), false);
       }
       for ( record <- acks ) {
         var ack = record.ack
@@ -348,27 +348,27 @@ class LevelDBStore extends LockableServiceSupport with BrokerServiceAware with P
       }
     }
 
-    def add(store:LevelDBStore#LevelDBMessageStore, message: Message, delay:Boolean) = {
+  def add(store:LevelDBStore#LevelDBMessageStore, messageContext:IndexListener.MessageContext, delay:Boolean) = {
       commitActions += new TransactionAction() {
         def commit(uow:DelayableUOW) = {
           if( prepared ) {
-            uow.dequeue(xacontainer_id, message.getMessageId)
+            uow.dequeue(xacontainer_id, messageContext.message.getMessageId)
           }
-          var copy = message.getMessageId.copy()
+          var copy = messageContext.message.getMessageId.copy()
           copy.setEntryLocator(null)
-          message.setMessageId(copy)
-          store.doAdd(uow, message, delay)
+          messageContext.message.setMessageId(copy)
+          store.doAdd(uow, messageContext, delay)
         }
 
         def prepare(uow:DelayableUOW) = {
           // add it to the xa container instead of the actual store container.
-          uow.enqueue(xacontainer_id, xaseqcounter.incrementAndGet, message, delay)
-          xarecovery._1 += message
+          uow.enqueue(xacontainer_id, xaseqcounter.incrementAndGet, messageContext.message, delay)
+          xarecovery._1 += messageContext.message
         }
 
         def rollback(uow:DelayableUOW) = {
           if( prepared ) {
-            uow.dequeue(xacontainer_id, message.getMessageId)
+            uow.dequeue(xacontainer_id, messageContext.message.getMessageId)
           }
         }
 
@@ -676,14 +676,19 @@ class LevelDBStore extends LockableServiceSupport with BrokerServiceAware with P
 
     def cursorResetPosition = 0L
 
-    def doAdd(uow: DelayableUOW, message: Message, delay:Boolean): CountDownFuture[AnyRef] = {
+    def doAdd(uow: DelayableUOW, messageContext:IndexListener.MessageContext, delay:Boolean): CountDownFuture[AnyRef] = {
       check_running
       val seq = lastSeq.incrementAndGet()
-      message.incrementReferenceCount()
+      messageContext.message.incrementReferenceCount()
       uow.addCompleteListener({
-        message.decrementReferenceCount()
+        messageContext.message.decrementReferenceCount()
       })
-      uow.enqueue(key, seq, message, delay)
+      val future = uow.enqueue(key, seq, messageContext.message, delay)
+      messageContext.message.getMessageId.setFutureOrSequenceLong(future)
+      if (indexListener != null) {
+        indexListener.onAdd(messageContext)
+      }
+      future
     }
 
     override def asyncAddQueueMessage(context: ConnectionContext, message: Message) = asyncAddQueueMessage(context, message, false)
@@ -691,11 +696,11 @@ class LevelDBStore extends LockableServiceSupport with BrokerServiceAware with P
       check_running
       message.getMessageId.setEntryLocator(null)
       if(  message.getTransactionId!=null ) {
-        transaction(message.getTransactionId).add(this, message, delay)
+        transaction(message.getTransactionId).add(this, new IndexListener.MessageContext(context, message, null), delay)
         DONE
       } else {
         withUow { uow=>
-          doAdd(uow, message, delay)
+          doAdd(uow, new IndexListener.MessageContext(context, message, null), delay)
         }
       }
     }

http://git-wip-us.apache.org/repos/asf/activemq/blob/54e2e3be/activemq-unit-tests/src/test/java/org/apache/activemq/broker/region/cursors/StoreQueueCursorNoDuplicateTest.java
----------------------------------------------------------------------
diff --git a/activemq-unit-tests/src/test/java/org/apache/activemq/broker/region/cursors/StoreQueueCursorNoDuplicateTest.java b/activemq-unit-tests/src/test/java/org/apache/activemq/broker/region/cursors/StoreQueueCursorNoDuplicateTest.java
index 9d9fa64..d517bcd 100644
--- a/activemq-unit-tests/src/test/java/org/apache/activemq/broker/region/cursors/StoreQueueCursorNoDuplicateTest.java
+++ b/activemq-unit-tests/src/test/java/org/apache/activemq/broker/region/cursors/StoreQueueCursorNoDuplicateTest.java
@@ -78,6 +78,7 @@ public class StoreQueueCursorNoDuplicateTest extends TestCase {
                 queueMessageStore, destinationStatistics, null);
 
         queueMessageStore.start();
+        queueMessageStore.registerIndexListener(null);
 
         QueueStorePrefetch underTest = new QueueStorePrefetch(queue, brokerService.getBroker());
         SystemUsage systemUsage = new SystemUsage();

http://git-wip-us.apache.org/repos/asf/activemq/blob/54e2e3be/activemq-unit-tests/src/test/java/org/apache/activemq/bugs/AMQ4485LowLimitTest.java
----------------------------------------------------------------------
diff --git a/activemq-unit-tests/src/test/java/org/apache/activemq/bugs/AMQ4485LowLimitTest.java b/activemq-unit-tests/src/test/java/org/apache/activemq/bugs/AMQ4485LowLimitTest.java
new file mode 100644
index 0000000..5af1124
--- /dev/null
+++ b/activemq-unit-tests/src/test/java/org/apache/activemq/bugs/AMQ4485LowLimitTest.java
@@ -0,0 +1,462 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.activemq.bugs;
+
+import java.io.IOException;
+import java.net.URI;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.Iterator;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.concurrent.ConcurrentLinkedQueue;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+import javax.jms.JMSException;
+import javax.jms.Message;
+import javax.jms.MessageListener;
+import javax.jms.MessageProducer;
+import javax.jms.QueueConnection;
+import javax.jms.QueueReceiver;
+import javax.jms.QueueSession;
+import javax.jms.Session;
+import javax.jms.TextMessage;
+import javax.management.ObjectName;
+import org.apache.activemq.ActiveMQConnection;
+import org.apache.activemq.ActiveMQConnectionFactory;
+import org.apache.activemq.JmsMultipleBrokersTestSupport;
+import org.apache.activemq.broker.BrokerService;
+import org.apache.activemq.broker.jmx.QueueViewMBean;
+import org.apache.activemq.broker.region.RegionBroker;
+import org.apache.activemq.broker.region.policy.PolicyEntry;
+import org.apache.activemq.broker.region.policy.PolicyMap;
+import org.apache.activemq.command.ActiveMQDestination;
+import org.apache.activemq.command.ActiveMQMessage;
+import org.apache.activemq.command.ActiveMQQueue;
+import org.apache.activemq.command.BrokerInfo;
+import org.apache.activemq.network.DiscoveryNetworkConnector;
+import org.apache.activemq.network.NetworkConnector;
+import org.apache.activemq.store.kahadb.KahaDBPersistenceAdapter;
+import org.apache.activemq.util.TimeUtils;
+import org.apache.activemq.util.Wait;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class AMQ4485LowLimitTest extends JmsMultipleBrokersTestSupport {
+    static final String payload = new String(new byte[10 * 1024]);
+    private static final Logger LOG = LoggerFactory.getLogger(AMQ4485LowLimitTest.class);
+    final int portBase = 61600;
+    final int numBrokers = 8;
+    final int numProducers = 30;
+    final int numMessages = 1000;
+    final int consumerSleepTime = 40;
+    StringBuilder brokersUrl = new StringBuilder();
+    HashMap<ActiveMQQueue, AtomicInteger> accumulators = new HashMap<ActiveMQQueue, AtomicInteger>();
+    private ArrayList<Throwable> exceptions = new ArrayList<Throwable>();
+
+    protected void buildUrlList() throws Exception {
+        for (int i = 0; i < numBrokers; i++) {
+            brokersUrl.append("tcp://localhost:" + (portBase + i));
+            if (i != numBrokers - 1) {
+                brokersUrl.append(',');
+            }
+        }
+    }
+
+    protected BrokerService createBroker(int brokerid) throws Exception {
+        return createBroker(brokerid, true);
+    }
+
+    protected BrokerService createBroker(int brokerid, boolean addToNetwork) throws Exception {
+
+        BrokerService broker = new BrokerService();
+        broker.setPersistent(true);
+        broker.setDeleteAllMessagesOnStartup(true);
+        broker.getManagementContext().setCreateConnector(false);
+
+
+        broker.setUseJmx(true);
+        broker.setBrokerName("B" + brokerid);
+        broker.addConnector(new URI("tcp://localhost:" + (portBase + brokerid)));
+
+        if (addToNetwork) {
+            addNetworkConnector(broker);
+        }
+        broker.setSchedulePeriodForDestinationPurge(0);
+        //broker.getSystemUsage().setSendFailIfNoSpace(true);
+        broker.getSystemUsage().getMemoryUsage().setLimit(256 * 1024 * 1024l);
+
+
+        PolicyMap policyMap = new PolicyMap();
+        PolicyEntry policyEntry = new PolicyEntry();
+        policyEntry.setExpireMessagesPeriod(0);
+        policyEntry.setQueuePrefetch(1000);
+        policyEntry.setMemoryLimit(2 * 1024 * 1024l);
+        policyEntry.setProducerFlowControl(false);
+        policyEntry.setEnableAudit(true);
+        policyEntry.setUseCache(true);
+        policyMap.put(new ActiveMQQueue("GW.>"), policyEntry);
+
+        PolicyEntry inPolicyEntry = new PolicyEntry();
+        inPolicyEntry.setExpireMessagesPeriod(0);
+        inPolicyEntry.setQueuePrefetch(1000);
+        inPolicyEntry.setMemoryLimit(5 * 1024 * 1024l);
+        inPolicyEntry.setProducerFlowControl(true);
+        inPolicyEntry.setEnableAudit(true);
+        inPolicyEntry.setUseCache(true);
+        policyMap.put(new ActiveMQQueue("IN"), inPolicyEntry);
+
+        broker.setDestinationPolicy(policyMap);
+
+        KahaDBPersistenceAdapter kahaDBPersistenceAdapter = (KahaDBPersistenceAdapter) broker.getPersistenceAdapter();
+        kahaDBPersistenceAdapter.setConcurrentStoreAndDispatchQueues(true);
+
+        brokers.put(broker.getBrokerName(), new BrokerItem(broker));
+        return broker;
+    }
+
+    private void addNetworkConnector(BrokerService broker) throws Exception {
+        StringBuilder networkConnectorUrl = new StringBuilder("static:(").append(brokersUrl.toString());
+        networkConnectorUrl.append(')');
+
+        for (int i = 0; i < 2; i++) {
+            NetworkConnector nc = new DiscoveryNetworkConnector(new URI(networkConnectorUrl.toString()));
+            nc.setName("Bridge-" + i);
+            nc.setNetworkTTL(1);
+            nc.setDecreaseNetworkConsumerPriority(true);
+            nc.setDynamicOnly(true);
+            nc.setPrefetchSize(100);
+            nc.setDynamicallyIncludedDestinations(
+                    Arrays.asList(new ActiveMQDestination[]{new ActiveMQQueue("GW.*")}));
+            broker.addNetworkConnector(nc);
+        }
+    }
+
+    // used to explore contention with concurrentStoreandDispatch - sync commit and task queue reversing
+    // order of cursor add and sequence assignment
+    public void x_testInterleavedSend() throws Exception {
+
+        BrokerService b = createBroker(0, false);
+        b.start();
+
+        ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory("tcp://localhost:" + (portBase + 0));
+        connectionFactory.setWatchTopicAdvisories(false);
+
+        QueueConnection c1 = connectionFactory.createQueueConnection();
+        QueueConnection c2 = connectionFactory.createQueueConnection();
+        QueueConnection c3 = connectionFactory.createQueueConnection();
+
+        c1.start();
+        c2.start();
+        c3.start();
+
+        ActiveMQQueue dest = new ActiveMQQueue("IN");
+        final Session s1 = c1.createQueueSession(true, Session.SESSION_TRANSACTED);
+        final TextMessage txMessage = s1.createTextMessage("TX");
+        final TextMessage noTxMessage = s1.createTextMessage("NO_TX");
+
+        final MessageProducer txProducer = s1.createProducer(dest);
+        final MessageProducer nonTxProducer = c2.createQueueSession(false, Session.AUTO_ACKNOWLEDGE).createProducer(dest);
+
+        txProducer.send(txMessage);
+
+        ExecutorService executorService = Executors.newFixedThreadPool(2);
+        executorService.execute(new Runnable() {
+            @Override
+            public void run() {
+                try {
+                    s1.commit();
+                } catch (JMSException e) {
+                    e.printStackTrace();
+                }
+            }
+        });
+
+        executorService.execute(new Runnable() {
+            @Override
+            public void run() {
+                try {
+                    nonTxProducer.send(noTxMessage);
+                } catch (JMSException e) {
+                    e.printStackTrace();
+                }
+            }
+        });
+
+        executorService.shutdown();
+        executorService.awaitTermination(10, TimeUnit.MINUTES);
+
+    }
+
+    public void testBrokers() throws Exception {
+
+        buildUrlList();
+
+        for (int i = 0; i < numBrokers; i++) {
+            createBroker(i);
+        }
+
+        startAllBrokers();
+        waitForBridgeFormation(numBrokers - 1);
+
+        verifyPeerBrokerInfos(numBrokers - 1);
+
+
+        final List<ConsumerState> consumerStates = startAllGWConsumers(numBrokers);
+
+        startAllGWFanoutConsumers(numBrokers);
+
+        LOG.info("Waiting for percolation of consumers..");
+        TimeUnit.SECONDS.sleep(5);
+
+        LOG.info("Produce mesages..");
+        long startTime = System.currentTimeMillis();
+
+        // produce
+        produce(numMessages);
+
+        assertTrue("Got all sent", Wait.waitFor(new Wait.Condition() {
+            @Override
+            public boolean isSatisified() throws Exception {
+                for (ConsumerState tally : consumerStates) {
+                    final int expected = numMessages * (tally.destination.isComposite() ? tally.destination.getCompositeDestinations().length : 1);
+                    LOG.info("Tally for: " + tally.brokerName + ", dest: " + tally.destination + " - " + tally.accumulator.get());
+                    if (tally.accumulator.get() != expected) {
+                        LOG.info("Tally for: " + tally.brokerName + ", dest: " + tally.destination + " - " + tally.accumulator.get() + " != " + expected + ", " + tally.expected);
+                        if (tally.accumulator.get() > expected - 50) {
+                            dumpQueueStat(tally.destination);
+                        }
+                        if (tally.expected.size() == 1) {
+                            startConsumer(tally.brokerName, tally.destination);
+                        };
+                        return false;
+                    }
+                    LOG.info("got tally on " + tally.brokerName);
+                }
+                return true;
+            }
+        }, 1000 * 60 * 1000l, 20*1000));
+
+        assertTrue("No exceptions:" + exceptions, exceptions.isEmpty());
+
+        LOG.info("done");
+        long duration = System.currentTimeMillis() - startTime;
+        LOG.info("Duration:" + TimeUtils.printDuration(duration));
+    }
+
+    private void startConsumer(String brokerName, ActiveMQDestination destination) throws Exception {
+        int id = Integer.parseInt(brokerName.substring(1));
+        ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory("tcp://localhost:" + (portBase + id));
+        connectionFactory.setWatchTopicAdvisories(false);
+        QueueConnection queueConnection = connectionFactory.createQueueConnection();
+        queueConnection.start();
+
+        queueConnection.createQueueSession(false, Session.AUTO_ACKNOWLEDGE).createConsumer(destination);
+        queueConnection.close();
+    }
+
+    private void dumpQueueStat(ActiveMQDestination destination) throws Exception {
+        Collection<BrokerItem> brokerList = brokers.values();
+        for (Iterator<BrokerItem> i = brokerList.iterator(); i.hasNext(); ) {
+            BrokerService brokerService = i.next().broker;
+            for (ObjectName objectName : brokerService.getAdminView().getQueues()) {
+                //if (objectName.toString().contains(destination.getQualifiedName())) {
+                    QueueViewMBean qViewMBean = (QueueViewMBean) brokerService.getManagementContext().newProxyInstance(objectName, QueueViewMBean.class, false);
+                    LOG.info(brokerService.getBrokerName() + ", " + qViewMBean.getName() + " Size: " + qViewMBean.getEnqueueCount());
+                //}
+            }
+        }
+    }
+
+    private void startAllGWFanoutConsumers(int nBrokers) throws Exception {
+
+        StringBuffer compositeDest = new StringBuffer();
+        for (int k = 0; k < nBrokers; k++) {
+            compositeDest.append("GW." + k);
+            if (k + 1 != nBrokers) {
+                compositeDest.append(',');
+            }
+        }
+        ActiveMQQueue compositeQ = new ActiveMQQueue(compositeDest.toString());
+
+        for (int id = 0; id < nBrokers; id++) {
+            ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory("failover:(tcp://localhost:" + (portBase + id) + ")");
+            connectionFactory.setWatchTopicAdvisories(false);
+
+            QueueConnection queueConnection = connectionFactory.createQueueConnection();
+            queueConnection.start();
+
+            final QueueSession queueSession = queueConnection.createQueueSession(true, Session.SESSION_TRANSACTED);
+
+            final MessageProducer producer = queueSession.createProducer(compositeQ);
+            queueSession.createReceiver(new ActiveMQQueue("IN")).setMessageListener(new MessageListener() {
+                @Override
+                public void onMessage(Message message) {
+                    try {
+                        producer.send(message);
+                        queueSession.commit();
+                    } catch (Exception e) {
+                        LOG.error("Failed to fanout to GW: " + message, e);
+                    }
+
+                }
+            });
+        }
+    }
+
+    private List<ConsumerState> startAllGWConsumers(int nBrokers) throws Exception {
+        List<ConsumerState> consumerStates = new LinkedList<ConsumerState>();
+        for (int id = 0; id < nBrokers; id++) {
+            ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory("failover:(tcp://localhost:" + (portBase + id) + ")");
+            connectionFactory.setWatchTopicAdvisories(false);
+
+            QueueConnection queueConnection = connectionFactory.createQueueConnection();
+            queueConnection.start();
+
+            final QueueSession queueSession = queueConnection.createQueueSession(false, Session.AUTO_ACKNOWLEDGE);
+
+            ActiveMQQueue destination = new ActiveMQQueue("GW." + id);
+            QueueReceiver queueReceiver = queueSession.createReceiver(destination);
+
+            final ConsumerState consumerState = new ConsumerState();
+            consumerState.brokerName = ((ActiveMQConnection) queueConnection).getBrokerName();
+            consumerState.receiver = queueReceiver;
+            consumerState.destination = destination;
+            for (int j = 0; j < numMessages * (consumerState.destination.isComposite() ? consumerState.destination.getCompositeDestinations().length : 1); j++) {
+                consumerState.expected.add(j);
+            }
+
+            if (!accumulators.containsKey(destination)) {
+                accumulators.put(destination, new AtomicInteger(0));
+            }
+            consumerState.accumulator = accumulators.get(destination);
+
+            queueReceiver.setMessageListener(new MessageListener() {
+                @Override
+                public void onMessage(Message message) {
+                    try {
+                        if (consumerSleepTime > 0) {
+                            TimeUnit.MILLISECONDS.sleep(consumerSleepTime);
+                        }
+                    } catch (InterruptedException e) {
+                        e.printStackTrace();
+                    }
+                    try {
+                        consumerState.accumulator.incrementAndGet();
+                        try {
+                            consumerState.expected.remove(((ActiveMQMessage) message).getProperty("NUM"));
+                        } catch (IOException e) {
+                            e.printStackTrace();
+                        }
+                        //queueSession.commit();
+                    } catch (Exception e) {
+                        LOG.error("Failed to commit slow receipt of " + message, e);
+                    }
+                }
+            });
+
+            consumerStates.add(consumerState);
+
+        }
+        return consumerStates;
+    }
+
+    private void produce(final int numMessages) throws Exception {
+        ExecutorService executorService = Executors.newFixedThreadPool(numProducers);
+        final AtomicInteger toSend = new AtomicInteger(numMessages);
+        for (int i = 1; i <= numProducers; i++) {
+            final int id = i % numBrokers;
+            executorService.execute(new Runnable() {
+                @Override
+                public void run() {
+                    try {
+                        ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory("failover:(tcp://localhost:" + (portBase + id) + ")");
+                        connectionFactory.setWatchTopicAdvisories(false);
+                        QueueConnection queueConnection = connectionFactory.createQueueConnection();
+                        queueConnection.start();
+                        QueueSession queueSession = queueConnection.createQueueSession(false, Session.AUTO_ACKNOWLEDGE);
+                        MessageProducer producer = queueSession.createProducer(null);
+                        int val = 0;
+                        while ((val = toSend.decrementAndGet()) >= 0) {
+
+                            int id = numMessages - val - 1;
+
+                            ActiveMQQueue compositeQ = new ActiveMQQueue("IN");
+                            //LOG.info("Send to: " + ((ActiveMQConnection) queueConnection).getBrokerName() + ", " + val + ", dest:" + compositeQ);
+                            Message textMessage = queueSession.createTextMessage(((ActiveMQConnection) queueConnection).getBrokerName() + "->" + id + " payload:" + payload);
+                            textMessage.setIntProperty("NUM", id);
+                            producer.send(compositeQ, textMessage);
+                        }
+                        queueConnection.close();
+
+                    } catch (Throwable throwable) {
+                        throwable.printStackTrace();
+                        exceptions.add(throwable);
+                    }
+                }
+            });
+        }
+    }
+
+    private void verifyPeerBrokerInfo(BrokerItem brokerItem, final int max) throws Exception {
+        final BrokerService broker = brokerItem.broker;
+        final RegionBroker regionBroker = (RegionBroker) broker.getRegionBroker();
+        Wait.waitFor(new Wait.Condition() {
+            @Override
+            public boolean isSatisified() throws Exception {
+                LOG.info("verify infos " + broker.getBrokerName() + ", len: " + regionBroker.getPeerBrokerInfos().length);
+                return max == regionBroker.getPeerBrokerInfos().length;
+            }
+        });
+        LOG.info("verify infos " + broker.getBrokerName() + ", len: " + regionBroker.getPeerBrokerInfos().length);
+        List<String> missing = new ArrayList<String>();
+        for (int i = 0; i < max; i++) {
+            missing.add("B" + i);
+        }
+        if (max != regionBroker.getPeerBrokerInfos().length) {
+            for (BrokerInfo info : regionBroker.getPeerBrokerInfos()) {
+                LOG.info(info.getBrokerName());
+                missing.remove(info.getBrokerName());
+            }
+            LOG.info("Broker infos off.." + missing);
+        }
+        assertEquals(broker.getBrokerName(), max, regionBroker.getPeerBrokerInfos().length);
+    }
+
+    private void verifyPeerBrokerInfos(final int max) throws Exception {
+        Collection<BrokerItem> brokerList = brokers.values();
+        for (Iterator<BrokerItem> i = brokerList.iterator(); i.hasNext(); ) {
+            verifyPeerBrokerInfo(i.next(), max);
+        }
+    }
+
+    protected void tearDown() throws Exception {
+        super.tearDown();
+    }
+
+    class ConsumerState {
+        AtomicInteger accumulator;
+        String brokerName;
+        QueueReceiver receiver;
+        ActiveMQDestination destination;
+        ConcurrentLinkedQueue<Integer> expected = new ConcurrentLinkedQueue<Integer>();
+    }
+}
\ No newline at end of file

http://git-wip-us.apache.org/repos/asf/activemq/blob/54e2e3be/activemq-unit-tests/src/test/java/org/apache/activemq/bugs/AMQ5212Test.java
----------------------------------------------------------------------
diff --git a/activemq-unit-tests/src/test/java/org/apache/activemq/bugs/AMQ5212Test.java b/activemq-unit-tests/src/test/java/org/apache/activemq/bugs/AMQ5212Test.java
index 64d57a5..4c07655 100644
--- a/activemq-unit-tests/src/test/java/org/apache/activemq/bugs/AMQ5212Test.java
+++ b/activemq-unit-tests/src/test/java/org/apache/activemq/bugs/AMQ5212Test.java
@@ -17,6 +17,7 @@
 
 package org.apache.activemq.bugs;
 
+import java.util.Arrays;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
 import java.util.concurrent.TimeUnit;
@@ -31,18 +32,31 @@ import org.apache.activemq.ActiveMQSession;
 import org.apache.activemq.broker.BrokerService;
 import org.apache.activemq.command.ActiveMQQueue;
 import org.apache.activemq.command.ActiveMQTextMessage;
+import org.apache.activemq.store.kahadb.KahaDBPersistenceAdapter;
+import org.apache.activemq.util.Wait;
 import org.junit.After;
 import org.junit.Before;
 import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.junit.runners.Parameterized;
 
 
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertNotNull;
 
+@RunWith(value = Parameterized.class)
 public class AMQ5212Test {
 
     BrokerService brokerService;
 
+    @Parameterized.Parameter(0)
+    public boolean concurrentStoreAndDispatchQ = true;
+
+    @Parameterized.Parameters(name = "concurrentStoreAndDispatch={0}")
+    public static Iterable<Object[]> getTestParameters() {
+        return Arrays.asList(new Object[][]{{Boolean.TRUE}, {Boolean.FALSE}});
+    }
+
     @Before
     public void setUp() throws Exception {
         start(true);
@@ -53,6 +67,7 @@ public class AMQ5212Test {
         if (deleteAllMessages) {
             brokerService.deleteAllMessages();
         }
+        ((KahaDBPersistenceAdapter)brokerService.getPersistenceAdapter()).setConcurrentStoreAndDispatchQueues(concurrentStoreAndDispatchQ);
         brokerService.addConnector("tcp://localhost:0");
         brokerService.setAdvisorySupport(false);
         brokerService.start();
@@ -118,6 +133,12 @@ public class AMQ5212Test {
         executorService.shutdown();
         executorService.awaitTermination(5, TimeUnit.MINUTES);
 
+        Wait.waitFor(new Wait.Condition() {
+            @Override
+            public boolean isSatisified() throws Exception {
+                return expectedTotalEnqueue == brokerService.getAdminView().getTotalEnqueueCount();
+            }
+        });
         assertEquals("total enqueue as expected", expectedTotalEnqueue, brokerService.getAdminView().getTotalEnqueueCount());
     }
 

http://git-wip-us.apache.org/repos/asf/activemq/blob/54e2e3be/activemq-unit-tests/src/test/java/org/apache/activemq/bugs/AMQ5266Test.java
----------------------------------------------------------------------
diff --git a/activemq-unit-tests/src/test/java/org/apache/activemq/bugs/AMQ5266Test.java b/activemq-unit-tests/src/test/java/org/apache/activemq/bugs/AMQ5266Test.java
index e819b6f..626fe6e 100644
--- a/activemq-unit-tests/src/test/java/org/apache/activemq/bugs/AMQ5266Test.java
+++ b/activemq-unit-tests/src/test/java/org/apache/activemq/bugs/AMQ5266Test.java
@@ -17,13 +17,14 @@
 package org.apache.activemq.bugs;
 
 import java.util.ArrayList;
+import java.util.Arrays;
 import java.util.Collections;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
-import java.util.Set;
 import java.util.TreeSet;
 import java.util.UUID;
+import java.util.concurrent.TimeUnit;
 import javax.jms.Message;
 import javax.jms.MessageConsumer;
 import javax.jms.MessageProducer;
@@ -41,10 +42,13 @@ import org.apache.activemq.broker.region.policy.PolicyMap;
 import org.apache.activemq.command.ActiveMQQueue;
 import org.apache.activemq.store.jdbc.JDBCPersistenceAdapter;
 import org.apache.activemq.store.jdbc.adapter.DefaultJDBCAdapter;
+import org.apache.activemq.store.kahadb.KahaDBPersistenceAdapter;
 import org.apache.derby.jdbc.EmbeddedDataSource;
 import org.junit.After;
 import org.junit.Before;
 import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.junit.runners.Parameterized;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -56,12 +60,54 @@ import static org.junit.Assert.assertEquals;
  * <p/>
  * Will kick of publisher and consumer simultaneously, and will usually result in stuck messages on the queue.
  */
+@RunWith(Parameterized.class)
 public class AMQ5266Test {
     static Logger LOG = LoggerFactory.getLogger(AMQ5266Test.class);
     String activemqURL = "tcp://localhost:61617";
     BrokerService brokerService;
     private EmbeddedDataSource dataSource;
 
+    public int messageSize = 1000;
+
+    @Parameterized.Parameter(0)
+    public int publisherMessagesPerThread = 1000;
+
+    @Parameterized.Parameter(1)
+    public int publisherThreadCount = 20;
+
+    @Parameterized.Parameter(2)
+    public int consumerThreadsPerQueue = 5;
+
+    @Parameterized.Parameter(3)
+    public int destMemoryLimit = 50 * 1024;
+
+    @Parameterized.Parameter(4)
+    public boolean useCache = true;
+
+    @Parameterized.Parameter(5)
+    public boolean useDefaultStore = false;
+
+    @Parameterized.Parameters(name="#{0},producerThreads:{1},consumerThreads:{2},mL:{3},useCache:{4},useDefaultStore:{5}")
+    public static Iterable<Object[]> parameters() {
+        return Arrays.asList(new Object[][]{
+                // jdbc
+                {1000, 20,  5,   50*1024,   true,  false},
+                {100,  20,  5,   50*1024,   false, false},
+                {1000, 5,   20,  50*1024,   true,  false},
+                {1000, 20,  20,  1024*1024, true,  false},
+                {1000, 100, 100, 1024*1024, true,  false},
+
+                // default store
+                {1000, 20,  5,   50*1024,   true,  true},
+                {100,  20,  5,   50*1024,   false, true},
+                {1000, 5,   20,  50*1024,   true,  true},
+                {1000, 20,  20,  1024*1024, true,  true},
+                {1000, 100, 100, 1024*1024, true,  true}
+        });
+    }
+
+    public int consumerBatchSize = 5;
+
     @Before
     public void startBroker() throws Exception {
         brokerService = new BrokerService();
@@ -70,19 +116,28 @@ public class AMQ5266Test {
         dataSource.setDatabaseName("target/derbyDb");
         dataSource.setCreateDatabase("create");
 
-        JDBCPersistenceAdapter persistenceAdapter = new JDBCPersistenceAdapter();
-        persistenceAdapter.setDataSource(dataSource);
-        brokerService.setPersistenceAdapter(persistenceAdapter);
+        JDBCPersistenceAdapter jdbcPersistenceAdapter = new JDBCPersistenceAdapter();
+        jdbcPersistenceAdapter.setDataSource(dataSource);
+        jdbcPersistenceAdapter.setUseLock(false);
+
+        if (!useDefaultStore) {
+            brokerService.setPersistenceAdapter(jdbcPersistenceAdapter);
+        } else {
+            KahaDBPersistenceAdapter kahaDBPersistenceAdapter = (KahaDBPersistenceAdapter) brokerService.getPersistenceAdapter();
+            kahaDBPersistenceAdapter.setConcurrentStoreAndDispatchQueues(true);
+        }
         brokerService.setDeleteAllMessagesOnStartup(true);
 
 
         PolicyMap policyMap = new PolicyMap();
         PolicyEntry defaultEntry = new PolicyEntry();
-        defaultEntry.setEnableAudit(false);
-        defaultEntry.setUseCache(false);
+        defaultEntry.setUseConsumerPriority(false); // java.lang.IllegalArgumentException: Comparison method violates its general contract!
+        defaultEntry.setMaxAuditDepth(publisherThreadCount);
+        defaultEntry.setEnableAudit(true);
+        defaultEntry.setUseCache(useCache);
         defaultEntry.setMaxPageSize(1000);
         defaultEntry.setOptimizedDispatch(false);
-        defaultEntry.setMemoryLimit(1024 * 1024);
+        defaultEntry.setMemoryLimit(destMemoryLimit);
         defaultEntry.setExpireMessagesPeriod(0);
         policyMap.setDefaultEntry(defaultEntry);
         brokerService.setDestinationPolicy(policyMap);
@@ -110,11 +165,6 @@ public class AMQ5266Test {
 
         String activemqQueues = "activemq,activemq2";//,activemq3,activemq4,activemq5,activemq6,activemq7,activemq8,activemq9";
 
-        int publisherMessagesPerThread = 1000;
-        int publisherThreadCount = 5;
-
-        int consumerThreadsPerQueue = 5;
-        int consumerBatchSize = 25;
         int consumerWaitForConsumption = 5 * 60 * 1000;
 
         ExportQueuePublisher publisher = null;
@@ -145,28 +195,36 @@ public class AMQ5266Test {
 
         publisher.waitForCompletion();
 
-        distinctPublishedCount = publisher.getIDs().size();
+        List publishedIds = publisher.getIDs();
+        distinctPublishedCount = new TreeSet(publishedIds).size();
 
-        LOG.info("Publisher Complete. Distinct IDs Published: " + distinctPublishedCount);
+        LOG.info("Publisher Complete. Published: " + publishedIds.size() + ", Distinct IDs Published: " + distinctPublishedCount);
 
 
         long endWait = System.currentTimeMillis() + consumerWaitForConsumption;
-
-
         while (!consumer.completed() && System.currentTimeMillis() < endWait) {
             try {
                 int secs = (int) (endWait - System.currentTimeMillis()) / 1000;
                 LOG.info("Waiting For Consumer Completion. Time left: " + secs + " secs");
-                DefaultJDBCAdapter.dumpTables(dataSource.getConnection());
+                if (!useDefaultStore) {
+                    DefaultJDBCAdapter.dumpTables(dataSource.getConnection());
+                }
                 Thread.sleep(10000);
             } catch (Exception e) {
             }
         }
 
-        LOG.info("\nConsumer Complete. Shutting Down.");
+        LOG.info("\nConsumer Complete: " + consumer.completed() +", Shutting Down.");
 
         consumer.shutdown();
 
+        TimeUnit.SECONDS.sleep(2);
+        LOG.info("DB Contents START");
+        if (!useDefaultStore) {
+            DefaultJDBCAdapter.dumpTables(dataSource.getConnection());
+        }
+        LOG.info("DB Contents END");
+
         LOG.info("Consumer Stats:");
 
         for (Map.Entry<String, List<String>> entry : consumer.getIDs().entrySet()) {
@@ -199,7 +257,8 @@ public class AMQ5266Test {
         // Collection of distinct IDs that the publisher has published.
         // After a message is published, its UUID will be written to this list for tracking.
         // This list of IDs (or distinct count) will be used to compare to the consumed list of IDs.
-        private Set<String> ids = Collections.synchronizedSet(new TreeSet<String>());
+        //private Set<String> ids = Collections.synchronizedSet(new TreeSet<String>());
+        private List<String> ids = Collections.synchronizedList(new ArrayList<String>());
         private List<PublisherThread> threads;
 
         public ExportQueuePublisher(String activemqURL, String activemqQueues, int messagesPerThread, int threadCount) throws Exception {
@@ -216,7 +275,7 @@ public class AMQ5266Test {
             }
         }
 
-        public Set<String> getIDs() {
+        public List<String> getIDs() {
             return ids;
         }
 
@@ -241,7 +300,7 @@ public class AMQ5266Test {
             return queueConnection.createSession(true, Session.SESSION_TRANSACTED);
         }
 
-        private QueueConnection newQueueConnection() throws Exception {
+        private synchronized QueueConnection newQueueConnection() throws Exception {
 
             if (connectionFactory == null) {
                 connectionFactory = new ActiveMQConnectionFactory(amqUser, amqPassword, activemqURL);
@@ -285,7 +344,7 @@ public class AMQ5266Test {
                     // Loop until we've published enough messages
                     while (count-- > 0) {
 
-                        TextMessage tm = session.createTextMessage("test");
+                        TextMessage tm = session.createTextMessage(getMessageText());
                         String id = UUID.randomUUID().toString();
                         tm.setStringProperty("KEY", id);
                         ids.add(id);                            // keep track of the key to compare against consumer
@@ -320,6 +379,28 @@ public class AMQ5266Test {
 
     }
 
+    String messageText;
+    private String getMessageText() {
+
+        if (messageText == null) {
+
+            synchronized (this) {
+
+                if (messageText == null) {
+
+                    StringBuilder sb = new StringBuilder();
+                    for (int i = 0; i < messageSize; i++) {
+                        sb.append("X");
+                    }
+                    messageText = sb.toString();
+                }
+            }
+        }
+
+        return messageText;
+    }
+
+
     public class ExportQueueConsumer {
 
         private final String amqUser = ActiveMQConnection.DEFAULT_USER;
@@ -404,7 +485,7 @@ public class AMQ5266Test {
             return queueConnection.createSession(true, Session.SESSION_TRANSACTED);
         }
 
-        private QueueConnection newQueueConnection() throws Exception {
+        private synchronized QueueConnection newQueueConnection() throws Exception {
 
             if (connectionFactory == null) {
                 connectionFactory = new ActiveMQConnectionFactory(amqUser, amqPassword, activemqURL);
@@ -452,7 +533,7 @@ public class AMQ5266Test {
                 qName = queueName;
                 qc = newQueueConnection();
                 session = newSession(qc);
-                Queue q = session.createQueue(queueName);
+                Queue q = session.createQueue(queueName + "?consumer.prefetchSize=" + batchSize);
                 mc = session.createConsumer(q);
 
                 idList = idsByQueue.get(queueName);
@@ -469,6 +550,7 @@ public class AMQ5266Test {
 
                         if (idList.size() >= totalToExpect) {
                             LOG.info("Got {} for q: {}", +idList.size(), qName);
+                            session.commit();
                             break;
                         }
                         Message m = mc.receive(4000);
@@ -497,7 +579,9 @@ public class AMQ5266Test {
                             // Sleep a little before trying to read after not getting a message
 
                             try {
-                                LOG.info("did not receive on {}, current count: {}", qName, idList.size());
+                                if (idList.size() < totalToExpect) {
+                                    LOG.info("did not receive on {}, current count: {}", qName, idList.size());
+                                }
                                 //sleep(3000);
                             } catch (Exception e) {
                             }

http://git-wip-us.apache.org/repos/asf/activemq/blob/54e2e3be/activemq-unit-tests/src/test/java/org/apache/activemq/network/DemandForwardingBridgeTest.java
----------------------------------------------------------------------
diff --git a/activemq-unit-tests/src/test/java/org/apache/activemq/network/DemandForwardingBridgeTest.java b/activemq-unit-tests/src/test/java/org/apache/activemq/network/DemandForwardingBridgeTest.java
index 02b207c..9794337 100755
--- a/activemq-unit-tests/src/test/java/org/apache/activemq/network/DemandForwardingBridgeTest.java
+++ b/activemq-unit-tests/src/test/java/org/apache/activemq/network/DemandForwardingBridgeTest.java
@@ -106,7 +106,6 @@ public class DemandForwardingBridgeTest extends NetworkTestSupport {
         }));
 
         assertEquals("broker dest stat dequeues", 1, destinationStatistics.getDequeues().getCount());
-        assertEquals("remote broker dest stat dequeues", 1, remoteBroker.getDestination(destination).getDestinationStatistics().getDequeues().getCount());
     }
 
     public void initCombosForTestAddConsumerThenSend() {

http://git-wip-us.apache.org/repos/asf/activemq/blob/54e2e3be/activemq-unit-tests/src/test/java/org/apache/activemq/usecases/MemoryLimitTest.java
----------------------------------------------------------------------
diff --git a/activemq-unit-tests/src/test/java/org/apache/activemq/usecases/MemoryLimitTest.java b/activemq-unit-tests/src/test/java/org/apache/activemq/usecases/MemoryLimitTest.java
index 21c39ef..3f0ea77 100644
--- a/activemq-unit-tests/src/test/java/org/apache/activemq/usecases/MemoryLimitTest.java
+++ b/activemq-unit-tests/src/test/java/org/apache/activemq/usecases/MemoryLimitTest.java
@@ -123,7 +123,7 @@ public class MemoryLimitTest extends TestSupport {
 
         // consume one message
         MessageConsumer consumer = sess.createConsumer(queue);
-        Message msg = consumer.receive();
+        Message msg = consumer.receive(5000);
         msg.acknowledge();
 
         // this should free some space and allow us to get new batch of messages in the memory
@@ -132,12 +132,12 @@ public class MemoryLimitTest extends TestSupport {
             @Override
             public boolean isSatisified() throws Exception {
                 LOG.info("Destination usage: " + dest.getMemoryUsage());
-                return dest.getMemoryUsage().getPercentUsage() >= 478;
+                return dest.getMemoryUsage().getPercentUsage() >= 470;
             }
         }));
 
         LOG.info("Broker usage: " + broker.getSystemUsage().getMemoryUsage());
-        assertTrue(broker.getSystemUsage().getMemoryUsage().getPercentUsage() >= 478);
+        assertTrue(broker.getSystemUsage().getMemoryUsage().getPercentUsage() >= 470);
 
         // let's make sure we can consume all messages
         for (int i = 1; i < 2000; i++) {


Mime
View raw message