diff --git a/activemq-broker/src/main/java/org/apache/activemq/advisory/AdvisoryBroker.java b/activemq-broker/src/main/java/org/apache/activemq/advisory/AdvisoryBroker.java
index b2d38d67a0..63d29a38fe 100755
--- a/activemq-broker/src/main/java/org/apache/activemq/advisory/AdvisoryBroker.java
+++ b/activemq-broker/src/main/java/org/apache/activemq/advisory/AdvisoryBroker.java
@@ -26,6 +26,7 @@ import java.util.Map;
 import java.util.Set;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ConcurrentMap;
+import java.util.concurrent.atomic.AtomicInteger;
 import java.util.concurrent.locks.ReentrantReadWriteLock;
 
 import org.apache.activemq.broker.Broker;
@@ -66,6 +67,7 @@ import org.apache.activemq.usage.Usage;
 import org.apache.activemq.util.IdGenerator;
 import org.apache.activemq.util.LongSequenceGenerator;
 import org.apache.activemq.util.SubscriptionKey;
+import org.apache.activemq.util.ThreadTracker;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -104,6 +106,11 @@ public class AdvisoryBroker extends BrokerFilter {
     protected final ConcurrentMap<BrokerInfo, ActiveMQMessage> networkBridges = new ConcurrentHashMap<BrokerInfo, ActiveMQMessage>();
     protected final ProducerId advisoryProducerId = new ProducerId();
 
+
+    public final static AtomicInteger adds = new AtomicInteger();
+    public final static AtomicInteger removes = new AtomicInteger();
+
+    final String brokerName;
     private final LongSequenceGenerator messageIdGenerator = new LongSequenceGenerator();
 
     private VirtualDestinationMatcher virtualDestinationMatcher = new DestinationFilterVirtualDestinationMatcher();
@@ -111,6 +118,7 @@ public class AdvisoryBroker extends BrokerFilter {
     public AdvisoryBroker(Broker next) {
         super(next);
         advisoryProducerId.setConnectionId(ID_GENERATOR.generateId());
+        brokerName = getBrokerName();
     }
 
     @Override
@@ -147,6 +155,8 @@ public class AdvisoryBroker extends BrokerFilter {
             } finally {
                 consumersLock.writeLock().unlock();
             }
+//            ThreadTracker.track("AddAdvisory");
+          //  adds.incrementAndGet();
             fireConsumerAdvisory(context, info.getDestination(), topic, info);
         } else {
             // We need to replay all the previously collected state objects
@@ -367,6 +377,8 @@ public class AdvisoryBroker extends BrokerFilter {
                 consumersLock.writeLock().unlock();
             }
             if (!dest.isTemporary() || destinations.containsKey(dest)) {
+                //ThreadTracker.track(brokerName + "-RemoveAdvisory");
+                // removes.incrementAndGet();
                 fireConsumerAdvisory(context, dest, topic, info.createRemoveCommand());
             }
         }
diff --git a/activemq-broker/src/main/java/org/apache/activemq/broker/TransportConnection.java b/activemq-broker/src/main/java/org/apache/activemq/broker/TransportConnection.java
index 43ea5bc132..3ff12658f5 100755
--- a/activemq-broker/src/main/java/org/apache/activemq/broker/TransportConnection.java
+++ b/activemq-broker/src/main/java/org/apache/activemq/broker/TransportConnection.java
@@ -698,6 +698,8 @@ public class TransportConnection implements Connection, Task, CommandVisitor {
                 ss.addConsumer(info);
                 addConsumerBrokerExchange(cs, info.getConsumerId());
             } catch (IllegalStateException e) {
+                e.printStackTrace();
+                LOG.error("Failed to add consumer", e);
                 broker.removeConsumer(cs.getContext(), info);
             }
 
@@ -1023,6 +1025,7 @@ public class TransportConnection implements Connection, Task, CommandVisitor {
             }
             return false;
         } catch (IOException e) {
+            LOG.error("Error:" + e);
             if (dispatchStopped.compareAndSet(false, true)) {
                 dispatchStoppedLatch.countDown();
             }
diff --git a/activemq-broker/src/main/java/org/apache/activemq/broker/region/PrefetchSubscription.java b/activemq-broker/src/main/java/org/apache/activemq/broker/region/PrefetchSubscription.java
index 227ae2d2d3..64431d49e2 100755
--- a/activemq-broker/src/main/java/org/apache/activemq/broker/region/PrefetchSubscription.java
+++ b/activemq-broker/src/main/java/org/apache/activemq/broker/region/PrefetchSubscription.java
@@ -63,6 +63,7 @@ public abstract class PrefetchSubscription extends AbstractSubscription {
     protected final Object dispatchLock = new Object();
     private final CountDownLatch okForAckAsDispatchDone = new CountDownLatch(1);
 
+    private int count = 0;
     public PrefetchSubscription(Broker broker, SystemUsage usageManager, ConnectionContext context, ConsumerInfo info, PendingMessageCursor cursor) throws JMSException {
         super(broker,context, info);
         this.usageManager=usageManager;
diff --git a/activemq-broker/src/main/java/org/apache/activemq/network/ConduitBridge.java b/activemq-broker/src/main/java/org/apache/activemq/network/ConduitBridge.java
index 10df1f92e6..e64a3ece6d 100644
--- a/activemq-broker/src/main/java/org/apache/activemq/network/ConduitBridge.java
+++ b/activemq-broker/src/main/java/org/apache/activemq/network/ConduitBridge.java
@@ -25,6 +25,7 @@ import org.apache.activemq.command.ConsumerInfo;
 import org.apache.activemq.command.SubscriptionInfo;
 import org.apache.activemq.filter.DestinationFilter;
 import org.apache.activemq.transport.Transport;
+import org.apache.activemq.util.ThreadTracker;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -68,6 +69,7 @@ public class ConduitBridge extends DemandForwardingBridge {
                 LOG.debug("{} {} with ids {} matched (add interest) {}", new Object[]{
                         configuration.getBrokerName(), info, info.getNetworkConsumerIds(), ds
                 });
+//                ThreadTracker.track("addDS");
                 // add the interest in the subscription
                 if (!info.isDurable()) {
                     ds.add(info.getConsumerId());
@@ -89,6 +91,10 @@ public class ConduitBridge extends DemandForwardingBridge {
 
     @Override
     protected void removeDemandSubscription(ConsumerId id) throws IOException {
+        LOG.debug(brokerService.getBrokerName() + " remove: info: " + id);
+//        if (!id.toString().contains("->")) {
+//            ThreadTracker.track("removeDS");
+//        }
         List<DemandSubscription> tmpList = new ArrayList<DemandSubscription>();
 
         for (DemandSubscription ds : subscriptionMapByLocalId.values()) {
@@ -96,9 +102,13 @@ public class ConduitBridge extends DemandForwardingBridge {
                 LOG.debug("{} on {} from {} removed interest for: {} from {}", new Object[]{
                         configuration.getBrokerName(), localBroker, remoteBrokerName, id, ds
                 });
-            }
-            if (ds.isEmpty()) {
-                tmpList.add(ds);
+                //}
+                if (ds.isEmpty()) {
+                    if (id.toString().contains("->")) {
+                        LOG.error("BINGO on remove for network sub:" + id);
+                    }
+                    tmpList.add(ds);
+                }
             }
         }
 
diff --git a/activemq-broker/src/main/java/org/apache/activemq/network/DemandForwardingBridgeSupport.java b/activemq-broker/src/main/java/org/apache/activemq/network/DemandForwardingBridgeSupport.java
index 322a0922b8..77fdd32edd 100644
--- a/activemq-broker/src/main/java/org/apache/activemq/network/DemandForwardingBridgeSupport.java
+++ b/activemq-broker/src/main/java/org/apache/activemq/network/DemandForwardingBridgeSupport.java
@@ -93,12 +93,7 @@ import org.apache.activemq.transport.TransportFilter;
 import org.apache.activemq.transport.nio.NIOSSLTransport;
 import org.apache.activemq.transport.failover.FailoverTransport;
 import org.apache.activemq.transport.tcp.SslTransport;
-import org.apache.activemq.util.IdGenerator;
-import org.apache.activemq.util.IntrospectionSupport;
-import org.apache.activemq.util.LongSequenceGenerator;
-import org.apache.activemq.util.MarshallingSupport;
-import org.apache.activemq.util.ServiceStopper;
-import org.apache.activemq.util.ServiceSupport;
+import org.apache.activemq.util.*;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -800,6 +795,8 @@ public abstract class DemandForwardingBridgeSupport implements NetworkBridge, Br
                 return;
             }
 
+//            ThreadTracker.track(configuration.getBrokerName() + "-ConsumerInfo");
+
             // in a cyclic network there can be multiple bridges per broker that can propagate
             // a network subscription so there is a need to synchronize on a shared entity
             // if duplicate suppression is required
@@ -851,6 +848,7 @@ public abstract class DemandForwardingBridgeSupport implements NetworkBridge, Br
                 localBroker.oneway(destInfo);
             }
         } else if (data.getClass() == RemoveInfo.class) {
+//            ThreadTracker.track(configuration.getBrokerName() + "-RemoveConsumerInfo");
             ConsumerId id = (ConsumerId) ((RemoveInfo) data).getObjectId();
             removeDemandSubscription(id);
         } else if (data.getClass() == RemoveSubscriptionInfo.class) {
@@ -954,7 +952,7 @@ public abstract class DemandForwardingBridgeSupport implements NetworkBridge, Br
 
     protected void removeSubscription(final DemandSubscription sub) throws IOException {
         if (sub != null) {
-            LOG.trace("{} remove local subscription: {} for remote {}", new Object[]{configuration.getBrokerName(), sub.getLocalInfo().getConsumerId(), sub.getRemoteInfo().getConsumerId()});
+            LOG.debug("{} remove local subscription: {} for remote {}", new Object[]{configuration.getBrokerName(), sub.getLocalInfo().getConsumerId(), sub.getRemoteInfo().getConsumerId()});
 
             // ensure not available for conduit subs pending removal
             subscriptionMapByLocalId.remove(sub.getLocalInfo().getConsumerId());
@@ -1221,6 +1219,7 @@ public abstract class DemandForwardingBridgeSupport implements NetworkBridge, Br
                 if (consumerInfo.isDurable()) {
                     sub.getDurableRemoteSubs().add(new SubscriptionInfo(sub.getRemoteInfo().getClientId(), consumerInfo.getSubscriptionName()));
                 }
+                //ThreadTracker.track("newDS");
                 addSubscription(sub);
                 LOG.debug("{} new demand subscription: {}", configuration.getBrokerName(), sub);
             }
diff --git a/activemq-broker/src/main/java/org/apache/activemq/network/DemandSubscription.java b/activemq-broker/src/main/java/org/apache/activemq/network/DemandSubscription.java
index b53646ae7c..9816821ab4 100644
--- a/activemq-broker/src/main/java/org/apache/activemq/network/DemandSubscription.java
+++ b/activemq-broker/src/main/java/org/apache/activemq/network/DemandSubscription.java
@@ -16,6 +16,7 @@
  */
 package org.apache.activemq.network;
 
+import java.util.HashSet;
 import java.util.Set;
 import java.util.concurrent.CopyOnWriteArraySet;
 import java.util.concurrent.TimeUnit;
@@ -55,7 +56,7 @@ public class DemandSubscription {
 
     @Override
     public String toString() {
-        return "DemandSub{" + localInfo.getConsumerId() + ",remotes:" + remoteSubsIds + "}";
+        return "DemandSub{" + localInfo.getConsumerId() + ",remotes:" + (remoteSubsIds.size() > 10 ? remoteSubsIds.size() : remoteSubsIds) + "}";
     }
 
     /**
diff --git a/activemq-broker/src/main/java/org/apache/activemq/network/DurableConduitBridge.java b/activemq-broker/src/main/java/org/apache/activemq/network/DurableConduitBridge.java
index 4b6a6c7272..1007b3df89 100644
--- a/activemq-broker/src/main/java/org/apache/activemq/network/DurableConduitBridge.java
+++ b/activemq-broker/src/main/java/org/apache/activemq/network/DurableConduitBridge.java
@@ -26,6 +26,7 @@ import org.apache.activemq.command.ConsumerId;
 import org.apache.activemq.command.ConsumerInfo;
 import org.apache.activemq.filter.DestinationFilter;
 import org.apache.activemq.transport.Transport;
+import org.apache.activemq.util.ThreadTracker;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -92,6 +93,8 @@ public class DurableConduitBridge extends ConduitBridge {
 
     @Override
     protected DemandSubscription createDemandSubscription(ConsumerInfo info) throws IOException {
+        LOG.debug(brokerService.getBrokerName() + " create - info: " + info);
+//        ThreadTracker.track("create");
         if (addToAlreadyInterestedConsumers(info)) {
             return null; // don't want this subscription added
         }
diff --git a/activemq-client/src/main/java/org/apache/activemq/transport/failover/FailoverTransport.java b/activemq-client/src/main/java/org/apache/activemq/transport/failover/FailoverTransport.java
index d26102f5c7..d022c32e60 100755
--- a/activemq-client/src/main/java/org/apache/activemq/transport/failover/FailoverTransport.java
+++ b/activemq-client/src/main/java/org/apache/activemq/transport/failover/FailoverTransport.java
@@ -277,7 +277,7 @@ public class FailoverTransport implements CompositeTransport {
                 }
 
                 LOG.warn("Transport ({}) failed {} attempting to automatically reconnect: {}",
-                         connectedTransportURI, (reconnectOk ? "," : ", not"), e);
+                         connectedTransportURI, (reconnectOk ? "," : ", not"), e, e);
 
                 failedConnectTransportURI = connectedTransportURI;
                 connectedTransportURI = null;
@@ -947,7 +947,7 @@ public class FailoverTransport implements CompositeTransport {
                             doRebalance = false;
                             return false;
                         } else {
-                            LOG.debug("Doing rebalance from: {} to {}", connectedTransportURI, connectList);
+                            LOG.info("Doing rebalance from: {} to {}", connectedTransportURI, connectList);
 
                             try {
                                 Transport transport = this.connectedTransport.getAndSet(null);
diff --git a/activemq-client/src/main/java/org/apache/activemq/transport/tcp/TcpTransport.java b/activemq-client/src/main/java/org/apache/activemq/transport/tcp/TcpTransport.java
index 8f515a8b1a..320de72991 100755
--- a/activemq-client/src/main/java/org/apache/activemq/transport/tcp/TcpTransport.java
+++ b/activemq-client/src/main/java/org/apache/activemq/transport/tcp/TcpTransport.java
@@ -566,9 +566,9 @@ public class TcpTransport extends TransportThreadSupport implements Transport, S
                     socket.close();
                     LOG.debug("Closed socket {}", socket);
                 } catch (IOException e) {
-                    if (LOG.isDebugEnabled()) {
-                        LOG.debug("Caught exception closing socket " + socket + ". This exception will be ignored.", e);
-                    }
+                    //if (LOG.isDebugEnabled()) {
+                        LOG.error("Caught exception closing socket " + socket + ". This exception will be ignored.", e);
+                    //}
                 }
             }
         }
diff --git a/activemq-unit-tests/src/test/java/org/apache/activemq/usecases/ClientRebalanceAdvisoryTest.java b/activemq-unit-tests/src/test/java/org/apache/activemq/usecases/ClientRebalanceAdvisoryTest.java
new file mode 100644
index 0000000000..5a781b8d79
--- /dev/null
+++ b/activemq-unit-tests/src/test/java/org/apache/activemq/usecases/ClientRebalanceAdvisoryTest.java
@@ -0,0 +1,226 @@
+/**
+ * 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.usecases;
+
+import org.apache.activemq.ActiveMQConnectionFactory;
+import org.apache.activemq.JmsMultipleBrokersTestSupport;
+import org.apache.activemq.broker.region.Destination;
+import org.apache.activemq.broker.region.RegionBroker;
+import org.apache.activemq.broker.region.Subscription;
+import org.apache.activemq.util.ThreadTracker;
+import org.apache.activemq.util.Wait;
+import org.apache.log4j.Logger;
+import org.springframework.core.io.ClassPathResource;
+
+import javax.jms.*;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
+
+import static org.apache.activemq.advisory.AdvisoryBroker.adds;
+import static org.apache.activemq.advisory.AdvisoryBroker.removes;
+
+public class ClientRebalanceAdvisoryTest extends JmsMultipleBrokersTestSupport {
+    private static final Logger LOG = Logger.getLogger(ClientRebalanceAdvisoryTest.class);
+    private static final String QUEUE_NAME = "Test";
+
+
+    public void testNoNetworkConsumersDangling() throws Exception {
+        createBroker(new ClassPathResource("org/apache/activemq/usecases/rebalance-broker1.xml"));
+        createBroker(new ClassPathResource("org/apache/activemq/usecases/rebalance-broker2.xml"));
+        createBroker(new ClassPathResource("org/apache/activemq/usecases/rebalance-broker3.xml"));
+
+        startAllBrokers();
+
+        brokers.get("b1").broker.waitUntilStarted();
+
+        TimeUnit.SECONDS.sleep(5);
+
+        LOG.info("Starting connection");
+
+        final ActiveMQConnectionFactory factory = new ActiveMQConnectionFactory("failover:(tcp://localhost:61616)?randomize=false&nested.closeAsync=false");
+        factory.setWatchTopicAdvisories(false);
+
+        /*
+        for (int j=0; j<3; j++) {
+            Connection conn = factory.createConnection();
+            conn.start();
+            Session session = conn.createSession(false, Session.AUTO_ACKNOWLEDGE);
+            Queue theQueue = session.createQueue(QUEUE_NAME);
+            MessageProducer producer = session.createProducer(theQueue);
+            for (int i = 0; i < numMessages / 3; i++) {
+                Message message = session.createTextMessage("Test message: " + i);
+                producer.send(message);
+            }
+            conn.close();
+        }
+*/
+
+        /*
+        final Connection conn = factory.createConnection();
+        conn.start();
+        Session session = conn.createSession(false, Session.AUTO_ACKNOWLEDGE);
+        Topic adv = session.createTopic(AdvisorySupport.CONSUMER_ADVISORY_TOPIC_PREFIX +">");
+        MessageConsumer consumer  = session.createConsumer(adv);
+        final AtomicInteger adds = new AtomicInteger();
+        final AtomicInteger removes = new AtomicInteger();
+
+        final HashSet<ConsumerId> trackedConsumers = new HashSet<ConsumerId>();
+        consumer.setMessageListener(new MessageListener() {
+            @Override
+            public void onMessage(Message message) {
+                switch (((ActiveMQMessage) message).getDataStructure().getDataStructureType()) {
+                    case ConsumerInfo.DATA_STRUCTURE_TYPE: {
+                        ConsumerInfo consumerInfo = (ConsumerInfo) (((ActiveMQMessage) message).getDataStructure());
+                        if (trackedConsumers.add(consumerInfo.getConsumerId())) {
+                            adds.incrementAndGet();
+                        }
+                        break;
+                    }
+                    case RemoveInfo.DATA_STRUCTURE_TYPE: {
+                        RemoveInfo removeInfo = (RemoveInfo) ((ActiveMQMessage) message).getDataStructure();
+                        if (trackedConsumers.remove((ConsumerId)removeInfo.getObjectId())) {
+                            removes.incrementAndGet();
+                        }
+                        break;
+                    }
+                }
+            }
+        });
+
+*/
+
+        final int numQueues = 100;
+        ExecutorService executorService = Executors.newFixedThreadPool(450);
+        LOG.info("Consume....");
+        for (int i = 0; i < 450; i++) {
+            final int count = i;
+            executorService.submit(new Runnable() {
+                @Override
+                public void run() {
+                    try {
+                        Connection conn = factory.createConnection();
+                        try {
+
+                            conn.start();
+                            Session session = conn.createSession(false, Session.AUTO_ACKNOWLEDGE);
+                            Queue theQueue = session.createQueue(QUEUE_NAME + ((count % numQueues)));
+// 100
+                            // 400 errors out
+                            for (int j = 0; j < 10; j++) {
+                                MessageConsumer consumer = session.createConsumer(theQueue);
+                            }
+                            //Message msg = consumer.receive(500);
+                            TimeUnit.MILLISECONDS.sleep(5000);
+
+                        } catch (Exception e) {
+                            LOG.error(e);
+                            e.printStackTrace();
+                        } finally {
+                            conn.close();
+                        }
+                    } catch (JMSException e) {
+                        e.printStackTrace();
+                    }
+                }
+            });
+        }
+
+        executorService.shutdown();
+        executorService.awaitTermination(10, TimeUnit.MINUTES);
+
+
+
+        // wait for connections to close
+        for (String name : brokers.keySet()) {
+            final RegionBroker b =
+                    (RegionBroker) brokers.get(name).broker.getRegionBroker();
+            LOG.info("Broker: " + name);
+            LOG.info(" connections: " + b.getConnectionStates().size());
+            assertTrue("all closed", Wait.waitFor(new Wait.Condition() {
+                @Override
+                public boolean isSatisified() throws Exception {
+                    LOG.info(b.getBrokerName() + " consumer count: " + b.getDestinationStatistics().getConsumers().getCount());
+                    LOG.info(b.getBrokerName() + " connections: " + b.getConnectionStates().size());
+               /*   try {
+                      for (ConnectionId id : b.getConnectionStates().keySet()) {
+                          LOG.info("Connection id: " + id);
+                      }
+                  } catch (ConcurrentModificationException expected) {} */
+                    // just the network connections
+                    return b.getConnectionStates().size() == 4l;
+                }
+            }, 10*60*1000));
+        }
+
+
+        TimeUnit.SECONDS.sleep(5);
+
+        long end  = System.currentTimeMillis() + TimeUnit.MINUTES.toMillis(3);
+        while (end > System.currentTimeMillis()) {
+            TimeUnit.SECONDS.sleep(20);
+            for (String name : brokers.keySet()) {
+                RegionBroker b =
+                        (RegionBroker) brokers.get(name).broker.getRegionBroker();
+                LOG.info("Broker: " + name);
+                LOG.info("Consumer count: " + b.getDestinationStatistics().getConsumers().getCount());
+                LOG.info(" connections: " + b.getConnectionStates().size());
+
+                //   LOG.info("Consumer tracker adds " + adds.get());
+                //   LOG.info("Consumer tracker removes " + removes.get());
+
+                for (Destination d : b.getQueueRegion().getDestinationMap().values()) {
+                    for (Subscription s : d.getConsumers()) {
+                        LOG.info(d.getName() + " q consumer:" + s);
+                    }
+                }
+            }
+        }
+
+
+        if (false) {
+            Wait.waitFor(new Wait.Condition() {
+                @Override
+                public boolean isSatisified() throws Exception {
+                    LOG.info("Adds: " + adds.get());
+                    LOG.info("Removes: " + removes.get());
+
+                    return (Long.compare(adds.get(), removes.get()) == 0);
+                }
+            }, 10 * 60 * 1000);
+        }
+
+        assertEquals("all advisories generated", adds.get(), removes.get() );
+
+        LOG.error("Result");
+        ThreadTracker.result();
+
+        try {
+            for (String name : brokers.keySet()) {
+                RegionBroker b =
+                        (RegionBroker) brokers.get(name).broker.getRegionBroker();
+                long consumersCount =  b.getDestinationStatistics().getConsumers().getCount();
+                if (consumersCount != numQueues) {
+                    dumpAllThreads("WRONG");
+                }
+               assertEquals("just network advisory consumers for " + name, 2*numQueues, consumersCount);
+            }
+        } finally {
+            destroyAllBrokers();
+        }
+    }
+}
\ No newline at end of file
diff --git a/activemq-unit-tests/src/test/resources/log4j.properties b/activemq-unit-tests/src/test/resources/log4j.properties
index 42d8c80d44..6060bd08f9 100755
--- a/activemq-unit-tests/src/test/resources/log4j.properties
+++ b/activemq-unit-tests/src/test/resources/log4j.properties
@@ -25,6 +25,7 @@ log4j.rootLogger=INFO, out, stdout
 #log4j.logger.org.apache.activemq.network.DemandForwardingBridgeSupport=DEBUG
 #log4j.logger.org.apache.activemq.transport.failover=TRACE
 #log4j.logger.org.apache.activemq.transport.TransportLogger.Connection=TRACE
+#log4j.logger.org.apache.activemq.broker.TransportConnection=TRACE
 #log4j.logger.org.apache.activemq.store.jdbc=TRACE
 #log4j.logger.org.apache.activemq.store.kahadb=TRACE
 #log4j.logger.org.apache.activemq.broker.region.cursors.AbstractStoreCursor=DEBUG
diff --git a/activemq-unit-tests/src/test/resources/org/apache/activemq/usecases/rebalance-broker1.xml b/activemq-unit-tests/src/test/resources/org/apache/activemq/usecases/rebalance-broker1.xml
index 0ed33d7bdd..8f60ae4507 100644
--- a/activemq-unit-tests/src/test/resources/org/apache/activemq/usecases/rebalance-broker1.xml
+++ b/activemq-unit-tests/src/test/resources/org/apache/activemq/usecases/rebalance-broker1.xml
@@ -18,8 +18,8 @@
 <beans xsi:schemaLocation=" http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd http://activemq.apache.org/schema/core http://activemq.apache.org/schema/core/activemq-core-5.5.0.xsd" xmlns="http://www.springframework.org/schema/beans" xmlns:amq="http://activemq.apache.org/schema/core" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance">
     <broker brokerName="b1" persistent="false" useJmx="false" xmlns="http://activemq.apache.org/schema/core">
         <networkConnectors>
-            <networkConnector name="b1_b2" uri="static:(tcp://localhost:61617)?useExponentialBackOff=false" duplex="false"/>
-            <networkConnector name="b1_b3" uri="static:(tcp://localhost:61618)?useExponentialBackOff=false"></networkConnector>
+            <networkConnector name="b1_b2" uri="static:(tcp://localhost:61617)?useExponentialBackOff=false" duplex="false" prefetchSize="100" suppressDuplicateQueueSubscriptions="true"/>
+            <networkConnector name="b1_b3" uri="static:(tcp://localhost:61618)?useExponentialBackOff=false" prefetchSize="100" suppressDuplicateQueueSubscriptions="true"></networkConnector>
         </networkConnectors>
 
         <!-- The transport connectors ActiveMQ will listen to -->
diff --git a/activemq-unit-tests/src/test/resources/org/apache/activemq/usecases/rebalance-broker2.xml b/activemq-unit-tests/src/test/resources/org/apache/activemq/usecases/rebalance-broker2.xml
index fd96b883bf..0dbca9c120 100644
--- a/activemq-unit-tests/src/test/resources/org/apache/activemq/usecases/rebalance-broker2.xml
+++ b/activemq-unit-tests/src/test/resources/org/apache/activemq/usecases/rebalance-broker2.xml
@@ -19,8 +19,8 @@
         <broker brokerName="b2" persistent="false" useJmx="false" xmlns="http://activemq.apache.org/schema/core">
 
         <networkConnectors>
-            <networkConnector name="b2_b1" uri="static:(tcp://localhost:61616)?useExponentialBackOff=false" duplex="false"/>
-            <networkConnector name="b2_b3" uri="static:(tcp://localhost:61618)?useExponentialBackOff=false"></networkConnector>
+            <networkConnector name="b2_b1" uri="static:(tcp://localhost:61616)?useExponentialBackOff=false" duplex="false" prefetchSize="100" suppressDuplicateQueueSubscriptions="true"/>
+            <networkConnector name="b2_b3" uri="static:(tcp://localhost:61618)?useExponentialBackOff=false" prefetchSize="100" suppressDuplicateQueueSubscriptions="true"></networkConnector>
         </networkConnectors>
 
         <!-- The transport connectors ActiveMQ will listen to -->
diff --git a/activemq-unit-tests/src/test/resources/org/apache/activemq/usecases/rebalance-broker3.xml b/activemq-unit-tests/src/test/resources/org/apache/activemq/usecases/rebalance-broker3.xml
index 3438feec17..23cea746ec 100644
--- a/activemq-unit-tests/src/test/resources/org/apache/activemq/usecases/rebalance-broker3.xml
+++ b/activemq-unit-tests/src/test/resources/org/apache/activemq/usecases/rebalance-broker3.xml
@@ -18,8 +18,8 @@
 <beans xsi:schemaLocation=" http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd http://activemq.apache.org/schema/core http://activemq.apache.org/schema/core/activemq-core-5.5.0.xsd" xmlns="http://www.springframework.org/schema/beans" xmlns:amq="http://activemq.apache.org/schema/core" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance">
     <broker brokerName="b3" persistent="false" useJmx="true" xmlns="http://activemq.apache.org/schema/core">
         <networkConnectors>
-            <networkConnector name="b3_b1" uri="static:(tcp://localhost:61616)?useExponentialBackOff=false"></networkConnector>
-            <networkConnector name="b3_b2" uri="static:(tcp://localhost:61617)?useExponentialBackOff=false"></networkConnector>
+            <networkConnector name="b3_b1" uri="static:(tcp://localhost:61616)?useExponentialBackOff=false" prefetchSize="100" suppressDuplicateQueueSubscriptions="true"></networkConnector>
+            <networkConnector name="b3_b2" uri="static:(tcp://localhost:61617)?useExponentialBackOff=false" prefetchSize="100" suppressDuplicateQueueSubscriptions="true"></networkConnector>
         </networkConnectors>
 
         <!-- The transport connectors ActiveMQ will listen to -->
