Update branch up to OSS 6.1.4, included post release commit and update poms to 6.1.5-TT.1-SNAPSHOT

activemq-all/pom.xml

@@ -14,7 +14,7 @@
   <parent>
     <groupId>org.apache.activemq</groupId>
     <artifactId>activemq-parent</artifactId>
-    <version>6.1.3-TT.4-SNAPSHOT</version>
+    <version>6.1.5-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>6.1.3-TT.4-SNAPSHOT</version>
+    <version>6.1.5-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>6.1.3-TT.4-SNAPSHOT</version>
+    <version>6.1.5-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>6.1.3-TT.4-SNAPSHOT</version>
+    <version>6.1.5-TT.1-SNAPSHOT</version>
   </parent>
 
   <artifactId>activemq-broker</artifactId>

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

@@ -255,12 +255,20 @@ public synchronized boolean tryAddMessageLast(MessageReference node, long wait)
             disableCache = true;
         }
 
-        if (disableCache && isCacheEnabled()) {
+        // AMQ-9625 - use this.cacheEnabled directly because the method isCacheEnabled() is overriden
+        // to try to re-enable the cache which we don't want at this point as we already skipped
+        // adding it to the cache
+        if (disableCache && this.cacheEnabled) {
             if (LOG.isTraceEnabled()) {
                 LOG.trace("{} - disabling cache on add {} {}", this, node.getMessageId(), node.getMessageId().getFutureOrSequenceLong());
             }
             syncWithStore(node.getMessage());
             setCacheEnabled(false);
+        } else if (!this.cacheEnabled) {
+            // AMQ-9625 - Verify and wait on previous in flight async messages here if another
+            // thread triggered the cache to be disabled
+            // see the waitForAsyncMessage() method and Jira for more info
+            waitForAsyncMessage(node.getMessage());
         }
         size++;
         return true;
@@ -319,6 +327,11 @@ private void syncWithStore(Message currentAdd) throws Exception {
             break;
         }
 
+        // AMQ-9625 - If we are disabling the cache and syncing the store then
+        // we need to wait for task to finish before updating the store batch
+        // see the waitForAsyncMessage() method and Jira for more info
+        waitForAsyncMessage(currentAdd);
+
         MessageId candidate = lastCachedIds[ASYNC_ADD];
         if (candidate != null) {
             // ensure we don't skip current possibly sync add b/c we waited on the future
@@ -530,4 +543,38 @@ public String toString() {
     public Subscription getSubscription() {
         return null;
     }
+
+    // AMQ-9625 - If the cache is disabled check if we need to wait for an async message
+    // to finish its task because the message is not being added to the cache.
+    // Normally, async messages will only be used if the cache is enabled so most of the time
+    // this check should not find any async messages to wait on if the cache is disabled
+    // and is basically a noop.
+    //
+    // However, while messages are being published, if the memory limit is reached the first
+    // thread that is adding the message that reaches the limit will disable the cache.
+    // This means there will be 1 or more potentially outstanding in flight adds that are
+    // queued up as async writes to the store.
+    //
+    // If the cache is disabled, we need to wait for any async message tasks to be
+    // finished otherwise there is a chance of missing the messages on dispatch
+    // when the queue pages in the next batch because store writes will finish after
+    // the store cursor has already moved ahead leading to a stuck message.
+    private void waitForAsyncMessage(Message node) {
+        // Note: isRecievedByDFBridge() was repurposed to be used to mark messages that
+        // are added to the store as async
+        if (node.getMessage().isRecievedByDFBridge()) {
+            final Object futureOrLong = node.getMessageId().getFutureOrSequenceLong();
+            if (futureOrLong instanceof Future) {
+                try {
+                    ((Future<?>) futureOrLong).get();
+                } catch (Exception exceptionOk) {
+                    // We don't care if we get an exception (cancelled, etc) we just want
+                    // to ensure the task is finished and not pending.
+                } finally {
+                    LOG.trace("{} - future finished inside waitForAsyncMessage {} {}", this,
+                        node.getMessageId(), futureOrLong);
+                }
+            }
+        }
+    }
 }

activemq-broker/src/main/java/org/apache/activemq/broker/region/virtual/BaseVirtualDestinationFilter.java

@@ -0,0 +1,39 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.activemq.broker.region.virtual;
+
+import org.apache.activemq.broker.region.BaseDestination;
+import org.apache.activemq.broker.region.Destination;
+import org.apache.activemq.broker.region.DestinationFilter;
+
+import java.util.Optional;
+
+public class BaseVirtualDestinationFilter extends DestinationFilter {
+
+    public BaseVirtualDestinationFilter(Destination next) {
+        super(next);
+    }
+
+    BaseDestination getBaseDestination(Destination virtualDest) {
+        if (virtualDest instanceof BaseDestination) {
+            return (BaseDestination) virtualDest;
+        } else if (virtualDest instanceof DestinationFilter) {
+            return ((DestinationFilter) virtualDest).getAdaptor(BaseDestination.class);
+        }
+        return null;
+    }
+}

activemq-broker/src/main/java/org/apache/activemq/broker/region/virtual/MappedQueueFilter.java

@@ -16,12 +16,12 @@
  */
 package org.apache.activemq.broker.region.virtual;
 
+import java.util.Optional;
 import java.util.Set;
 
 import org.apache.activemq.broker.ConnectionContext;
 import org.apache.activemq.broker.region.BaseDestination;
 import org.apache.activemq.broker.region.Destination;
-import org.apache.activemq.broker.region.DestinationFilter;
 import org.apache.activemq.broker.region.IndirectMessageReference;
 import org.apache.activemq.broker.region.RegionBroker;
 import org.apache.activemq.broker.region.Subscription;
@@ -34,7 +34,7 @@
  * Creates a mapped Queue that can recover messages from subscription recovery
  * policy of its Virtual Topic.
  */
-public class MappedQueueFilter extends DestinationFilter {
+public class MappedQueueFilter extends BaseVirtualDestinationFilter {
 
     private final ActiveMQDestination virtualDestination;
 
@@ -87,15 +87,6 @@ public synchronized void addSubscription(ConnectionContext context, Subscription
         }
     }
 
-    private BaseDestination getBaseDestination(Destination virtualDest) {
-        if (virtualDest instanceof BaseDestination) {
-            return (BaseDestination) virtualDest;
-        } else if (virtualDest instanceof DestinationFilter) {
-            return ((DestinationFilter) virtualDest).getAdaptor(BaseDestination.class);
-        }
-        return null;
-    }
-
     @Override
     public synchronized void removeSubscription(ConnectionContext context, Subscription sub, long lastDeliveredSequenceId) throws Exception {
         super.removeSubscription(context, sub, lastDeliveredSequenceId);

activemq-broker/src/main/java/org/apache/activemq/broker/region/virtual/SelectorAwareVirtualTopicInterceptor.java

@@ -17,9 +17,9 @@
 package org.apache.activemq.broker.region.virtual;
 
 import org.apache.activemq.broker.Broker;
+import org.apache.activemq.broker.region.BaseDestination;
 import org.apache.activemq.broker.region.Destination;
 import org.apache.activemq.broker.region.Subscription;
-import org.apache.activemq.broker.region.Topic;
 import org.apache.activemq.command.Message;
 import org.apache.activemq.filter.BooleanExpression;
 import org.apache.activemq.filter.MessageEvaluationContext;
@@ -41,8 +41,11 @@ public class SelectorAwareVirtualTopicInterceptor extends VirtualTopicIntercepto
 
     public SelectorAwareVirtualTopicInterceptor(Destination next, VirtualTopic virtualTopic) {
         super(next, virtualTopic);
+        BaseDestination baseDestination = getBaseDestination(next);
         selectorCachePlugin = (SubQueueSelectorCacheBroker)
-                ((Topic)next).createConnectionContext().getBroker().getAdaptor(SubQueueSelectorCacheBroker.class);
+                (baseDestination != null
+                ? baseDestination.createConnectionContext().getBroker().getAdaptor(SubQueueSelectorCacheBroker.class)
+                : null);
     }
 
     /**

activemq-broker/src/main/java/org/apache/activemq/broker/region/virtual/VirtualTopicInterceptor.java

@@ -25,7 +25,6 @@
 import org.apache.activemq.broker.ConnectionContext;
 import org.apache.activemq.broker.ProducerBrokerExchange;
 import org.apache.activemq.broker.region.Destination;
-import org.apache.activemq.broker.region.DestinationFilter;
 import org.apache.activemq.broker.region.Topic;
 import org.apache.activemq.command.ActiveMQDestination;
 import org.apache.activemq.command.ActiveMQQueue;
@@ -39,7 +38,7 @@
 /**
  * A Destination which implements <a href="https://activemq.apache.org/virtual-destinations">Virtual Topic</a>
  */
-public class VirtualTopicInterceptor extends DestinationFilter {
+public class VirtualTopicInterceptor extends BaseVirtualDestinationFilter {
 
     private final String prefix;
     private final String postfix;

activemq-cf/pom.xml

@@ -24,7 +24,7 @@
   <parent>
     <groupId>org.apache.activemq</groupId>
     <artifactId>activemq-parent</artifactId>
-    <version>6.1.3-TT.4-SNAPSHOT</version>
+    <version>6.1.5-TT.1-SNAPSHOT</version>
   </parent>
 
   <artifactId>activemq-cf</artifactId>

activemq-client/pom.xml

@@ -22,7 +22,7 @@
   <parent>
     <groupId>org.apache.activemq</groupId>
     <artifactId>activemq-parent</artifactId>
-    <version>6.1.3-TT.4-SNAPSHOT</version>
+    <version>6.1.5-TT.1-SNAPSHOT</version>
   </parent>
 
   <artifactId>activemq-client</artifactId>

activemq-client/src/main/java/org/apache/activemq/util/DataByteArrayOutputStream.java

@@ -210,7 +210,6 @@ public void writeUTF(String text) throws IOException {
         ensureEnoughBuffer((int)(pos + encodedsize + 2));
         writeShort((int)encodedsize);
 
-        byte[] buffer = new byte[(int)encodedsize];
         MarshallingSupport.writeUTFBytesToBuffer(text, (int) encodedsize, buf, pos);
         pos += encodedsize;
     }