Merge 5.19.1

activemq-all/pom.xml

@@ -14,7 +14,7 @@
   <parent>
     <groupId>org.apache.activemq</groupId>
     <artifactId>activemq-parent</artifactId>
-    <version>5.19.1-TT.5-SNAPSHOT</version>
+    <version>5.19.2-TT.1-SNAPSHOT</version>
   </parent>
 
   <artifactId>activemq-all</artifactId>

activemq-amqp/pom.xml

@@ -22,7 +22,7 @@
   <parent>
     <groupId>org.apache.activemq</groupId>
     <artifactId>activemq-parent</artifactId>
-    <version>5.19.1-TT.5-SNAPSHOT</version>
+    <version>5.19.2-TT.1-SNAPSHOT</version>
   </parent>
 
   <artifactId>activemq-amqp</artifactId>

activemq-blueprint/pom.xml

@@ -22,7 +22,7 @@
   <parent>
     <groupId>org.apache.activemq</groupId>
     <artifactId>activemq-parent</artifactId>
-    <version>5.19.1-TT.5-SNAPSHOT</version>
+    <version>5.19.2-TT.1-SNAPSHOT</version>
   </parent>
 
   <artifactId>activemq-blueprint</artifactId>

activemq-broker/pom.xml

@@ -22,7 +22,7 @@
   <parent>
     <groupId>org.apache.activemq</groupId>
     <artifactId>activemq-parent</artifactId>
-    <version>5.19.1-TT.5-SNAPSHOT</version>
+    <version>5.19.2-TT.1-SNAPSHOT</version>
   </parent>
 
   <artifactId>activemq-broker</artifactId>

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

@@ -298,6 +298,16 @@ public void removePending(MessageReference node) throws IOException {
         pending.remove(node);
     }
 
