fix AMQ-1970 - pagedInMessages in slave was being filled due to more than 200(pageSize) unacked messages and slave not modifying the inflight count which is used in the pageIn logic

git-svn-id: https://svn.apache.org/repos/asf/activemq/trunk@704142 13f79535-47bb-0310-9956-ffa450edef68

activemq-core/src/main/java/org/apache/activemq/broker/region/PrefetchSubscription.java

@@ -163,6 +163,7 @@ public abstract class PrefetchSubscription extends AbstractSubscription {
                             pending.remove();
                             createMessageDispatch(node, node.getMessage());
                             dispatched.add(node);
+                            onDispatch(node, node.getMessage());
                         }
                         return;
                     }
@@ -173,7 +174,7 @@ public abstract class PrefetchSubscription extends AbstractSubscription {
         }
         throw new JMSException(
                 "Slave broker out of sync with master: Dispatched message ("
-                        + mdn.getMessageId() + ") was not in the pending list");
+                        + mdn.getMessageId() + ") was not in the pending list for " + mdn.getDestination().getPhysicalName());
     }
 
     public final void acknowledge(final ConnectionContext context,final MessageAck ack) throws Exception {
@@ -205,9 +206,7 @@ public abstract class PrefetchSubscription extends AbstractSubscription {
                             if (!this.getConsumerInfo().isBrowser()) {
                                 node.getRegionDestination().getDestinationStatistics().getDequeues().increment();
                             }
-                            if (!isSlave()) {
+                            node.getRegionDestination().getDestinationStatistics().getInflight().decrement();
-                                node.getRegionDestination().getDestinationStatistics().getInflight().decrement();
-                            }
                         } else {
                             // setup a Synchronization to remove nodes from the
                             // dispatched list.

activemq-core/src/test/java/org/apache/activemq/advisory/MasterSlaveTempQueueMemoryTest.java

@@ -16,14 +16,26 @@
  */
 package org.apache.activemq.advisory;
 
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+
+import javax.jms.Message;
+import javax.jms.MessageConsumer;
+import javax.jms.MessageListener;
+import javax.jms.MessageProducer;
+import javax.jms.Session;
+
+import org.apache.activemq.ActiveMQSession;
 import org.apache.activemq.broker.BrokerService;
+import org.apache.activemq.broker.region.Queue;
 import org.apache.activemq.broker.region.RegionBroker;
 
+
 public class MasterSlaveTempQueueMemoryTest extends TempQueueMemoryTest {
     String masterBindAddress = "tcp://localhost:61616";
     String slaveBindAddress = "tcp://localhost:62616";
     BrokerService slave;
-    
+
     /*
      * add a slave broker
      * @see org.apache.activemq.EmbeddedBrokerTestSupport#createBroker()
@@ -93,4 +105,46 @@ public class MasterSlaveTempQueueMemoryTest extends TempQueueMemoryTest {
                 masterRb.getDestinationStatistics().getDispatched().getCount());
     }
     
+    public void testMoreThanPageSizeUnacked() throws Exception {
+        
+        final int messageCount = Queue.MAX_PAGE_SIZE + 10;
+        final CountDownLatch latch = new CountDownLatch(1);
+        
+        serverSession = serverConnection.createSession(false, Session.AUTO_ACKNOWLEDGE);
+        ActiveMQSession s = (ActiveMQSession) serverSession;
+        s.setSessionAsyncDispatch(true);
+        
+        MessageConsumer serverConsumer = serverSession.createConsumer(serverDestination);
+        serverConsumer.setMessageListener(new MessageListener() {
+           
+            public void onMessage(Message msg) {
+                try {
+                    latch.await(30L, TimeUnit.SECONDS);
+                } catch (Exception e) {
+                    // TODO Auto-generated catch block
+                    e.printStackTrace();
+                }
+            }
+        });
+            
+        MessageProducer producer = clientSession.createProducer(serverDestination);
+        for (int i =0; i< messageCount; i++) {
+            Message msg = clientSession.createMessage();
+            producer.send(msg);
+        }
+        
+        RegionBroker slaveRb = (RegionBroker) slave.getBroker().getAdaptor(
+                RegionBroker.class);
+        RegionBroker masterRb = (RegionBroker) broker.getBroker().getAdaptor(
+                RegionBroker.class);
+        
+        Thread.sleep(4000);
+        assertEquals("inflight match expected", messageCount, masterRb.getDestinationStatistics().getInflight().getCount());        
+        assertEquals("inflight match on slave and master", slaveRb.getDestinationStatistics().getInflight().getCount(), masterRb.getDestinationStatistics().getInflight().getCount());
+        
+        latch.countDown();
+        Thread.sleep(4000);
+        assertEquals("inflight match expected", 0, masterRb.getDestinationStatistics().getInflight().getCount());        
+        assertEquals("inflight match on slave and master", slaveRb.getDestinationStatistics().getInflight().getCount(), masterRb.getDestinationStatistics().getInflight().getCount());
+    }
 }

activemq-core/src/test/java/org/apache/activemq/advisory/TempQueueMemoryTest.java

@@ -32,11 +32,11 @@ import org.apache.activemq.command.ActiveMQQueue;
  * @version $Revision: 397249 $
  */
 public class TempQueueMemoryTest extends EmbeddedBrokerTestSupport {
-    private Connection serverConnection;
+    protected Connection serverConnection;
-    private Session serverSession;
+    protected Session serverSession;
-    private Connection clientConnection;
+    protected Connection clientConnection;
-    private Session clientSession;
+    protected Session clientSession;
-    private Destination serverDestination;
+    protected Destination serverDestination;
     protected static final int COUNT = 2000;
 
     public void testLoadRequestReply() throws Exception {