activemq-all/pom.xml

@@ -14,7 +14,7 @@
   <parent>
     <groupId>org.apache.activemq</groupId>
     <artifactId>activemq-parent</artifactId>
-    <version>5.15.4</version>
+    <version>5.15.6</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.15.4</version>
+    <version>5.15.6</version>
   </parent>
 
   <artifactId>activemq-amqp</artifactId>

activemq-amqp/src/main/java/org/apache/activemq/transport/amqp/message/AMQPNativeOutboundTransformer.java

@@ -41,12 +41,16 @@ public class AMQPNativeOutboundTransformer implements OutboundTransformer {
     }
 
     static EncodedMessage transform(OutboundTransformer options, ActiveMQBytesMessage message) throws JMSException {
-        long messageFormat;
+        final long messageFormat;
+        if (message.propertyExists(JMS_AMQP_MESSAGE_FORMAT)) {
             try {
                 messageFormat = message.getLongProperty(JMS_AMQP_MESSAGE_FORMAT);
             } catch (MessageFormatException e) {
                 return null;
             }
+        } else {
+            messageFormat = 0;
+        }
 
         Binary encodedMessage = getBinaryFromMessageBody(message);
         byte encodedData[] = encodedMessage.getArray();

activemq-amqp/src/main/java/org/apache/activemq/transport/amqp/message/JMSMappingOutboundTransformer.java

@@ -88,7 +88,6 @@ import org.apache.qpid.proton.amqp.messaging.Section;
 import org.apache.qpid.proton.codec.AMQPDefinedTypes;
 import org.apache.qpid.proton.codec.DecoderImpl;
 import org.apache.qpid.proton.codec.EncoderImpl;
-import org.fusesource.hawtbuf.UTF8Buffer;
 
 public class JMSMappingOutboundTransformer implements OutboundTransformer {
 
@@ -102,11 +101,17 @@ public class JMSMappingOutboundTransformer implements OutboundTransformer {
     public static final byte TEMP_QUEUE_TYPE = 0x02;
     public static final byte TEMP_TOPIC_TYPE = 0x03;
 
+    private final UTF8BufferType utf8BufferEncoding;
+
     // For now Proton requires that we create a decoder to create an encoder
     private final DecoderImpl decoder = new DecoderImpl();
     private final EncoderImpl encoder = new EncoderImpl(decoder);
     {
         AMQPDefinedTypes.registerAllTypes(decoder, encoder);
+
+        utf8BufferEncoding = new UTF8BufferType(encoder, decoder);
+
+        encoder.register(utf8BufferEncoding);
     }
 
     @Override
@@ -159,7 +164,7 @@ public class JMSMappingOutboundTransformer implements OutboundTransformer {
             }
             properties.setTo(destination.getQualifiedName());
             if (maMap == null) {
-                maMap = new HashMap<Symbol, Object>();
+                maMap = new HashMap<>();
             }
             maMap.put(JMS_DEST_TYPE_MSG_ANNOTATION, destinationType(destination));
         }
@@ -170,7 +175,7 @@ public class JMSMappingOutboundTransformer implements OutboundTransformer {
             }
             properties.setReplyTo(replyTo.getQualifiedName());
             if (maMap == null) {
-                maMap = new HashMap<Symbol, Object>();
+                maMap = new HashMap<>();
             }
             maMap.put(JMS_REPLY_TO_TYPE_MSG_ANNOTATION, destinationType(replyTo));
         }
@@ -250,9 +255,6 @@ public class JMSMappingOutboundTransformer implements OutboundTransformer {
         for (Map.Entry<String, Object> entry : entries.entrySet()) {
             String key = entry.getKey();
             Object value = entry.getValue();
-            if (value instanceof UTF8Buffer) {
-                value = value.toString();
-            }
 
             if (key.startsWith(JMS_AMQP_PREFIX)) {
                 if (key.startsWith(NATIVE, JMS_AMQP_PREFIX_LENGTH)) {
@@ -276,7 +278,7 @@ public class JMSMappingOutboundTransformer implements OutboundTransformer {
                     continue;
                 } else if (key.startsWith(MESSAGE_ANNOTATION_PREFIX, JMS_AMQP_PREFIX_LENGTH)) {
                     if (maMap == null) {
-                        maMap = new HashMap<Symbol, Object>();
+                        maMap = new HashMap<>();
                     }
                     String name = key.substring(JMS_AMQP_MESSAGE_ANNOTATION_PREFIX.length());
                     maMap.put(Symbol.valueOf(name), value);
@@ -307,14 +309,14 @@ public class JMSMappingOutboundTransformer implements OutboundTransformer {
                     continue;
                 } else if (key.startsWith(DELIVERY_ANNOTATION_PREFIX, JMS_AMQP_PREFIX_LENGTH)) {
                     if (daMap == null) {
-                        daMap = new HashMap<Symbol, Object>();
+                        daMap = new HashMap<>();
                     }
                     String name = key.substring(JMS_AMQP_DELIVERY_ANNOTATION_PREFIX.length());
                     daMap.put(Symbol.valueOf(name), value);
                     continue;
                 } else if (key.startsWith(FOOTER_PREFIX, JMS_AMQP_PREFIX_LENGTH)) {
                     if (footerMap == null) {
-                        footerMap = new HashMap<Object, Object>();
+                        footerMap = new HashMap<>();
                     }
                     String name = key.substring(JMS_AMQP_FOOTER_PREFIX.length());
                     footerMap.put(name, value);
@@ -328,7 +330,7 @@ public class JMSMappingOutboundTransformer implements OutboundTransformer {
             // The property didn't map into any other slot so we store it in the
             // Application Properties section of the message.
             if (apMap == null) {
-                apMap = new HashMap<String, Object>();
+                apMap = new HashMap<>();
             }
             apMap.put(key, value);
         }
@@ -409,7 +411,7 @@ public class JMSMappingOutboundTransformer implements OutboundTransformer {
         } else if (messageType == CommandTypes.ACTIVEMQ_MAP_MESSAGE) {
             body = new AmqpValue(getMapFromMessageBody((ActiveMQMapMessage) message));
         } else if (messageType == CommandTypes.ACTIVEMQ_STREAM_MESSAGE) {
-            ArrayList<Object> list = new ArrayList<Object>();
+            ArrayList<Object> list = new ArrayList<>();
             final ActiveMQStreamMessage m = (ActiveMQStreamMessage) message;
             try {
                 while (true) {

activemq-amqp/src/main/java/org/apache/activemq/transport/amqp/message/UTF8BufferType.java

+/*
+ * 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.transport.amqp.message;
+
+import java.util.Arrays;
+import java.util.Collection;
+
+import org.apache.qpid.proton.codec.DecoderImpl;
+import org.apache.qpid.proton.codec.EncoderImpl;
+import org.apache.qpid.proton.codec.EncodingCodes;
+import org.apache.qpid.proton.codec.PrimitiveType;
+import org.apache.qpid.proton.codec.PrimitiveTypeEncoding;
+import org.apache.qpid.proton.codec.TypeEncoding;
+import org.apache.qpid.proton.codec.WritableBuffer;
+import org.fusesource.hawtbuf.UTF8Buffer;
+
+/**
+ * AMQP Type used to allow to proton-j codec to deal with UTF8Buffer types as if
+ * they were String elements.
+ */
+public class UTF8BufferType implements PrimitiveType<UTF8Buffer> {
+
+    private final UTF8BufferEncoding largeBufferEncoding;
+    private final UTF8BufferEncoding smallBufferEncoding;
+
+    public UTF8BufferType(EncoderImpl encoder, DecoderImpl decoder) {
+        this.largeBufferEncoding = new LargeUTF8BufferEncoding(encoder, decoder);
+        this.smallBufferEncoding = new SmallUTF8BufferEncoding(encoder, decoder);
+    }
+
+    @Override
+    public Class<UTF8Buffer> getTypeClass() {
+        return UTF8Buffer.class;
+    }
+
+    @Override
+    public PrimitiveTypeEncoding<UTF8Buffer> getEncoding(UTF8Buffer value) {
+        return value.getLength() <= 255 ? smallBufferEncoding : largeBufferEncoding;
+    }
+
+    @Override
+    public PrimitiveTypeEncoding<UTF8Buffer> getCanonicalEncoding() {
+        return largeBufferEncoding;
+    }
+
+    @Override
+    public Collection<? extends PrimitiveTypeEncoding<UTF8Buffer>> getAllEncodings() {
+        return Arrays.asList(smallBufferEncoding, largeBufferEncoding);
+    }
+
+    @Override
+    public void write(UTF8Buffer value) {
+        final TypeEncoding<UTF8Buffer> encoding = getEncoding(value);
+        encoding.writeConstructor();
+        encoding.writeValue(value);
+    }
+
+    public abstract class UTF8BufferEncoding implements PrimitiveTypeEncoding<UTF8Buffer> {
+
+        private final EncoderImpl encoder;
+        private final DecoderImpl decoder;
+
+        public UTF8BufferEncoding(EncoderImpl encoder, DecoderImpl decoder) {
+            this.encoder = encoder;
+            this.decoder = decoder;
+        }
+
+        @Override
+        public int getConstructorSize() {
+            return 1;
+        }
+
+        @Override
+        public boolean isFixedSizeVal() {
+            return false;
+        }
+
+        @Override
+        public boolean encodesJavaPrimitive() {
+            return false;
+        }
+
+        /**
+         * @return the number of bytes the size portion of the encoded value requires.
+         */
+        public abstract int getSizeBytes();
+
+        @Override
+        public void writeConstructor() {
+            getEncoder().writeRaw(getEncodingCode());
+        }
+
+        @Override
+        public void writeValue(UTF8Buffer value) {
+            writeSize(value);
+            WritableBuffer buffer = getEncoder().getBuffer();
+            buffer.put(value.getData(), value.getOffset(), value.getLength());
+        }
+
+        /**
+         * Write the size of the buffer using the appropriate type (byte or int) depending
+         * on the encoding type being used.
+         *
+         * @param value
+         *      The UTF8Buffer value that is being encoded.
+         */
+        public abstract void writeSize(UTF8Buffer value);
+
+        @Override
+        public int getValueSize(UTF8Buffer value) {
+            return getSizeBytes() + value.getLength();
+        }
+
+        @Override
+        public Class<UTF8Buffer> getTypeClass() {
+            return UTF8Buffer.class;
+        }
+
+        @Override
+        public PrimitiveType<UTF8Buffer> getType() {
+            return UTF8BufferType.this;
+        }
+
+        @Override
+        public boolean encodesSuperset(TypeEncoding<UTF8Buffer> encoding) {
+            return (getType() == encoding.getType());
+        }
+
+        @Override
+        public UTF8Buffer readValue() {
+            throw new UnsupportedOperationException("No decoding to UTF8Buffer exists");
+        }
+
+        @Override
+        public void skipValue() {
+            throw new UnsupportedOperationException("No decoding to UTF8Buffer exists");
+        }
+
+        public DecoderImpl getDecoder() {
+            return decoder;
+        }
+
+        public EncoderImpl getEncoder() {
+            return encoder;
+        }
+    }
+
+    public class LargeUTF8BufferEncoding extends UTF8BufferEncoding {
+
+        public LargeUTF8BufferEncoding(EncoderImpl encoder, DecoderImpl decoder) {
+            super(encoder, decoder);
+        }
+
+        @Override
+        public byte getEncodingCode() {
+            return EncodingCodes.STR32;
+        }
+
+        @Override
+        public int getSizeBytes() {
+            return Integer.BYTES;
+        }
+
+        @Override
+        public void writeSize(UTF8Buffer value) {
+            getEncoder().getBuffer().putInt(value.getLength());
+        }
+    }
+
+    public class SmallUTF8BufferEncoding extends UTF8BufferEncoding {
+
+        public SmallUTF8BufferEncoding(EncoderImpl encoder, DecoderImpl decoder) {
+            super(encoder, decoder);
+        }
+
+        @Override
+        public byte getEncodingCode() {
+            return EncodingCodes.STR8;
+        }
+
+        @Override
+        public int getSizeBytes() {
+            return Byte.BYTES;
+        }
+
+        @Override
+        public void writeSize(UTF8Buffer value) {
+            getEncoder().getBuffer().put((byte) value.getLength());
+        }
+    }
+}

activemq-amqp/src/main/java/org/apache/activemq/transport/amqp/protocol/AmqpSender.java

@@ -17,7 +17,6 @@
 package org.apache.activemq.transport.amqp.protocol;
 
 import static org.apache.activemq.transport.amqp.AmqpSupport.toLong;
-import static org.apache.activemq.transport.amqp.message.AmqpMessageSupport.JMS_AMQP_MESSAGE_FORMAT;
 
 import java.io.IOException;
 import java.util.LinkedList;
@@ -449,22 +448,8 @@ public class AmqpSender extends AmqpAbstractLink<Sender> {
 
                 ActiveMQMessage temp = null;
                 if (md.getMessage() != null) {
-
-                    // Topics can dispatch the same Message to more than one consumer
-                    // so we must copy to prevent concurrent read / write to the same
-                    // message object.
-                    if (md.getDestination().isTopic()) {
-                        synchronized (md.getMessage()) {
                     temp = (ActiveMQMessage) md.getMessage().copy();
                 }
-                    } else {
-                        temp = (ActiveMQMessage) md.getMessage();
-                    }
-
-                    if (!temp.getProperties().containsKey(JMS_AMQP_MESSAGE_FORMAT)) {
-                        temp.setProperty(JMS_AMQP_MESSAGE_FORMAT, 0);
-                    }
-                }
 
                 final ActiveMQMessage jms = temp;
                 if (jms == null) {

activemq-amqp/src/test/java/org/apache/activemq/transport/amqp/JMSClientTest.java

@@ -36,6 +36,7 @@ import javax.jms.DeliveryMode;
 import javax.jms.Destination;
 import javax.jms.ExceptionListener;
 import javax.jms.JMSException;
+import javax.jms.MapMessage;
 import javax.jms.Message;
 import javax.jms.MessageConsumer;
 import javax.jms.MessageListener;
@@ -99,6 +100,33 @@ public class JMSClientTest extends JMSClientTestSupport {
         }
     }
 
+    @Test(timeout = 60000)
+    public void testSendJMSMapMessage() throws Exception {
+        ActiveMQAdmin.enableJMSFrameTracing();
+
+        connection = createConnection();
+        {
+            Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
+            assertNotNull(session);
+            Queue queue = session.createQueue(name.getMethodName());
+            MessageProducer producer = session.createProducer(queue);
+            MapMessage message = session.createMapMessage();
+            message.setBoolean("Boolean", false);
+            message.setString("STRING", "TEST");
+            producer.send(message);
+            QueueViewMBean proxy = getProxyToQueue(name.getMethodName());
+            assertEquals(1, proxy.getQueueSize());
+
+            MessageConsumer consumer = session.createConsumer(queue);
+            Message received = consumer.receive(5000);
+            assertNotNull(received);
+            assertTrue(received instanceof MapMessage);
+            MapMessage map = (MapMessage) received;
+            assertEquals("TEST", map.getString("STRING"));
+            assertEquals(false, map.getBooleanProperty("Boolean"));
+        }
+    }
+
     @Test(timeout=30000)
     public void testAnonymousProducerConsume() throws Exception {
         ActiveMQAdmin.enableJMSFrameTracing();

activemq-amqp/src/test/java/org/apache/activemq/transport/amqp/interop/OpenWireToAmqpConcurrentStoreAndDispatchTest.java

+/*
+ * 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.transport.amqp.interop;
+
+import org.apache.activemq.ActiveMQConnectionFactory;
+import org.apache.activemq.command.ActiveMQQueue;
+import org.apache.activemq.transport.amqp.client.AmqpClient;
+import org.apache.activemq.transport.amqp.client.AmqpClientTestSupport;
+import org.apache.activemq.transport.amqp.client.AmqpConnection;
+import org.apache.activemq.transport.amqp.client.AmqpMessage;
+import org.apache.activemq.transport.amqp.client.AmqpReceiver;
+import org.apache.activemq.transport.amqp.client.AmqpSession;
+import org.junit.Ignore;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.junit.runners.Parameterized;
+import org.junit.runners.Parameterized.Parameters;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.jms.BytesMessage;
+import javax.jms.Connection;
+import javax.jms.JMSException;
+import javax.jms.MessageProducer;
+import javax.jms.Session;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.Random;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import static org.apache.activemq.store.kahadb.KahaDBStore.PROPERTY_CANCELED_TASK_MOD_METRIC;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+
+@RunWith(Parameterized.class)
+public class OpenWireToAmqpConcurrentStoreAndDispatchTest extends AmqpClientTestSupport {
+
+    protected static final Logger LOG = LoggerFactory.getLogger(OpenWireToAmqpConcurrentStoreAndDispatchTest.class);
+
+    private final String transformer;
+
+    @Parameters(name="Transformer->{0}")
+    public static Collection<Object[]> data() {
+        System.setProperty(PROPERTY_CANCELED_TASK_MOD_METRIC, "100");
+        return Arrays.asList(new Object[][] {
+                {"jms"}
+            });
+    }
+
+    public OpenWireToAmqpConcurrentStoreAndDispatchTest(String transformer) {
+        this.transformer = transformer;
+    }
+
+    @Override
+    protected String getAmqpTransformer() {
+        return transformer;
+    }
+
+    @Override
+    protected boolean isPersistent() {
+        return true;
+    }
+
+    @Override
+    protected boolean isUseOpenWireConnector() {
+        return true;
+    }
+
+    @Test(timeout = 6000000)
+    @Ignore("takes more than 6 mins to complete but fails earlier without fix")
+    public void testNoErrorOnSend() throws Exception {
+
+        final int numIterations = 100;
+        int numConsumers = 3;
+        final int numProducers = 10;
+        final int numMessages = 2000;
+        final AtomicBoolean done = new AtomicBoolean(false);
+        final AtomicInteger sent = new AtomicInteger();
+        final AtomicInteger received = new AtomicInteger();
+        final AtomicBoolean errorOnSend = new AtomicBoolean(false);
+
+        final AtomicInteger toSend = new AtomicInteger(numMessages);
+
+        final Random random = new Random();
+        for (int i=0; i<numIterations; i++) {
+            done.set(false);
+            sent.set(0);
+            received.set(0);
+            toSend.set(numMessages);
+
+            ExecutorService executorService = Executors.newCachedThreadPool();
+            for (int j = 0; j < numConsumers; j++) {
+                executorService.execute(new Runnable() {
+                    @Override
+                    public void run() {
+                        AmqpConnection connection = null;
+                        try {
+                            AmqpClient client = createAmqpClient();
+                            connection = trackConnection(client.connect());
+                            AmqpSession session = connection.createSession();
+
+                            AmqpReceiver receiver = session.createReceiver("queue://" + getTestName(), null, false, true);
+
+                            while (!done.get() && received.get() < numMessages) {
+                                receiver.flow(1);
+                                AmqpMessage message = receiver.receive(5, TimeUnit.SECONDS);
+                                if (message != null) {
+                                    received.incrementAndGet();
+                                }
+                            }
+                        } catch (Exception e) {
+                            e.printStackTrace();
+                        } finally {
+                            if (connection != null) {
+                                connection.close();
+                            }
+                        }
+                    }
+                });
+            }
+
+            final byte[] payload = new byte[100];
+            for (int k = 0; k < numProducers; k++) {
+                executorService.execute(new Runnable() {
+                    @Override
+                    public void run() {
+                        Connection connection = null;
+                        try {
+                            ActiveMQConnectionFactory connectionFactory =
+                                    new ActiveMQConnectionFactory(brokerService.getTransportConnectorByScheme("tcp").getPublishableConnectString());
+                            connection = connectionFactory.createConnection();
+                            Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
+                            MessageProducer producer = session.createProducer(new ActiveMQQueue(getTestName()));
+                            BytesMessage bytesMessage = session.createBytesMessage();
+                            bytesMessage.writeBytes(payload);
+                            bytesMessage.setStringProperty("PP", "VALUE");
+                            while (!done.get() && toSend.decrementAndGet() >= 0) {
+                                producer.send(bytesMessage);
+                                sent.incrementAndGet();
+                            }
+                        } catch (Exception e) {
+                            e.printStackTrace();
+                            errorOnSend.set(true);
+                        } finally {
+                            if (connection != null) {
+                                try {
+                                    connection.close();
+                                } catch (JMSException ignored) {}
+                            }
+                        }
+                    }
+                });
+            }
+
+            executorService.shutdown();
+            executorService.awaitTermination(30, TimeUnit.SECONDS);
+
+            done.set(true);
+            assertEquals("[" + i + "] sent all requested", numMessages, sent.get());
+            assertEquals("[" + i + "] got all sent", numMessages, received.get());
+            assertFalse("[" + i + "] no error on send", errorOnSend.get());
+        }
+    }
+}

activemq-amqp/src/test/java/org/apache/activemq/transport/amqp/message/UTF8BufferTypeTest.java

+/*
+ * 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.transport.amqp.message;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertTrue;
+
+import java.nio.charset.StandardCharsets;
+import java.util.UUID;
+
+import org.apache.qpid.proton.codec.AMQPDefinedTypes;
+import org.apache.qpid.proton.codec.DecoderImpl;
+import org.apache.qpid.proton.codec.EncoderImpl;
+import org.apache.qpid.proton.codec.EncodingCodes;
+import org.apache.qpid.proton.codec.PrimitiveTypeEncoding;
+import org.apache.qpid.proton.codec.ReadableBuffer;
+import org.fusesource.hawtbuf.UTF8Buffer;
+import org.junit.Before;
+import org.junit.Test;
+
+/**
+ * Test the UTF8Buffer type encoder
+ */
+public class UTF8BufferTypeTest {
+
+    private final UTF8BufferType utf8BufferEncoding;
+    private final DecoderImpl decoder = new DecoderImpl();
+    private final EncoderImpl encoder = new EncoderImpl(decoder);
+    {
+        AMQPDefinedTypes.registerAllTypes(decoder, encoder);
+
+        utf8BufferEncoding = new UTF8BufferType(encoder, decoder);
+
+        encoder.register(utf8BufferEncoding);
+    }
+
+    private String smallString = UUID.randomUUID().toString();
+    private String largeString = UUID.randomUUID().toString() + UUID.randomUUID().toString() +
+                                 UUID.randomUUID().toString() + UUID.randomUUID().toString() +
+                                 UUID.randomUUID().toString() + UUID.randomUUID().toString() +
+                                 UUID.randomUUID().toString() + UUID.randomUUID().toString();
+
+    private UTF8Buffer smallBuffer;
+    private UTF8Buffer largeBuffer;
+
+    @Before
+    public void setUp() {
+        smallBuffer = new UTF8Buffer(smallString.getBytes(StandardCharsets.UTF_8));
+        largeBuffer = new UTF8Buffer(largeString.getBytes(StandardCharsets.UTF_8));
+    }
+
+    @Test
+    public void testGetAllEncodings() {
+        assertEquals(2, utf8BufferEncoding.getAllEncodings().size());
+    }
+
+    @Test
+    public void testGetTypeClass() {
+        assertEquals(UTF8Buffer.class, utf8BufferEncoding.getTypeClass());
+    }
+
+    @Test
+    public void testGetCanonicalEncoding() {
+        assertNotNull(utf8BufferEncoding.getCanonicalEncoding());
+    }
+
+    @Test
+    public void testGetEncodingForSmallUTF8Buffer() {
+        PrimitiveTypeEncoding<UTF8Buffer> encoding = utf8BufferEncoding.getEncoding(smallBuffer);
+
+        assertTrue(encoding instanceof UTF8BufferType.SmallUTF8BufferEncoding);
+        assertEquals(1, encoding.getConstructorSize());
+        assertEquals(smallBuffer.getLength() + Byte.BYTES, encoding.getValueSize(smallBuffer));
+        assertEquals(EncodingCodes.STR8, encoding.getEncodingCode());
+        assertFalse(encoding.encodesJavaPrimitive());
+        assertEquals(utf8BufferEncoding, encoding.getType());
+    }
+
+    @Test
+    public void testGetEncodingForLargeUTF8Buffer() {
+        PrimitiveTypeEncoding<UTF8Buffer> encoding = utf8BufferEncoding.getEncoding(largeBuffer);
+
+        assertTrue(encoding instanceof UTF8BufferType.LargeUTF8BufferEncoding);
+        assertEquals(1, encoding.getConstructorSize());
+        assertEquals(largeBuffer.getLength() + Integer.BYTES, encoding.getValueSize(largeBuffer));
+        assertEquals(EncodingCodes.STR32, encoding.getEncodingCode());
+        assertFalse(encoding.encodesJavaPrimitive());
+        assertEquals(utf8BufferEncoding, encoding.getType());
+    }
+
+    @Test
+    public void testEncodeDecodeEmptyStringBuffer() {
+        final AmqpWritableBuffer buffer = new AmqpWritableBuffer();
+        encoder.setByteBuffer(buffer);
+        encoder.writeObject(new UTF8Buffer(""));
+
+        byte[] copy = new byte[buffer.getArrayLength()];
+        System.arraycopy(buffer.getArray(), 0, copy, 0, buffer.getArrayLength());
+
+        ReadableBuffer encoded = ReadableBuffer.ByteBufferReader.wrap(copy);
+        decoder.setBuffer(encoded);
+
+        Object valueRead = decoder.readObject();
+        assertTrue(valueRead instanceof String);
+        String decodedString = (String) valueRead;
+        assertEquals("", decodedString);
+    }
+
+    @Test
+    public void testEncodeDecodeSmallBuffer() {
+        final AmqpWritableBuffer buffer = new AmqpWritableBuffer();
+        encoder.setByteBuffer(buffer);
+        encoder.writeObject(smallBuffer);
+
+        byte[] copy = new byte[buffer.getArrayLength()];
+        System.arraycopy(buffer.getArray(), 0, copy, 0, buffer.getArrayLength());
+
+        ReadableBuffer encoded = ReadableBuffer.ByteBufferReader.wrap(copy);
+        decoder.setBuffer(encoded);
+
+        Object valueRead = decoder.readObject();
+        assertTrue(valueRead instanceof String);
+        String decodedString = (String) valueRead;
+        assertEquals(smallString, decodedString);
+    }
+
+    @Test
+    public void testEncodeDecodeLargeBuffer() {
+        final AmqpWritableBuffer buffer = new AmqpWritableBuffer();
+        encoder.setByteBuffer(buffer);
+        encoder.writeObject(largeBuffer);
+
+        byte[] copy = new byte[buffer.getArrayLength()];
+        System.arraycopy(buffer.getArray(), 0, copy, 0, buffer.getArrayLength());
+
+        ReadableBuffer encoded = ReadableBuffer.ByteBufferReader.wrap(copy);
+        decoder.setBuffer(encoded);
+
+        Object valueRead = decoder.readObject();
+        assertTrue(valueRead instanceof String);
+        String decodedString = (String) valueRead;
+        assertEquals(largeString, decodedString);
+    }
+}

activemq-blueprint/pom.xml

@@ -22,7 +22,7 @@
   <parent>
     <groupId>org.apache.activemq</groupId>
     <artifactId>activemq-parent</artifactId>
-    <version>5.15.4</version>
+    <version>5.15.6</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.15.4</version>
+    <version>5.15.6</version>
   </parent>
 
   <artifactId>activemq-broker</artifactId>

activemq-broker/src/main/java/org/apache/activemq/broker/ConnectionContext.java

@@ -27,6 +27,7 @@ import org.apache.activemq.command.TransactionId;
 import org.apache.activemq.command.WireFormatInfo;
 import org.apache.activemq.command.XATransactionId;
 import org.apache.activemq.filter.MessageEvaluationContext;
+import org.apache.activemq.filter.NonCachedMessageEvaluationContext;
 import org.apache.activemq.security.MessageAuthorizationPolicy;
 import org.apache.activemq.security.SecurityContext;
 import org.apache.activemq.state.ConnectionState;
@@ -64,7 +65,7 @@ public class ConnectionContext {
     private XATransactionId xid;
 
     public ConnectionContext() {
-        this.messageEvaluationContext = new MessageEvaluationContext();
+        this.messageEvaluationContext = new NonCachedMessageEvaluationContext();
     }
 
     public ConnectionContext(MessageEvaluationContext messageEvaluationContext) {

activemq-broker/src/main/java/org/apache/activemq/broker/ProducerBrokerExchange.java

@@ -16,10 +16,6 @@
  */
 package org.apache.activemq.broker;
 
-import java.io.IOException;
-import java.util.concurrent.atomic.AtomicBoolean;
-import java.util.concurrent.atomic.AtomicLong;
-
 import org.apache.activemq.broker.region.Destination;
 import org.apache.activemq.broker.region.Region;
 import org.apache.activemq.command.Message;
@@ -28,6 +24,10 @@ import org.apache.activemq.state.ProducerState;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import java.io.IOException;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicLong;
+
 /**
  * Holds internal state in the broker for a MessageProducer
  */
@@ -213,7 +213,7 @@ public class ProducerBrokerExchange {
     }
 
     public int getPercentageBlocked() {
-        double value = flowControlInfo.getSendsBlocked() / flowControlInfo.getTotalSends();
+        double value = flowControlInfo.getTotalSends() == 0 ? 0 : flowControlInfo.getSendsBlocked() / flowControlInfo.getTotalSends();
         return (int) value * 100;
     }
 

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

@@ -51,7 +51,7 @@ import org.apache.activemq.command.ActiveMQMessage;
 import org.apache.activemq.command.ActiveMQTextMessage;
 import org.apache.activemq.command.Message;
 import org.apache.activemq.filter.BooleanExpression;
-import org.apache.activemq.filter.MessageEvaluationContext;
+import org.apache.activemq.filter.NonCachedMessageEvaluationContext;
 import org.apache.activemq.selector.SelectorParser;
 import org.apache.activemq.store.MessageStore;
 import org.apache.activemq.util.URISupport;
@@ -213,7 +213,7 @@ public class DestinationView implements DestinationViewMBean {
         Message[] messages = destination.browse();
         ArrayList<CompositeData> c = new ArrayList<CompositeData>();
 
-        MessageEvaluationContext ctx = new MessageEvaluationContext();
+        NonCachedMessageEvaluationContext ctx = new NonCachedMessageEvaluationContext();
         ctx.setDestination(destination.getActiveMQDestination());
         BooleanExpression selectorExpression = selector == null ? null : SelectorParser.parse(selector);
 
@@ -256,7 +256,7 @@ public class DestinationView implements DestinationViewMBean {
         Message[] messages = destination.browse();
         ArrayList<Object> answer = new ArrayList<Object>();
 
-        MessageEvaluationContext ctx = new MessageEvaluationContext();
+        NonCachedMessageEvaluationContext ctx = new NonCachedMessageEvaluationContext();
         ctx.setDestination(destination.getActiveMQDestination());
         BooleanExpression selectorExpression = selector == null ? null : SelectorParser.parse(selector);
 
@@ -297,7 +297,7 @@ public class DestinationView implements DestinationViewMBean {
         TabularType tt = new TabularType("MessageList", "MessageList", ct, new String[] { "JMSMessageID" });
         TabularDataSupport rc = new TabularDataSupport(tt);
 
-        MessageEvaluationContext ctx = new MessageEvaluationContext();
+        NonCachedMessageEvaluationContext ctx = new NonCachedMessageEvaluationContext();
         ctx.setDestination(destination.getActiveMQDestination());
         BooleanExpression selectorExpression = selector == null ? null : SelectorParser.parse(selector);
 

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

@@ -162,7 +162,7 @@ public abstract class AbstractRegion implements Region {
                     addSubscriptionsForDestination(context, dest);
                     destinations.put(destination, dest);
                     updateRegionDestCounts(destination, 1);
-                    destinationMap.put(destination, dest);
+                    destinationMap.unsynchronizedPut(destination, dest);
                 }
                 if (dest == null) {
                     throw new DestinationDoesNotExistException(destination.getQualifiedName());
@@ -217,7 +217,7 @@ public abstract class AbstractRegion implements Region {
                 // If a destination isn't specified, then just count up
                 // non-advisory destinations (ie count all destinations)
                 int destinationSize = (int) (entry.getDestination() != null ?
-                        destinationMap.get(entry.getDestination()).size() : regionStatistics.getDestinations().getCount());
+                        destinationMap.unsynchronizedGet(entry.getDestination()).size() : regionStatistics.getDestinations().getCount());
                 if (destinationSize >= entry.getMaxDestinations()) {
                     if (entry.getDestination() != null) {
                         throw new IllegalStateException(
@@ -296,7 +296,7 @@ public abstract class AbstractRegion implements Region {
                         dest.removeSubscription(context, sub, 0l);
                     }
                 }
-                destinationMap.remove(destination, dest);
+                destinationMap.unsynchronizedRemove(destination, dest);
                 dispose(context, dest);
                 DestinationInterceptor destinationInterceptor = broker.getDestinationInterceptor();
                 if (destinationInterceptor != null) {
@@ -321,7 +321,7 @@ public abstract class AbstractRegion implements Region {
     public Set<Destination> getDestinations(ActiveMQDestination destination) {
         destinationsLock.readLock().lock();
         try{
-            return destinationMap.get(destination);
+            return destinationMap.unsynchronizedGet(destination);
         } finally {
             destinationsLock.readLock().unlock();
         }
@@ -387,7 +387,7 @@ public abstract class AbstractRegion implements Region {
             List<Destination> addList = new ArrayList<Destination>();
             destinationsLock.readLock().lock();
             try {
-                for (Destination dest : (Set<Destination>) destinationMap.get(info.getDestination())) {
+                for (Destination dest : (Set<Destination>) destinationMap.unsynchronizedGet(info.getDestination())) {
                     addList.add(dest);
                 }
                 // ensure sub visible to any new dest addSubscriptionsForDestination
@@ -467,7 +467,7 @@ public abstract class AbstractRegion implements Region {
             List<Destination> removeList = new ArrayList<Destination>();
             destinationsLock.readLock().lock();
             try {
-                for (Destination dest : (Set<Destination>) destinationMap.get(info.getDestination())) {
+                for (Destination dest : (Set<Destination>) destinationMap.unsynchronizedGet(info.getDestination())) {
                     removeList.add(dest);
                 }
             } finally {
@@ -552,15 +552,7 @@ public abstract class AbstractRegion implements Region {
                 // Try to auto create the destination... re-invoke broker
                 // from the
                 // top so that the proper security checks are performed.
-                context.getBroker().addDestination(context, destination, createTemporary);
-                dest = addDestination(context, destination, false);
-                // We should now have the dest created.
-                destinationsLock.readLock().lock();
-                try {
-                    dest = destinations.get(destination);
-                } finally {
-                    destinationsLock.readLock().unlock();
-                }
+                dest = context.getBroker().addDestination(context, destination, createTemporary);
             }
 
             if (dest == null) {
@@ -644,7 +636,7 @@ public abstract class AbstractRegion implements Region {
     public void addProducer(ConnectionContext context, ProducerInfo info) throws Exception {
         destinationsLock.readLock().lock();
         try {
-            for (Destination dest : (Set<Destination>) destinationMap.get(info.getDestination())) {
+            for (Destination dest : (Set<Destination>) destinationMap.unsynchronizedGet(info.getDestination())) {
                 dest.addProducer(context, info);
             }
         } finally {
@@ -665,7 +657,7 @@ public abstract class AbstractRegion implements Region {
     public void removeProducer(ConnectionContext context, ProducerInfo info) throws Exception {
         destinationsLock.readLock().lock();
         try {
-            for (Destination dest : (Set<Destination>) destinationMap.get(info.getDestination())) {
+            for (Destination dest : (Set<Destination>) destinationMap.unsynchronizedGet(info.getDestination())) {
                 dest.removeProducer(context, info);
             }
         } finally {

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

@@ -841,7 +841,7 @@ public abstract class BaseDestination implements Destination {
     }
 
     public ConnectionContext createConnectionContext() {
-        ConnectionContext answer = new ConnectionContext(new NonCachedMessageEvaluationContext());
+        ConnectionContext answer = new ConnectionContext();
         answer.setBroker(this.broker);
         answer.getMessageEvaluationContext().setDestination(getActiveMQDestination());
         answer.setSecurityContext(SecurityContext.BROKER_SECURITY_CONTEXT);

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

@@ -1207,9 +1207,9 @@ public class Queue extends BaseDestination implements Task, UsageListener, Index
         }
 
         LOG.trace("max {}, alreadyPagedIn {}, messagesCount {}, memoryUsage {}%", new Object[]{max, alreadyPagedIn, messagesInQueue, memoryUsage.getPercentUsage()});
-        return (alreadyPagedIn < max)
+        return (alreadyPagedIn == 0 || (alreadyPagedIn < max)
                 && (alreadyPagedIn < messagesInQueue)
-                && messages.hasSpace();
+                && messages.hasSpace());
     }
 
     private void addAll(Collection<? extends MessageReference> refs, List<Message> l, int max,

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

@@ -484,7 +484,7 @@ public class FilePendingMessageCursor extends AbstractPendingMessageCursor imple
     private void discardExpiredMessage(MessageReference reference) {
         LOG.debug("Discarding expired message {}", reference);
         if (reference.isExpired() && broker.isExpired(reference)) {
-            ConnectionContext context = new ConnectionContext(new NonCachedMessageEvaluationContext());
+            ConnectionContext context = new ConnectionContext();
             context.setBroker(broker);
             ((Destination)reference.getRegionDestination()).messageExpired(context, null, new IndirectMessageReference(reference.getMessage()));
         }

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

@@ -59,7 +59,7 @@ public class MappedQueueFilter extends DestinationFilter {
             final Set<Destination> virtualDests = regionBroker.getDestinations(virtualDestination);
 
             final ActiveMQDestination newDestination = sub.getActiveMQDestination();
-            final BaseDestination regionDest = getBaseDestination((Destination) regionBroker.getDestinations(newDestination).toArray()[0]);
+            BaseDestination regionDest = null;
 
             for (Destination virtualDest : virtualDests) {
                 if (virtualDest.getActiveMQDestination().isTopic() &&
@@ -75,6 +75,9 @@ public class MappedQueueFilter extends DestinationFilter {
                             final Message copy = message.copy();
                             copy.setOriginalDestination(message.getDestination());
                             copy.setDestination(newDestination);
+                            if (regionDest == null) {
+                                regionDest = getBaseDestination((Destination) regionBroker.getDestinations(newDestination).toArray()[0]);
+                            }
                             copy.setRegionDestination(regionDest);
                             sub.addRecoveredMessage(context, newDestination.isQueue() ? new IndirectMessageReference(copy) : copy);
                         }

activemq-broker/src/main/java/org/apache/activemq/broker/scheduler/SchedulerBroker.java

@@ -55,6 +55,7 @@ import org.slf4j.LoggerFactory;
 public class SchedulerBroker extends BrokerFilter implements JobListener {
     private static final Logger LOG = LoggerFactory.getLogger(SchedulerBroker.class);
     private static final IdGenerator ID_GENERATOR = new IdGenerator();
+    private static final LongSequenceGenerator longGenerator = new LongSequenceGenerator();
     private final LongSequenceGenerator messageIdGenerator = new LongSequenceGenerator();
     private final AtomicBoolean started = new AtomicBoolean();
     private final WireFormat wireFormat = new OpenWireFormat();
@@ -337,7 +338,10 @@ public class SchedulerBroker extends BrokerFilter implements JobListener {
             repeat = (Integer) TypeConversionSupport.convert(repeatValue, Integer.class);
         }
 
-        getInternalScheduler().schedule(msg.getMessageId().toString(),
+        //job id should be unique for every job (Same format as MessageId)
+        MessageId jobId = new MessageId(messageSend.getMessageId().getProducerId(), longGenerator.getNextSequenceId());
+
+        getInternalScheduler().schedule(jobId.toString(),
                 new ByteSequence(packet.data, packet.offset, packet.length), cronEntry, delay, period, repeat);
     }