Add queuePurge(count) test to PurgeTest

Author: mattrpav

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

@@ -1341,7 +1341,7 @@ public void purge(long numberOfMessages) throws Exception {
                 pagedInMessagesLock.readLock().lock();
                 try {
                     list = new ArrayList<MessageReference>(pagedInMessages.values());
-                }finally {
+                } finally {
                     pagedInMessagesLock.readLock().unlock();
                 }
 
@@ -1365,7 +1365,7 @@ public void purge(long numberOfMessages) throws Exception {
                     this.destinationStatistics.getMessages().getCount() > 0 &&
                     purgeCount < numberOfMessages);
 
-            if (this.destinationStatistics.getMessages().getCount() > 0) {
+            if (numberOfMessages == originalMessageCount && this.destinationStatistics.getMessages().getCount() > 0) {
                 LOG.warn("{} after purge {} of {} messages, message count stats report: {}", getActiveMQDestination().getQualifiedName(), numberOfMessages, originalMessageCount, this.destinationStatistics.getMessages().getCount());
             }
         } finally {

activemq-unit-tests/src/test/java/org/apache/activemq/broker/jmx/PurgeTest.java

@@ -112,6 +112,39 @@ public void testPurge() throws Exception {
         producer.close();
     }
 
+    public void testPurgeCount() throws Exception {
+        // Send some messages
+        int messagesSent = 1_000;
+        int messagesPurge = 200;
+
+        connection = connectionFactory.createConnection();
+        connection.setClientID(clientID);
+        connection.start();
+        Session session = connection.createSession(transacted, authMode);
+        destination = createDestination();
+        MessageProducer producer = session.createProducer(destination);
+        for (int i = 0; i < messagesSent; i++) {
+            Message message = session.createTextMessage("Message: " + i);
+            producer.send(message);
+        }
+
+        // Now get the QueueViewMBean and purge
+        String objectNameStr = broker.getBrokerObjectName().toString();
+        objectNameStr += ",destinationType=Queue,destinationName="+getDestinationString();
+        ObjectName queueViewMBeanName = assertRegisteredObjectName(objectNameStr);
+        QueueViewMBean proxy = (QueueViewMBean)MBeanServerInvocationHandler.newProxyInstance(mbeanServer, queueViewMBeanName, QueueViewMBean.class, true);
+
+        long count = proxy.getQueueSize();
+        assertEquals("Queue size", count, messagesSent);
+
+        for (int i = 1; i <= 5; i++) {
+            proxy.purge(messagesPurge);
+            count = proxy.getQueueSize();
+            assertEquals("Queue size", count, messagesSent - (messagesPurge * i));
+        }
+        producer.close();
+    }
+
     public void initCombosForTestDelete() {
         addCombinationValues("persistenceAdapter", new Object[] {new MemoryPersistenceAdapter(), new KahaDBPersistenceAdapter()});
     }