ARTEMIS-5376 Preventing OME from QueueImpl::iteration while paging

Author: clebertsuconic

artemis-server/src/main/java/org/apache/activemq/artemis/core/server/ActiveMQServerLogger.java

@@ -1526,4 +1526,7 @@ void slowConsumerDetected(String sessionID,
 
    @LogMessage(id = 224157, value = "At least one of the components failed to start under the lockCoordinator {}. A retry will be executed", level = LogMessage.Level.INFO)
    void retryLockCoordinator(String name);
+
+   @LogMessage(id = 224158, value = "The operation {} on queue {} cannot read more data from paging into memory. The operation will be interrupted.", level = LogMessage.Level.INFO)
+   void preventQueueManagementToFloodMemory(String operation, String queue);
 }

artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/QueueImpl.java

@@ -2025,7 +2025,7 @@ public int deleteMatchingReferences(Filter filter) throws Exception {
 
    @Override
    public int deleteMatchingReferences(final int flushLimit, final Filter filter1, AckReason ackReason) throws Exception {
-      return iterQueue(flushLimit, filter1, createDeleteMatchingAction(ackReason));
+      return iterQueue("deleteMatchingReferences", flushLimit, filter1, createDeleteMatchingAction(ackReason), false);
    }
 
    QueueIterateAction createDeleteMatchingAction(AckReason ackReason) {
@@ -2043,9 +2043,11 @@ public boolean actMessage(Transaction tx, MessageReference ref) throws Exception
     * This is a generic method for any method interacting on the Queue to move or delete messages Instead of duplicate
     * the feature we created an abstract class where you pass the logic for each message.
     */
-   private int iterQueue(final int flushLimit,
+   private int iterQueue(final String operationName,
+                         final int flushLimit,
                          final Filter filter1,
-                         QueueIterateAction messageAction) throws Exception {
+                         QueueIterateAction messageAction,
+                         boolean allowDepaging) throws Exception {
       int count = 0;
       int txCount = 0;
 
@@ -2112,26 +2114,59 @@ private int iterQueue(final int flushLimit,
          }
 
          if (pageIterator != null) {
-            while (pageIterator.hasNext() && !messageAction.expectedHitsReached(count)) {
-               PagedReference reference = pageIterator.next();
-               pageIterator.remove();
+            PageIterator theIterator;
 
-               if (messageAction.match(reference)) {
-                  if (!messageAction.actMessage(tx, reference)) {
-                     addTail(reference, false);
+            if (allowDepaging) {
+               theIterator = pageIterator;
+            } else {
+               // if not allowed to depage, it will open a new iterator
+               theIterator = pageSubscription.iterator();
+            }
+
+            try {
+               while (theIterator.hasNext() && !messageAction.expectedHitsReached(count)) {
+                  PagedReference reference = theIterator.next();
+
+                  if (messageAction.match(reference)) {
+                     if (messageAction.actMessage(tx, reference)) {
+                        theIterator.remove();
+                     } else {
+                        if (allowDepaging) {
+                           theIterator.remove();
+                           addTail(reference, false);
+                           if (!needsDepage()) {
+                              ActiveMQServerLogger.LOGGER.preventQueueManagementToFloodMemory(operationName, String.valueOf(QueueImpl.this.getName()));
+                              break;
+                           }
+                        }
+                     }
+                     txCount++;
+                     count++;
+                  } else {
+                     if (allowDepaging) {
+                        theIterator.remove();
+                        addTail(reference, false);
+                        if (!needsDepage()) {
+                           ActiveMQServerLogger.LOGGER.preventQueueManagementToFloodMemory(operationName, String.valueOf(QueueImpl.this.getName()));
+                           break;
+                        }
+                     }
+                  }
+
+                  if (txCount > 0 && txCount % flushLimit == 0) {
+                     tx.commit();
+                     tx = new TransactionImpl(storageManager);
+                     txCount = 0;
                   }
-                  txCount++;
-                  count++;
-               } else {
-                  addTail(reference, false);
                }
 
-               if (txCount > 0 && txCount % flushLimit == 0) {
-                  tx.commit();
-                  tx = new TransactionImpl(storageManager);
-                  txCount = 0;
+            } finally {
+               if (!allowDepaging) {
+                  // close the iterator if not allowed to depage
+                  theIterator.close();
                }
             }
+
          }
 
          if (txCount > 0) {
@@ -2441,7 +2476,7 @@ public void run() {
    @Override
    public boolean sendMessageToDeadLetterAddress(final long messageID) throws Exception {
 
-      return iterQueue(DEFAULT_FLUSH_LIMIT, null, new QueueIterateAction(messageID) {
+      return iterQueue("sendMessageToDeadLetterAddress", DEFAULT_FLUSH_LIMIT, null, new QueueIterateAction(messageID) {
 
          @Override
          public boolean actMessage(Transaction tx, MessageReference ref) throws Exception {
@@ -2450,14 +2485,14 @@ public boolean actMessage(Transaction tx, MessageReference ref) throws Exception
             return true;
          }
 
-      }) == 1;
+      }, false) == 1;
 
    }
 
    @Override
    public int sendMessagesToDeadLetterAddress(Filter filter) throws Exception {
 
-      return iterQueue(DEFAULT_FLUSH_LIMIT, filter, new QueueIterateAction() {
+      return iterQueue("sendMessagesToDeadLetterAddress", DEFAULT_FLUSH_LIMIT, filter, new QueueIterateAction() {
 
          @Override
          public boolean actMessage(Transaction tx, MessageReference ref) throws Exception {
@@ -2466,7 +2501,7 @@ public boolean actMessage(Transaction tx, MessageReference ref) throws Exception
             return true;
          }
 
-      });
+      }, false);
 
    }
 
@@ -2476,7 +2511,7 @@ public boolean moveReference(final long messageID,
                                 final Binding binding,
                                 final boolean rejectDuplicate) throws Exception {
 
-      return iterQueue(DEFAULT_FLUSH_LIMIT, null, new QueueIterateAction(messageID) {
+      return iterQueue("moveReference", DEFAULT_FLUSH_LIMIT, null, new QueueIterateAction(messageID) {
 
          @Override
          public boolean actMessage(Transaction tx, MessageReference ref) throws Exception {
@@ -2485,7 +2520,7 @@ public boolean actMessage(Transaction tx, MessageReference ref) throws Exception
             return true;
          }
 
-      }) == 1;
+      }, false) == 1;
 
    }
 
@@ -2513,7 +2548,7 @@ public int moveReferences(final int flushLimit,
       final Integer expectedHits = messageCount > 0 ? messageCount : null;
       final DuplicateIDCache targetDuplicateCache = postOffice.getDuplicateIDCache(toAddress);
 
-      return iterQueue(flushLimit, filter, new QueueIterateAction(expectedHits) {
+      return iterQueue("moveReferences", flushLimit, filter, new QueueIterateAction(expectedHits) {
          @Override
          public boolean actMessage(Transaction tx, MessageReference ref) throws Exception {
             boolean ignored = false;
@@ -2537,37 +2572,37 @@ public boolean actMessage(Transaction tx, MessageReference ref) throws Exception
 
             return true;
          }
-      });
+      }, false);
    }
 
    public int moveReferencesBetweenSnFQueues(final SimpleString queueSuffix) throws Exception {
-      return iterQueue(DEFAULT_FLUSH_LIMIT, null, new QueueIterateAction() {
+      return iterQueue("moveReferencesBetweenSnFQueues", DEFAULT_FLUSH_LIMIT, null, new QueueIterateAction() {
          @Override
          public boolean actMessage(Transaction tx, MessageReference ref) throws Exception {
             return moveBetweenSnFQueues(queueSuffix, tx, ref, null);
          }
-      });
+      }, false);
    }
 
    @Override
    public boolean copyReference(final long messageID,
                                 final SimpleString toQueue,
                                 final Binding binding) throws Exception {
 
-      return iterQueue(DEFAULT_FLUSH_LIMIT, null, new QueueIterateAction(messageID) {
+      return iterQueue("copyReference", DEFAULT_FLUSH_LIMIT, null, new QueueIterateAction(messageID) {
 
          @Override
          public boolean actMessage(Transaction tx, MessageReference ref) throws Exception {
             copy(tx, toQueue, binding, ref);
             return false;
          }
 
-      }) == 1;
+      }, false) == 1;
 
    }
 
    public int rerouteMessages(final SimpleString queueName, final Filter filter) throws Exception {
-      return iterQueue(DEFAULT_FLUSH_LIMIT, filter, new QueueIterateAction() {
+      return iterQueue("rerouteMessages", DEFAULT_FLUSH_LIMIT, filter, new QueueIterateAction() {
          @Override
          public boolean actMessage(Transaction tx, MessageReference ref) throws Exception {
             RoutingContext routingContext = new RoutingContextImpl(tx);
@@ -2579,7 +2614,7 @@ public boolean actMessage(Transaction tx, MessageReference ref) throws Exception
             postOffice.processRoute(m, routingContext, false);
             return false;
          }
-      });
+      }, false);
    }
 
    @Override
@@ -2592,7 +2627,7 @@ public int retryMessages(Filter filter, Integer expectedHits) throws Exception {
 
       final HashMap<String, Long> queues = new HashMap<>();
 
-      return iterQueue(DEFAULT_FLUSH_LIMIT, filter, new QueueIterateAction(expectedHits) {
+      return iterQueue("retryMessages", DEFAULT_FLUSH_LIMIT, filter, new QueueIterateAction(expectedHits) {
 
          @Override
          public boolean actMessage(Transaction tx, MessageReference ref) throws Exception {
@@ -2630,37 +2665,37 @@ public boolean actMessage(Transaction tx, MessageReference ref) throws Exception
             }
             return false;
          }
-      });
+      }, false);
 
    }
 
    @Override
    public boolean changeReferencePriority(final long messageID, final byte newPriority) throws Exception {
 
-      return iterQueue(DEFAULT_FLUSH_LIMIT, null, new QueueIterateAction(messageID) {
+      return iterQueue("changeReferencePriority", DEFAULT_FLUSH_LIMIT, null, new QueueIterateAction(messageID) {
 
          @Override
          public boolean actMessage(Transaction tx, MessageReference ref) throws Exception {
             ref.getMessage().setPriority(newPriority);
             return false;
          }
 
-      }) == 1;
+      }, true) == 1;
 
    }
 
    @Override
    public int changeReferencesPriority(final Filter filter, final byte newPriority) throws Exception {
 
-      return iterQueue(DEFAULT_FLUSH_LIMIT, filter, new QueueIterateAction() {
+      return iterQueue("changeReferencesPriority", DEFAULT_FLUSH_LIMIT, filter, new QueueIterateAction() {
 
          @Override
          public boolean actMessage(Transaction tx, MessageReference ref) throws Exception {
             ref.getMessage().setPriority(newPriority);
             return false;
          }
 
-      });
+      }, true);
 
    }