+    @Override
+    protected void processExpiredAck(ConnectionContext context, Destination dest,
+        MessageReference node) {
+
+        // Each subscription needs to expire both on the store and
+        // decrement the reference count
+        super.processExpiredAck(context, dest, node);
+        node.decrementReferenceCount();
+    }
+
     @Override
     protected void doAddRecoveredMessage(MessageReference message) throws Exception {
         synchronized (pending) {

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

@@ -298,9 +298,8 @@ public final void acknowledge(final ConnectionContext context,final MessageAck a
                         inAckRange = true;
                     }
                     if (inAckRange) {
-                        Destination regionDestination = nodeDest;
                         if (broker.isExpired(node)) {
-                            regionDestination.messageExpired(context, this, node);
+                            processExpiredAck(context, nodeDest, node);
                         }
                         iter.remove();
                         decrementPrefetchCounter(node);
@@ -396,6 +395,11 @@ public final void acknowledge(final ConnectionContext context,final MessageAck a
         }
     }
 
+    protected void processExpiredAck(final ConnectionContext context, final Destination dest,
+        final MessageReference node) {
+        dest.messageExpired(context, this, node);
+    }
+
     private void registerRemoveSync(ConnectionContext context, final MessageReference node) {
         // setup a Synchronization to remove nodes from the
         // dispatched list.

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

@@ -17,10 +17,12 @@
 package org.apache.activemq.broker.region;
 
 import java.io.IOException;
+import java.io.UncheckedIOException;
 import java.util.ArrayList;
 import java.util.LinkedList;
 import java.util.List;
 import java.util.Map;
+import java.util.Map.Entry;
 import java.util.concurrent.CancellationException;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ConcurrentMap;
@@ -29,6 +31,7 @@
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.locks.ReentrantReadWriteLock;
 
+import java.util.stream.Collectors;
 import org.apache.activemq.advisory.AdvisorySupport;
 import org.apache.activemq.broker.BrokerService;
 import org.apache.activemq.broker.ConnectionContext;
@@ -51,8 +54,8 @@
 import org.apache.activemq.command.SubscriptionInfo;
 import org.apache.activemq.filter.MessageEvaluationContext;
 import org.apache.activemq.filter.NonCachedMessageEvaluationContext;
-import org.apache.activemq.management.MessageFlowStats;
 import org.apache.activemq.store.MessageRecoveryListener;
+import org.apache.activemq.store.MessageStore.StoreType;
 import org.apache.activemq.store.NoLocalSubscriptionAware;
 import org.apache.activemq.store.PersistenceAdapter;
 import org.apache.activemq.store.TopicMessageStore;
@@ -702,6 +705,11 @@ public boolean isDuplicate(MessageId id) {
                     for (DurableTopicSubscription sub : durableSubscribers.values()) {
                         if (!sub.isActive() || sub.isEnableMessageExpirationOnActiveDurableSubs()) {
                             message.setRegionDestination(this);
+                            // AMQ-9721 - Remove message from the cursor if it exists after
+                            // loading from the store.  Store recoverExpired() does not inc
+                            // the ref count so we don't need to decrement here, but if
+                            // the cursor finds its own copy in memory it will dec that ref.
+                            sub.removePending(message);
                             messageExpired(connectionContext, sub, message);
                         }
                     }
@@ -825,15 +833,99 @@ protected void dispatch(final ConnectionContext context, Message message) throws
         }
     }
 
-    private final AtomicBoolean expiryTaskInProgress = new AtomicBoolean(false);
-    private final Runnable expireMessagesWork = new Runnable() {
+
+    /**
+     * Simple recovery listener that will check if the topic memory usage is full
+     * when hasSpace() is called. This could be enhanced in the future if needed.
+     */
+    private final MessageRecoveryListener expiryListener =  new MessageRecoveryListener() {
+
         @Override
-        public void run() {
-            List<Message> browsedMessages = new InsertionCountList<Message>();
-            doBrowse(browsedMessages, getMaxExpirePageSize());
+        public boolean recoverMessage(Message message) {
+            return true;
+        }
+
+        @Override
+        public boolean recoverMessageReference(MessageId ref)  {
+            return true;
+        }
+
+        @Override
+        public boolean hasSpace() {
+            return !Topic.this.memoryUsage.isFull();
+        }
+
+        @Override
+        public boolean isDuplicate(MessageId ref) {
+            return false;
+        }
+    };
+
+    private final AtomicBoolean expiryTaskInProgress = new AtomicBoolean(false);
+    private final Runnable expireMessagesWork = () -> {
+        try {
+            final TopicMessageStore store = Topic.this.topicStore;
+            if (store != null && store.getType() == StoreType.KAHADB) {
+                if (store.getMessageCount() == 0) {
+                    LOG.debug("Skipping topic expiration check for {}, store size is 0", destination);
+                    return;
+                }
+
+                // get the sub keys that should be checked for expired messages
+                final var subs = durableSubscribers.entrySet().stream()
+                    .filter(entry -> isEligibleForExpiration(entry.getValue()))
+                    .map(Entry::getKey).collect(Collectors.toSet());
+
+                if (subs.isEmpty()) {
+                    LOG.debug("Skipping topic expiration check for {}, no eligible subscriptions to check", destination);
+                    return;
+                }
+
+                // For each eligible subscription, return the messages in the store that are expired
+                // The same message refs are shared between subs if duplicated so this is efficient
+                var expired = store.recoverExpired(subs, getMaxExpirePageSize(),
+                    expiryListener);
+
+                final ConnectionContext connectionContext = createConnectionContext();
+                // Go through any expired messages and remove for each sub
+                for (Entry<SubscriptionKey, List<Message>> entry : expired.entrySet()) {
+                    DurableTopicSubscription sub = durableSubscribers.get(entry.getKey());
+                    List<Message> expiredMessages = entry.getValue();
+
+                    // If the sub still exists and there are expired messages then process
+                    if (sub != null && !expiredMessages.isEmpty()) {
+                        // There's a small race condition here if the sub comes online,
+                        // but it's not a big deal as at worst there maybe be duplicate acks for
+                        // the expired message but the store can handle it
+                        if (isEligibleForExpiration(sub)) {
+                            expiredMessages.forEach(message -> {
+                                message.setRegionDestination(Topic.this);
+                                try {
+                                    // AMQ-9721 - Remove message from the cursor if it exists after
+                                    // loading from the store.  Store recoverExpired() does not inc
+                                    // the ref count so we don't need to decrement here, but if
+                                    // the cursor finds its own copy in memory it will dec that ref.
+                                    sub.removePending(message);
+                                } catch (IOException e) {
+                                    throw new UncheckedIOException(e);
+                                }
+                                messageExpired(connectionContext, sub, message);
+                            });
+                        }
+                    }
+                }
+            } else {
+                // If not KahaDB, fall back to the legacy browse method because
+                // the recoverExpired() method is not supported
+                doBrowse(new InsertionCountList<>(), getMaxExpirePageSize());
+            }
+        } catch (Throwable e) {
+            LOG.warn("Failed to expire messages on Topic: {}", getActiveMQDestination().getPhysicalName(), e);
+        } finally {
             expiryTaskInProgress.set(false);
         }
     };
+
     private final Runnable expireMessagesTask = new Runnable() {
         @Override
         public void run() {
@@ -855,9 +947,6 @@ public void messageExpired(ConnectionContext context, Subscription subs, Message
         ack.setDestination(destination);
         ack.setMessageID(reference.getMessageId());
         try {
-            if (subs instanceof DurableTopicSubscription) {
-                ((DurableTopicSubscription)subs).removePending(reference);
-            }
             acknowledge(context, subs, ack, reference);
         } catch (Exception e) {
             LOG.error("Failed to remove expired Message from the store ", e);
@@ -927,6 +1016,10 @@ private void clearPendingAndDispatch(DurableTopicSubscription durableTopicSubscr
         }
     }
 
+    private static boolean isEligibleForExpiration(DurableTopicSubscription sub) {
+        return sub.isEnableMessageExpirationOnActiveDurableSubs() || !sub.isActive();
+    }
+
     public Map<SubscriptionKey, DurableTopicSubscription> getDurableTopicSubs() {
         return durableSubscribers;
     }

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

@@ -188,6 +188,9 @@ public void add(MessageReference node) throws Exception {
                                 messagesToEvict = oldMessages.length;
                                 for (int i = 0; i < messagesToEvict; i++) {
                                     MessageReference oldMessage = oldMessages[i];
+                                    // AMQ-9721 - discard no longer removes from matched so remove here
+                                    oldMessage.decrementReferenceCount();
+                                    matched.remove(oldMessage);
                                     //Expired here is false as we are discarding due to the messageEvictingStrategy
                                     discard(oldMessage, false);
                                 }
@@ -751,8 +754,6 @@ public void onFailure() {
     private void discard(MessageReference message, boolean expired) {
         discarding = true;
         try {
-            message.decrementReferenceCount();
-            matched.remove(message);
             if (destination != null) {
                 destination.getDestinationStatistics().getDequeues().increment();
                 if(destination.isAdvancedNetworkStatisticsEnabled() && getContext() != null && getContext().isNetworkConnection()) {

activemq-broker/src/main/java/org/apache/activemq/broker/region/cursors/AbstractStoreCursor.java

@@ -114,7 +114,7 @@ public synchronized boolean recoverMessage(Message message, boolean cached) thro
                 }
             }
             message.incrementReferenceCount();
-            batchList.addMessageLast(message);
+            batchList.addMessageLast(createBatchListRef(message));
             clearIterator(true);
             recovered = true;
         } else if (!cached) {
@@ -136,6 +136,10 @@ public synchronized boolean recoverMessage(Message message, boolean cached) thro
         return recovered;
     }
 
+    protected MessageReference createBatchListRef(Message message) {
+        return message;
+    }
+
     protected boolean duplicateFromStoreExcepted(Message message) {
         // expected for messages pending acks with kahadb.concurrentStoreAndDispatchQueues=true for
         // which this existing unused flag has been repurposed
@@ -448,13 +452,15 @@ public final synchronized void remove() {
 
     @Override
     public final synchronized void remove(MessageReference node) {
-        if (batchList.remove(node) != null) {
+        final PendingNode message = batchList.remove(node);
+        if (message != null) {
             size--;
             setCacheEnabled(false);
+            // decrement reference count if removed from batchList
+            message.getMessage().decrementReferenceCount();
         }
     }
 
-
     @Override
     public final synchronized void clear() {
         gc();

activemq-broker/src/main/java/org/apache/activemq/broker/region/cursors/FilePendingMessageCursor.java

@@ -162,8 +162,7 @@ public synchronized void release() {
     @Override
     public synchronized void destroy() throws Exception {
         stop();
-        for (Iterator<MessageReference> i = memoryList.iterator(); i.hasNext();) {
-            MessageReference node = i.next();
+        for (MessageReference node : memoryList) {
             node.decrementReferenceCount();
         }
         memoryList.clear();
@@ -365,11 +364,19 @@ public synchronized long messageSize() {
      */
     @Override
     public synchronized void clear() {
+        // AMQ-9726 - Iterate over all nodes to decrement the ref count
+        // to decrement the memory usage tracker
+        for (MessageReference node : memoryList) {
+            node.decrementReferenceCount();
+        }
         memoryList.clear();
         if (!isDiskListEmpty()) {
             try {
-                getDiskList().destroy();
-            } catch (IOException e) {
+                // AMQ-9726 - This method will destroy the list and
+                // set the reference to null so it will be reset
+                // for future writes
+                destroyDiskList();
+            } catch (Exception e) {
                 throw new RuntimeException(e);
             }
         }

activemq-broker/src/main/java/org/apache/activemq/broker/region/cursors/StoreDurableSubscriberCursor.java

@@ -28,6 +28,7 @@
 import org.apache.activemq.broker.region.Destination;
 import org.apache.activemq.broker.region.DurableTopicSubscription;
 import org.apache.activemq.broker.region.MessageReference;
+import org.apache.activemq.broker.region.NullMessageReference;
 import org.apache.activemq.broker.region.Topic;
 import org.apache.activemq.command.Message;
 import org.apache.activemq.usage.SystemUsage;
@@ -274,8 +275,27 @@ public synchronized void remove() {
 
     @Override
     public synchronized void remove(MessageReference node) {
-        for (PendingMessageCursor tsp : storePrefetches) {
-            tsp.remove(node);
+        // AMQ-9721 - Check if message is persistent or non-persistent.
+        // Removing from the non-persistent cursor requires searching the
+        // entire list if it's paged onto disk which is quite slow,
+        // so it doesn't make sense to try and remove as it will never
+        // exist if it's persistent.
+
+        // MessageReference can be a null reference if called from DurableSubscriptionView
+        // so we do not know if it's persistent and just need to search everything.
+        if (node instanceof NullMessageReference) {
+            for (PendingMessageCursor tsp : storePrefetches) {
+                tsp.remove(node);
+            }
+        } else if (node.isPersistent()) {
+            for (PendingMessageCursor tsp : storePrefetches) {
+                if (tsp.equals(nonPersistent)) {
+                    continue;
+                }
+                tsp.remove(node);
+            }
+        } else {
+            nonPersistent.remove(node);
         }
     }