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

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 {} 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 number) throws Exception;
+
     /**
      * Copies a given message to another destination.
      * 

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

@@ -1313,9 +1313,15 @@ public QueueMessageReference getMessage(String id) {
     }
 
     public void purge() throws Exception {
+        purge(this.destinationStatistics.getMessages().getCount());
+    }
+
+    public void purge(long numberOfMessages) throws Exception {
         ConnectionContext c = createConnectionContext();
         List<MessageReference> list = null;
         sendLock.lock();
+
+        long purgeCount = 0L;
         try {
             long originalMessageCount = this.destinationStatistics.getMessages().getCount();
             do {
@@ -1327,20 +1333,28 @@ public void purge() throws Exception {
                     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());
+                LOG.warn("{} after purge of {} messages, message count stats report: {}", getActiveMQDestination().getQualifiedName(), numberOfMessages, this.destinationStatistics.getMessages().getCount());
             }
         } finally {
             sendLock.unlock();