[AMQ-9858] Support purging the first number of messages from a queue

Author: mattrpav

activemq-broker/src/main/java/org/apache/activemq/broker/jmx/QueueView.java

@@ -64,6 +64,14 @@ public synchronized void purge() throws Exception {
         LOG.info("{} purge of {} messages", destination.getActiveMQDestination().getQualifiedName(), originalMessageCount);
     }
 
+    public synchronized void purge(long numberOfMessages) throws Exception {
+        final long originalMessageCount = destination.getDestinationStatistics().getMessages().getCount();
+
+        ((Queue)destination).purge(numberOfMessages);
+
+        LOG.info("{} purge {} of {} messages", destination.getActiveMQDestination().getQualifiedName(), numberOfMessages, originalMessageCount);
+    }
+
     public synchronized boolean removeMessage(String messageId) throws Exception {
         return ((Queue)destination).removeMessage(messageId);
     }

activemq-broker/src/main/java/org/apache/activemq/broker/jmx/QueueViewMBean.java

@@ -70,6 +70,14 @@ public interface QueueViewMBean extends DestinationViewMBean {
     @MBeanInfo("Removes all of the messages in the queue.")
     void purge() throws Exception;
 
+    /**
+     * Removes the first number of messages in the queue.
+     *
+     * @throws Exception
+     */
+    @MBeanInfo("Removes the first number of messages in the queue.")
+    void purge(long numberOfMessages) throws Exception;
+
     /**
      * Copies a given message to another destination.
      * 

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

@@ -1320,34 +1320,53 @@ public QueueMessageReference getMessage(String id) {
     }
 
     public void purge() throws Exception {
+        purge(this.destinationStatistics.getMessages().getCount());
+    }
+
+    public void purge(long numberOfMessages) throws Exception {
+
+        if (numberOfMessages <= 0) {
+            return;
+        }
+
         ConnectionContext c = createConnectionContext();
         List<MessageReference> list = null;
         sendLock.lock();
+
+        long purgeCount = 0L;
         try {
             long originalMessageCount = this.destinationStatistics.getMessages().getCount();
             do {
                 doPageIn(true, false, getMaxPageSize());  // signal no expiry processing needed.
                 pagedInMessagesLock.readLock().lock();
                 try {
                     list = new ArrayList<MessageReference>(pagedInMessages.values());
-                }finally {
+                } finally {
                     pagedInMessagesLock.readLock().unlock();
                 }
 
-                for (MessageReference ref : list) {
+                int deleteCount = list.size();
+                if ((numberOfMessages - purgeCount) < list.size()) {
+                    deleteCount = (int)(numberOfMessages - purgeCount);
+                }
+
+                for (int n=0; n < deleteCount; n++) {
                     try {
-                        QueueMessageReference r = (QueueMessageReference) ref;
+                        QueueMessageReference r = (QueueMessageReference) list.get(n);
                         removeMessage(c, r);
                         messages.rollback(r.getMessageId());
+                        purgeCount++;
                     } catch (IOException e) {
                     }
                 }
                 // don't spin/hang if stats are out and there is nothing left in the
                 // store
-            } while (!list.isEmpty() && this.destinationStatistics.getMessages().getCount() > 0);
+            } while (!list.isEmpty() &&
+                    this.destinationStatistics.getMessages().getCount() > 0 &&
+                    purgeCount < numberOfMessages);
 
-            if (this.destinationStatistics.getMessages().getCount() > 0) {
-                LOG.warn("{} after purge of {} messages, message count stats report: {}", getActiveMQDestination().getQualifiedName(), originalMessageCount, this.destinationStatistics.getMessages().getCount());
+            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 {
             sendLock.unlock();

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()});
     }