diff --git a/Jenkinsfile b/Jenkinsfile
index d4c09be..67d1cd0 100644
--- a/Jenkinsfile
+++ b/Jenkinsfile
@@ -87,6 +87,13 @@
             }
         }
 
+        stage('Verify') {
+            steps {
+                echo 'Running apache-rat:check'
+                sh 'mvn apache-rat:check'
+            }
+        }
+
         stage('Tests') {
             steps {
                 echo 'Running tests'
diff --git a/activemq-broker/pom.xml b/activemq-broker/pom.xml
index 25750c3..2b421a2 100644
--- a/activemq-broker/pom.xml
+++ b/activemq-broker/pom.xml
@@ -106,9 +106,9 @@
             <link>http://junit.sourceforge.net/javadoc/</link>
           </links>
           <stylesheetfile>${basedir}/../etc/css/stylesheet.css</stylesheetfile>
-          <linksource>true</linksource>
           <maxmemory>256m</maxmemory>
           <source>${source-version}</source>
+          <noindex>true</noindex>
           <groups>
             <group>
               <title>JMS Client</title>
diff --git a/activemq-broker/src/test/java/org/apache/activemq/ActiveMQJms2Test.java b/activemq-broker/src/test/java/org/apache/activemq/ActiveMQJms2Test.java
deleted file mode 100644
index 51e2614..0000000
--- a/activemq-broker/src/test/java/org/apache/activemq/ActiveMQJms2Test.java
+++ /dev/null
@@ -1,169 +0,0 @@
-/**
- * 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;
-
-import javax.jms.Connection;
-import javax.jms.JMSException;
-import javax.jms.MessageProducer;
-import javax.jms.Session;
-
-import org.junit.After;
-import org.junit.AfterClass;
-import org.junit.Before;
-import org.junit.BeforeClass;
-import org.junit.Test;
-
-public class ActiveMQJms2Test {
-
-    protected static ActiveMQConnectionFactory activemqConnectionFactory = null;
-
-    protected Connection connection = null;
-    protected Session session = null;
-    protected MessageProducer messageProducer = null;
-
-    @BeforeClass
-    public static void setUpClass() {
-        activemqConnectionFactory = new ActiveMQConnectionFactory("vm://localhost?marshal=false&broker.persistent=false"); 	
-    }
-
-    @AfterClass
-    public static void tearDownClass() {
-        activemqConnectionFactory = null;
-    }
-
-    @Before
-    public void setUp() throws JMSException {
-        connection = activemqConnectionFactory.createConnection();
-        connection.start();
-
-        session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
-
-        messageProducer = session.createProducer(session.createQueue("AMQ.JMS2.TEST"));
-    }
-
-    @After
-    public void tearDown() {
-        if(messageProducer != null) {
-            try { messageProducer.close(); } catch (Exception e) { } finally { messageProducer = null; }
-        }
-
-        if(session != null) {
-            try { session.close(); } catch (Exception e) { } finally { session = null; }
-        }
-
-        if(connection != null) {
-            try { connection.close(); } catch (Exception e) { } finally { connection = null; }
-        }
-    }
-
-    @Test(expected = UnsupportedOperationException.class)
-    public void testConnectionFactoryCreateContext() {
-        activemqConnectionFactory.createContext();
-    }
-
-    @Test(expected = UnsupportedOperationException.class)
-    public void testConnectionFactoryCreateContextSession() {
-        activemqConnectionFactory.createContext(Session.AUTO_ACKNOWLEDGE);
-    }
-
-    @Test(expected = UnsupportedOperationException.class)
-    public void testConnectionFactoryCreateContextUserPass() {
-        activemqConnectionFactory.createContext("admin", "admin");
-    }
-
-    @Test(expected = UnsupportedOperationException.class)
-    public void testConnectionFactoryCreateContextUserPassSession() {
-        activemqConnectionFactory.createContext("admin", "admin", Session.AUTO_ACKNOWLEDGE);
-    }
-
-    @Test(expected = UnsupportedOperationException.class)
-    public void testConnectionSharedConnectionConsumer() throws JMSException {
-        connection.createSharedConnectionConsumer(null, null, null, null, 10);
-    }
-
-    @Test(expected = UnsupportedOperationException.class)
-    public void testConnectionSharedDurableConnectionConsumer() throws JMSException {
-        connection.createSharedDurableConnectionConsumer(null, null, null, null, 10);
-    }
-
-    @Test(expected = UnsupportedOperationException.class)
-    public void testSessionAckMode() throws JMSException {
-        connection.createSession(Session.AUTO_ACKNOWLEDGE);
-    }
-
-    @Test(expected = UnsupportedOperationException.class)
-    public void testSessionDurableConsumer() throws JMSException {
-        session.createDurableConsumer(null, null);
-    }
-
-    @Test(expected = UnsupportedOperationException.class)
-    public void testSessionDurableConsumerSelectorNoLocal() throws JMSException {
-        session.createDurableConsumer(null, null, null, true);
-    }
-
-    @Test(expected = UnsupportedOperationException.class)
-    public void testSessionSharedConsumer() throws JMSException {
-        session.createSharedConsumer(null, null);
-    }
-
-    @Test(expected = UnsupportedOperationException.class)
-    public void testSessionSharedConsumerSelector() throws JMSException {
-        session.createSharedConsumer(null, null, null);
-    }
-
-    @Test(expected = UnsupportedOperationException.class)
-    public void testSessionSharedDurableConsumer() throws JMSException {
-        session.createSharedDurableConsumer(null, null);
-    }
-
-    @Test(expected = UnsupportedOperationException.class)
-    public void testSessionSharedDurableConsumerSelector() throws JMSException {
-        session.createSharedDurableConsumer(null, null, null);
-    }
-
-    @Test(expected = UnsupportedOperationException.class)
-    public void testProducerDeliveryDelayGet() throws JMSException {
-        messageProducer.getDeliveryDelay();
-    }
-
-    @Test(expected = UnsupportedOperationException.class)
-    public void testProducerDeliveryDelaySet() throws JMSException {
-        messageProducer.setDeliveryDelay(1000l);
-    }
-
-    @Test(expected = UnsupportedOperationException.class)
-    public void testProducerSendMessageCompletionListener() throws JMSException {
-         messageProducer.send(session.createQueue("AMQ.TEST"), null);
-    }
-
-    @Test(expected = UnsupportedOperationException.class)
-    public void testProducerSendMessageQoSParamsCompletionListener() throws JMSException {
-         messageProducer.send(null, 1, 4, 0l, null);
-    }
-
-    @Test(expected = UnsupportedOperationException.class)
-    public void testProducerSendDestinationMessageCompletionListener() throws JMSException {
-         messageProducer.send(session.createQueue("AMQ.TEST"), null, null);
-    }
-
-    @Test(expected = UnsupportedOperationException.class)
-    public void testProducerSendDestinationMessageQosParamsCompletionListener() throws JMSException {
-         messageProducer.send(session.createQueue("AMQ.TEST"), null, 1, 4, 0l, null);
-    }
-
-}
diff --git a/activemq-client/pom.xml b/activemq-client/pom.xml
index dc10320..96f9880 100644
--- a/activemq-client/pom.xml
+++ b/activemq-client/pom.xml
@@ -103,9 +103,9 @@
             <link>http://junit.sourceforge.net/javadoc/</link>
           </links>
           <stylesheetfile>${basedir}/../etc/css/stylesheet.css</stylesheetfile>
-          <linksource>true</linksource>
           <maxmemory>256m</maxmemory>
           <source>${source-version}</source>
+          <noindex>true</noindex>
           <groups>
             <group>
               <title>JMS Client</title>
diff --git a/activemq-client/src/main/java/org/apache/activemq/ActiveMQConnectionFactory.java b/activemq-client/src/main/java/org/apache/activemq/ActiveMQConnectionFactory.java
index 461dbe1..2324448 100644
--- a/activemq-client/src/main/java/org/apache/activemq/ActiveMQConnectionFactory.java
+++ b/activemq-client/src/main/java/org/apache/activemq/ActiveMQConnectionFactory.java
@@ -292,7 +292,11 @@
      */
     @Override
     public JMSContext createContext() {
-        throw new UnsupportedOperationException("createContext() is not supported");
+        try {
+            return new ActiveMQContext(createActiveMQConnection());
+        } catch (JMSException e) {
+            throw JMSExceptionSupport.convertToJMSRuntimeException(e);
+        }
     }
 
     /**
@@ -300,7 +304,11 @@
      */
     @Override
     public JMSContext createContext(String userName, String password) {
-        throw new UnsupportedOperationException("createContext() is not supported");
+        try {
+            return new ActiveMQContext(createActiveMQConnection(userName, password));
+        } catch (JMSException e) {
+            throw JMSExceptionSupport.convertToJMSRuntimeException(e);
+        }
     }
 
     /**
@@ -308,7 +316,11 @@
      */
     @Override
     public JMSContext createContext(String userName, String password, int sessionMode) {
-        throw new UnsupportedOperationException("createContext() is not supported");
+        try {
+            return new ActiveMQContext(createActiveMQConnection(userName, password), sessionMode);
+        } catch (JMSException e) {
+            throw JMSExceptionSupport.convertToJMSRuntimeException(e);
+        }
     }
 
     /**
diff --git a/activemq-client/src/main/java/org/apache/activemq/ActiveMQConsumer.java b/activemq-client/src/main/java/org/apache/activemq/ActiveMQConsumer.java
new file mode 100644
index 0000000..409506d
--- /dev/null
+++ b/activemq-client/src/main/java/org/apache/activemq/ActiveMQConsumer.java
@@ -0,0 +1,116 @@
+/**
+ * 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;
+
+import javax.jms.JMSConsumer;
+import javax.jms.JMSException;
+import javax.jms.JMSRuntimeException;
+import javax.jms.Message;
+import javax.jms.MessageConsumer;
+import javax.jms.MessageListener;
+
+import org.apache.activemq.util.JMSExceptionSupport;
+
+public class ActiveMQConsumer implements JMSConsumer {
+    
+    private final ActiveMQContext activemqContext;
+    private final MessageConsumer activemqMessageConsumer;
+
+    ActiveMQConsumer(ActiveMQContext activemqContext, MessageConsumer activemqMessageConsumer) {
+        this.activemqContext = activemqContext;
+        this.activemqMessageConsumer = activemqMessageConsumer;
+    }
+
+    @Override
+    public String getMessageSelector() {
+        try {
+            return activemqMessageConsumer.getMessageSelector();
+        } catch (JMSException e) {
+            throw JMSExceptionSupport.convertToJMSRuntimeException(e);
+        }
+    }
+
+    @Override
+    public MessageListener getMessageListener() throws JMSRuntimeException {
+        try {
+            return activemqMessageConsumer.getMessageListener();
+        } catch (JMSException e) {
+            throw JMSExceptionSupport.convertToJMSRuntimeException(e);
+        }
+    }
+
+    @Override
+    public void setMessageListener(MessageListener listener) throws JMSRuntimeException {
+        try {
+            activemqMessageConsumer.setMessageListener(listener);
+        } catch (JMSException e) {
+            throw JMSExceptionSupport.convertToJMSRuntimeException(e);
+        }
+    }
+
+    @Override
+    public Message receive() {
+        try {
+            return activemqMessageConsumer.receive();
+        } catch (JMSException e) {
+            throw JMSExceptionSupport.convertToJMSRuntimeException(e);
+        }
+    }
+
+    @Override
+    public Message receive(long timeout) {
+        try {
+            return activemqMessageConsumer.receive(timeout);
+        } catch (JMSException e) {
+            throw JMSExceptionSupport.convertToJMSRuntimeException(e);
+        }
+    }
+
+    @Override
+    public Message receiveNoWait() {
+        try {
+            return activemqMessageConsumer.receiveNoWait();
+        } catch (JMSException e) {
+            throw JMSExceptionSupport.convertToJMSRuntimeException(e);
+        }
+    }
+
+    @Override
+    public void close() {
+        try {
+            activemqMessageConsumer.close();
+        } catch (JMSException e) {
+            throw JMSExceptionSupport.convertToJMSRuntimeException(e);
+        }        
+    }
+
+    @Override
+    public <T> T receiveBody(Class<T> c) {
+        throw new UnsupportedOperationException("receiveBody(Class<T>) is not supported");
+    }
+
+    @Override
+    public <T> T receiveBody(Class<T> c, long timeout) {
+        throw new UnsupportedOperationException("receiveBody(Class<T>, long) is not supported");
+    }
+
+    @Override
+    public <T> T receiveBodyNoWait(Class<T> c) {
+        throw new UnsupportedOperationException("receiveBodyNoWait(Class<T>) is not supported");
+    }
+
+}
diff --git a/activemq-client/src/main/java/org/apache/activemq/ActiveMQContext.java b/activemq-client/src/main/java/org/apache/activemq/ActiveMQContext.java
new file mode 100644
index 0000000..4966d09
--- /dev/null
+++ b/activemq-client/src/main/java/org/apache/activemq/ActiveMQContext.java
@@ -0,0 +1,560 @@
+/**
+ * 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;
+
+import java.io.Serializable;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicLong;
+
+import javax.jms.BytesMessage;
+import javax.jms.ConnectionMetaData;
+import javax.jms.Destination;
+import javax.jms.ExceptionListener;
+import javax.jms.IllegalStateRuntimeException;
+import javax.jms.JMSConsumer;
+import javax.jms.JMSContext;
+import javax.jms.JMSException;
+import javax.jms.JMSProducer;
+import javax.jms.JMSRuntimeException;
+import javax.jms.MapMessage;
+import javax.jms.Message;
+import javax.jms.ObjectMessage;
+import javax.jms.Queue;
+import javax.jms.QueueBrowser;
+import javax.jms.Session;
+import javax.jms.StreamMessage;
+import javax.jms.TemporaryQueue;
+import javax.jms.TemporaryTopic;
+import javax.jms.TextMessage;
+import javax.jms.Topic;
+
+import org.apache.activemq.util.JMSExceptionSupport;
+
+/**
+ * In terms of the JMS 1.1 API a JMSContext should be thought of as 
+ * representing both a Connection and a Session. Although the simplified 
+ * API removes the need for applications to use those objects, the concepts 
+ * of connection and session remain important. A connection represents a 
+ * physical link to the JMS server and a session represents a 
+ * single-threaded context for sending and receiving messages. 
+ *
+ *
+ * @see javax.jms.JMSContext
+ */
+
+public class ActiveMQContext implements JMSContext {
+
+    private static final boolean DEFAULT_AUTO_START = true;
+
+    private final ActiveMQConnection activemqConnection;
+    private final AtomicLong connectionCounter;
+    private ActiveMQSession activemqSession = null;
+
+    // Configuration
+    private boolean autoStart = DEFAULT_AUTO_START;
+    private final int sessionMode;
+
+    // State
+    private boolean closeInvoked = false;
+    private final AtomicBoolean startInvoked = new AtomicBoolean(false);
+    private ActiveMQMessageProducer activemqMessageProducer = null;
+
+    ActiveMQContext(final ActiveMQConnection activemqConnection) {
+        this.activemqConnection = activemqConnection;
+        this.sessionMode = AUTO_ACKNOWLEDGE;
+        this.connectionCounter = new AtomicLong(1l);
+    }
+
+    ActiveMQContext(final ActiveMQConnection activemqConnection, final int sessionMode) {
+        this.activemqConnection = activemqConnection;
+        this.sessionMode = sessionMode;
+        this.connectionCounter = new AtomicLong(1l);
+    }
+
+    private ActiveMQContext(final ActiveMQConnection activemqConnection, final int sessionMode, final AtomicLong connectionCounter) {
+        this.activemqConnection = activemqConnection;
+        this.sessionMode = sessionMode;
+        this.connectionCounter = connectionCounter;
+    }
+
+    @Override
+    public JMSContext createContext(int sessionMode) {
+        if(connectionCounter.get() == 0l) {
+            throw new JMSRuntimeException("Context already closed");
+        }
+
+        connectionCounter.incrementAndGet();
+        return new ActiveMQContext(activemqConnection, sessionMode, connectionCounter);
+    }
+
+    @Override
+    public JMSProducer createProducer() {
+        return new ActiveMQProducer(this, getCreatedActiveMQMessageProducer());
+    }
+
+    @Override
+    public String getClientID() {
+        try {
+            return this.activemqConnection.getClientID();
+        } catch (JMSException e) {
+            throw JMSExceptionSupport.convertToJMSRuntimeException(e);
+        }
+    }
+
+    @Override
+    public void setClientID(String clientID) {
+        try {
+            this.activemqConnection.setClientID(clientID);
+        } catch (JMSException e) {
+            throw JMSExceptionSupport.convertToJMSRuntimeException(e);
+        }
+    }
+
+    @Override
+    public ConnectionMetaData getMetaData() {
+        checkContextState();
+        try {
+            return this.activemqConnection.getMetaData();
+        } catch (JMSException e) {
+                throw JMSExceptionSupport.convertToJMSRuntimeException(e);
+        }
+    }
+
+    @Override
+    public ExceptionListener getExceptionListener() {
+        checkContextState();
+        try {
+            return this.activemqConnection.getExceptionListener();
+        } catch (JMSException e) {
+            throw JMSExceptionSupport.convertToJMSRuntimeException(e);
+        }
+    }
+
+    @Override
+    public void setExceptionListener(ExceptionListener listener) {
+        checkContextState();
+        try {
+            this.activemqConnection.setExceptionListener(listener);
+        } catch (JMSException e) {
+            throw JMSExceptionSupport.convertToJMSRuntimeException(e);
+        }
+    }
+
+    @Override
+    public void start() {
+        checkContextState();
+        try {
+            if(startInvoked.compareAndSet(false, true)) {
+                this.activemqConnection.start();
+            }
+        } catch (JMSException e) {
+            throw JMSExceptionSupport.convertToJMSRuntimeException(e);
+        }
+    }
+
+    @Override
+    public void stop() {
+        checkContextState();
+        try {
+            if(startInvoked.compareAndSet(true, false)) {
+                this.activemqConnection.stop();
+            }
+        } catch (JMSException e) {
+            throw JMSExceptionSupport.convertToJMSRuntimeException(e);
+        }
+    }
+
+    @Override
+    public void setAutoStart(boolean autoStart) {
+       this.autoStart = autoStart;
+    }
+
+    @Override
+    public boolean getAutoStart() {
+        return this.autoStart;
+    }
+
+    @Override
+    public synchronized void close() {
+        JMSRuntimeException firstException = null;
+
+        if(this.activemqMessageProducer != null) {
+            try {
+                this.activemqMessageProducer.close();
+            } catch (JMSException e) {
+                if(firstException == null) { 
+                    firstException = JMSExceptionSupport.convertToJMSRuntimeException(e);
+                }
+            }
+        }
+
+        if(this.activemqSession != null) {
+            try {
+               this.activemqSession.close();
+            } catch (JMSException e) {
+               if(firstException == null) {
+                   firstException = JMSExceptionSupport.convertToJMSRuntimeException(e);
+               }
+            }
+        }
+
+        if(connectionCounter.decrementAndGet() == 0) {
+            if(this.activemqConnection != null) {
+                try {
+                    closeInvoked = true;
+                    this.activemqConnection.close();
+                } catch (JMSException e) {
+                    if(firstException == null) {
+                        firstException = JMSExceptionSupport.convertToJMSRuntimeException(e);
+                    }
+                }
+            }
+        }
+
+        if(firstException != null) {
+            throw firstException;
+        }
+    }
+
+    @Override
+    public BytesMessage createBytesMessage() {
+        checkContextState();
+        try {
+            return activemqSession.createBytesMessage();
+        } catch (JMSException e) {
+            throw JMSExceptionSupport.convertToJMSRuntimeException(e);
+        }
+     }
+
+    @Override
+    public MapMessage createMapMessage() {
+        checkContextState();
+        try {
+            return activemqSession.createMapMessage();
+        } catch (JMSException e) {
+            throw JMSExceptionSupport.convertToJMSRuntimeException(e);
+        }
+    }
+
+    @Override
+    public Message createMessage() {
+        checkContextState();
+        try {
+            return activemqSession.createMessage();
+        } catch (JMSException e) {
+            throw JMSExceptionSupport.convertToJMSRuntimeException(e);
+        }
+    }
+
+    @Override
+    public ObjectMessage createObjectMessage() {
+        checkContextState();
+        try {
+            return activemqSession.createObjectMessage();
+        } catch (JMSException e) {
+            throw JMSExceptionSupport.convertToJMSRuntimeException(e);
+        }
+    }
+
+    @Override
+    public ObjectMessage createObjectMessage(Serializable object) {
+        checkContextState();
+        try {
+            return activemqSession.createObjectMessage(object);
+        } catch (JMSException e) {
+            throw JMSExceptionSupport.convertToJMSRuntimeException(e);
+        }
+    }
+
+    @Override
+    public StreamMessage createStreamMessage() {
+        checkContextState();
+        try {
+            return activemqSession.createStreamMessage();
+        } catch (JMSException e) {
+            throw JMSExceptionSupport.convertToJMSRuntimeException(e);
+        }
+    }
+
+    @Override
+    public TextMessage createTextMessage() {
+        checkContextState();
+        try {
+            return activemqSession.createTextMessage();
+        } catch (JMSException e) {
+            throw JMSExceptionSupport.convertToJMSRuntimeException(e);
+        }
+    }
+
+    @Override
+    public TextMessage createTextMessage(String text) {
+        checkContextState();
+        try {
+            return activemqSession.createTextMessage(text);
+        } catch (JMSException e) {
+            throw JMSExceptionSupport.convertToJMSRuntimeException(e);
+        }
+    }
+
+    @Override
+    public boolean getTransacted() {
+        checkContextState();
+        try {
+            return activemqSession.getTransacted();
+        } catch (JMSException e) {
+            throw JMSExceptionSupport.convertToJMSRuntimeException(e);
+        }
+    }
+
+    @Override
+    public int getSessionMode() {
+        return this.sessionMode;
+    }
+
+    @Override
+    public void commit() {
+        checkContextState();
+        try {
+            activemqSession.commit();
+        } catch (JMSException e) {
+            throw JMSExceptionSupport.convertToJMSRuntimeException(e);
+        }
+    }
+
+    @Override
+    public void rollback() {
+        checkContextState();
+        try {
+            activemqSession.rollback();
+        } catch (JMSException e) {
+            throw JMSExceptionSupport.convertToJMSRuntimeException(e);
+        }
+    }
+
+    @Override
+    public void recover() {
+        checkContextState();
+        try {
+            activemqSession.recover();
+        } catch (JMSException e) {
+            throw JMSExceptionSupport.convertToJMSRuntimeException(e);
+        }
+    }
+
+    @Override
+    public JMSConsumer createConsumer(Destination destination) {
+        checkContextState();
+        try {
+            if(getAutoStart()) {
+                start();
+            }
+            return new ActiveMQConsumer(this, activemqSession.createConsumer(destination));
+        } catch (JMSException e) {
+            throw JMSExceptionSupport.convertToJMSRuntimeException(e);
+        }
+    }
+
+    @Override
+    public JMSConsumer createConsumer(Destination destination, String messageSelector) {
+        checkContextState();
+        try {
+            if(getAutoStart()) {
+                start();
+            }
+            return new ActiveMQConsumer(this, activemqSession.createConsumer(destination, messageSelector));
+        } catch (JMSException e) {
+            throw JMSExceptionSupport.convertToJMSRuntimeException(e);
+        }
+    }
+
+    @Override
+    public JMSConsumer createConsumer(Destination destination, String messageSelector, boolean noLocal) {
+        checkContextState();
+        try {
+            if(getAutoStart()) {
+                start();
+            }
+            return new ActiveMQConsumer(this, activemqSession.createConsumer(destination, messageSelector, noLocal));
+        } catch (JMSException e) {
+            throw JMSExceptionSupport.convertToJMSRuntimeException(e);
+        }
+    }
+
+    @Override
+    public Queue createQueue(String queueName) {
+        checkContextState();
+        try {
+            return activemqSession.createQueue(queueName);
+        } catch (JMSException e) {
+            throw JMSExceptionSupport.convertToJMSRuntimeException(e);
+        }
+    }
+
+    @Override
+    public Topic createTopic(String topicName) {
+        checkContextState();
+        try {
+            return activemqSession.createTopic(topicName);
+        } catch (JMSException e) {
+            throw JMSExceptionSupport.convertToJMSRuntimeException(e);
+        }
+    }
+
+    @Override
+    public JMSConsumer createDurableConsumer(Topic topic, String name) {
+        checkContextState();
+        try {
+            if(getAutoStart()) {
+                start();
+            }
+            return new ActiveMQConsumer(this, activemqSession.createDurableConsumer(topic, name));
+        } catch (JMSException e) {
+            throw JMSExceptionSupport.convertToJMSRuntimeException(e);
+        }
+    }
+
+    @Override
+    public JMSConsumer createDurableConsumer(Topic topic, String name, String messageSelector, boolean noLocal) {
+        checkContextState();
+        try {
+            if(getAutoStart()) {
+                start();
+            }
+            return new ActiveMQConsumer(this, activemqSession.createDurableConsumer(topic, name, messageSelector, noLocal));
+        } catch (JMSException e) {
+            throw JMSExceptionSupport.convertToJMSRuntimeException(e);
+        }
+    }
+
+    @Override
+    public JMSConsumer createSharedDurableConsumer(Topic topic, String name) {
+        throw new UnsupportedOperationException("createSharedDurableConsumer(topic, name) is not supported");
+    }
+
+    @Override
+    public JMSConsumer createSharedDurableConsumer(Topic topic, String name, String messageSelector) {
+        throw new UnsupportedOperationException("createDurableConsumer(topic, name, messageSelector) is not supported");
+    }
+
+    @Override
+    public JMSConsumer createSharedConsumer(Topic topic, String sharedSubscriptionName) {
+        throw new UnsupportedOperationException("createSharedConsumer(topic, sharedSubscriptionName) is not supported");
+    }
+
+    @Override
+    public JMSConsumer createSharedConsumer(Topic topic, String sharedSubscriptionName, String messageSelector) {
+        throw new UnsupportedOperationException("createSharedConsumer(topic, sharedSubscriptionName, messageSelector) is not supported");
+    }
+
+    @Override
+    public QueueBrowser createBrowser(Queue queue) {
+        checkContextState();
+        try {
+            if(getAutoStart()) {
+                start();
+            }
+            return activemqSession.createBrowser(queue);
+        } catch (JMSException e) {
+            throw JMSExceptionSupport.convertToJMSRuntimeException(e);
+        }
+    }
+
+    @Override
+    public QueueBrowser createBrowser(Queue queue, String messageSelector) {
+        checkContextState();
+        try {
+            if(getAutoStart()) {
+                start();
+            }
+            return activemqSession.createBrowser(queue, messageSelector);
+        } catch (JMSException e) {
+            throw JMSExceptionSupport.convertToJMSRuntimeException(e);
+        }
+    }
+
+    @Override
+    public TemporaryQueue createTemporaryQueue() {
+        checkContextState();
+        try {
+            return activemqSession.createTemporaryQueue();
+        } catch (JMSException e) {
+            throw JMSExceptionSupport.convertToJMSRuntimeException(e);
+        }
+    }
+
+    @Override
+    public TemporaryTopic createTemporaryTopic() {
+        checkContextState();
+        try {
+            return activemqSession.createTemporaryTopic();
+        } catch (JMSException e) {
+            throw JMSExceptionSupport.convertToJMSRuntimeException(e);
+        }
+    }
+
+    @Override
+    public void unsubscribe(String name) {
+        checkContextState();
+        try {
+            activemqSession.unsubscribe(name);
+        } catch (JMSException e) {
+            throw JMSExceptionSupport.convertToJMSRuntimeException(e);
+        }
+    }
+
+    @Override
+    public void acknowledge() {
+        checkContextState();
+        try {
+            activemqSession.acknowledge();
+        } catch (JMSException e) {
+            throw JMSExceptionSupport.convertToJMSRuntimeException(e);
+        }
+    }
+
+    private void checkContextState() {
+        if (activemqConnection == null) {
+            throw new JMSRuntimeException("Connection not available");
+        }
+
+        if (activemqSession == null) {
+            if (closeInvoked) {
+                throw new IllegalStateRuntimeException("Context is closed");
+            }
+            try {
+            	Session jmsSession = activemqConnection.createSession(SESSION_TRANSACTED == sessionMode, sessionMode);
+            	activemqSession = ActiveMQSession.class.cast(jmsSession);
+            } catch (JMSException e) {
+                throw JMSExceptionSupport.convertToJMSRuntimeException(e);
+            }
+         }
+    }
+
+    private ActiveMQMessageProducer getCreatedActiveMQMessageProducer() {
+        checkContextState();
+        
+        if (this.activemqMessageProducer == null) {
+            try {
+                this.activemqMessageProducer = new ActiveMQMessageProducer(activemqSession, activemqSession.getNextProducerId(), null, activemqConnection.getSendTimeout());
+            } catch (JMSException e) {
+                throw JMSExceptionSupport.convertToJMSRuntimeException(e);
+            }
+         }
+        return this.activemqMessageProducer;
+    }
+
+}
diff --git a/activemq-client/src/main/java/org/apache/activemq/ActiveMQMessageProducer.java b/activemq-client/src/main/java/org/apache/activemq/ActiveMQMessageProducer.java
index 9fd2439..9e9ebd4 100644
--- a/activemq-client/src/main/java/org/apache/activemq/ActiveMQMessageProducer.java
+++ b/activemq-client/src/main/java/org/apache/activemq/ActiveMQMessageProducer.java
@@ -223,7 +223,7 @@
     public void send(Destination destination, Message message, int deliveryMode, int priority, long timeToLive) throws JMSException {
         this.send(destination, message, deliveryMode, priority, timeToLive, (AsyncCallback)null);
     }
-    
+
     /**
      *
      * @param message the message to send
@@ -240,7 +240,6 @@
     @Override
     public void send(Message message, CompletionListener completionListener) throws JMSException {
         throw new UnsupportedOperationException("send(Message, CompletionListener) is not supported");
-
     }
 
     @Override
@@ -287,6 +286,11 @@
     }
 
     public void send(Destination destination, Message message, int deliveryMode, int priority, long timeToLive, AsyncCallback onComplete) throws JMSException {
+        this.send(destination, message, deliveryMode, priority, timeToLive, getDisableMessageID(), getDisableMessageTimestamp(), onComplete);
+    }
+
+    public void send(Destination destination, Message message, int deliveryMode, int priority, long timeToLive, boolean disableMessageID, boolean disableMessageTimestamp, AsyncCallback onComplete) throws JMSException {
+
         checkClosed();
         if (destination == null) {
             if (info.getDestination() == null) {
@@ -322,7 +326,7 @@
             }
         }
 
-        this.session.send(this, dest, message, deliveryMode, priority, timeToLive, producerWindow, sendTimeout, onComplete);
+        this.session.send(this, dest, message, deliveryMode, priority, timeToLive, disableMessageID, disableMessageTimestamp, producerWindow, sendTimeout, onComplete);
 
         stats.onMessage();
     }
diff --git a/activemq-client/src/main/java/org/apache/activemq/ActiveMQMessageProducerSupport.java b/activemq-client/src/main/java/org/apache/activemq/ActiveMQMessageProducerSupport.java
index b7dd6cf..accc7e5 100644
--- a/activemq-client/src/main/java/org/apache/activemq/ActiveMQMessageProducerSupport.java
+++ b/activemq-client/src/main/java/org/apache/activemq/ActiveMQMessageProducerSupport.java
@@ -16,11 +16,15 @@
  */
 package org.apache.activemq;
 
+import java.util.Set;
+
 import javax.jms.DeliveryMode;
 import javax.jms.Destination;
 import javax.jms.IllegalStateException;
+import javax.jms.IllegalStateRuntimeException;
 import javax.jms.JMSException;
 import javax.jms.Message;
+import javax.jms.MessageFormatRuntimeException;
 import javax.jms.MessageProducer;
 
 /**
@@ -82,13 +86,13 @@
      * <P>
      * Message IDs are enabled by default.
      *
-     * @param value indicates if message IDs are disabled
+     * @param disableMessageID indicates if message IDs are disabled
      * @throws javax.jms.JMSException if the JMS provider fails to close the producer due to
      *                      some internal error.
      */
-    public void setDisableMessageID(boolean value) throws JMSException {
+    public void setDisableMessageID(boolean disableMessageID) throws JMSException {
         checkClosed();
-        this.disableMessageID = value;
+        this.disableMessageID = disableMessageID;
     }
 
     /**
@@ -118,13 +122,13 @@
      * <P>
      * Message timestamps are enabled by default.
      *
-     * @param value indicates if message timestamps are disabled
+     * @param disableMessageTimestamp indicates if message timestamps are disabled
      * @throws javax.jms.JMSException if the JMS provider fails to close the producer due to
      *                      some internal error.
      */
-    public void setDisableMessageTimestamp(boolean value) throws JMSException {
+    public void setDisableMessageTimestamp(boolean disableMessageTimestamp) throws JMSException {
         checkClosed();
-        this.disableMessageTimestamp = value;
+        this.disableMessageTimestamp = disableMessageTimestamp;
     }
 
     /**
@@ -345,4 +349,46 @@
     public void setSendTimeout(int sendTimeout) {
         this.sendTimeout = sendTimeout;
     }
+
+    // See JMS 2.0 spec sections 3.5.1 and 3.8.1.1
+    public static final Set<String> JMS_PROPERTY_NAMES_DISALLOWED = Set.of("JMSDeliveryMode", "JMSPriority", "JMSMessageID", "JMSTimestamp", "JMSCorrelationID", "JMSType", "NULL", "TRUE", "FALSE", "NOT", "AND", "OR", "BETWEEN", "LIKE", "IN", "IS", "ESCAPE");
+
+    public static void validateValidPropertyName(String propertyName) throws IllegalStateRuntimeException {
+        if(propertyName == null || propertyName.length() == 0) {
+            throw new IllegalArgumentException("Invalid JMS property name must not be null or empty");
+        }
+
+        if(JMS_PROPERTY_NAMES_DISALLOWED.contains(propertyName)) {
+            throw new IllegalArgumentException("Invalid JMS property: " + propertyName + " name is in disallowed list");
+        }
+
+        char first = propertyName.charAt(0);
+        if(!(Character.isJavaIdentifierStart(first))) {
+            throw new IllegalArgumentException("Invalid JMS property: " + propertyName + " name starts with invalid character: " + first);
+        }
+
+        for (int i=1; i < propertyName.length(); i++) {
+            char c = propertyName.charAt(i);
+            if (!(Character.isJavaIdentifierPart(c))) {
+                throw new IllegalArgumentException("Invalid JMS property: " + propertyName + " name contains invalid character: " + c);
+            }
+        }
+    }
+
+    public static void validateValidPropertyValue(String propertyName, Object propertyValue) throws IllegalStateRuntimeException {
+        boolean invalid = true;
+
+        if(propertyValue == null || propertyValue instanceof String ||
+           propertyValue instanceof Integer || propertyValue instanceof Short ||
+           propertyValue instanceof Float || propertyValue instanceof Long ||
+           propertyValue instanceof Boolean || propertyValue instanceof Byte ||
+           propertyValue instanceof Character || propertyValue instanceof Double) {
+           invalid = false;
+        }
+
+        if(invalid) {
+            throw new MessageFormatRuntimeException("Invalid JMS property: " + propertyName + " value class: " + propertyValue.getClass().getName() + " is not permitted by specification");
+        }
+    }
+
 }
diff --git a/activemq-client/src/main/java/org/apache/activemq/ActiveMQMessageTransformation.java b/activemq-client/src/main/java/org/apache/activemq/ActiveMQMessageTransformation.java
index 2d6094e..f456b5f 100644
--- a/activemq-client/src/main/java/org/apache/activemq/ActiveMQMessageTransformation.java
+++ b/activemq-client/src/main/java/org/apache/activemq/ActiveMQMessageTransformation.java
@@ -16,6 +16,8 @@
  */
 package org.apache.activemq;
 
+import java.lang.reflect.Method;
+import java.lang.reflect.Modifier;
 import java.net.MalformedURLException;
 import java.util.Enumeration;
 
@@ -179,6 +181,7 @@
         toMessage.setJMSReplyTo(transformDestination(fromMessage.getJMSReplyTo()));
         toMessage.setJMSDestination(transformDestination(fromMessage.getJMSDestination()));
         toMessage.setJMSDeliveryMode(fromMessage.getJMSDeliveryMode());
+        toMessage.setJMSDeliveryTime(getFromMessageDeliveryTime(fromMessage)); // TODO: AMQ-8500 DeliveryTime support ref: ActiveMQSession#send
         toMessage.setJMSRedelivered(fromMessage.getJMSRedelivered());
         toMessage.setJMSType(fromMessage.getJMSType());
         toMessage.setJMSExpiration(fromMessage.getJMSExpiration());
@@ -193,4 +196,23 @@
             toMessage.setObjectProperty(name, obj);
         }
     }
+
+    private static long getFromMessageDeliveryTime(Message fromMessage) throws JMSException {
+        Method deliveryTimeGetMethod = null;
+        try {
+            Class<?> clazz = fromMessage.getClass();
+            Method method = clazz.getMethod("getJMSDeliveryTime");
+            if (!Modifier.isAbstract(method.getModifiers())) {
+                deliveryTimeGetMethod = method;
+            }
+        } catch (NoSuchMethodException e) {
+            // We fallback to JMSTimestamp for jms v1.x
+        }
+
+        if (deliveryTimeGetMethod != null) {
+            return fromMessage.getJMSDeliveryTime();
+        }
+
+        return fromMessage.getJMSTimestamp();
+    }
 }
diff --git a/activemq-client/src/main/java/org/apache/activemq/ActiveMQProducer.java b/activemq-client/src/main/java/org/apache/activemq/ActiveMQProducer.java
new file mode 100644
index 0000000..7a6c621
--- /dev/null
+++ b/activemq-client/src/main/java/org/apache/activemq/ActiveMQProducer.java
@@ -0,0 +1,545 @@
+/**
+ * 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;
+
+import java.io.Serializable;
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Set;
+
+import javax.jms.BytesMessage;
+import javax.jms.CompletionListener;
+import javax.jms.DeliveryMode;
+import javax.jms.Destination;
+import javax.jms.IllegalStateRuntimeException;
+import javax.jms.JMSException;
+import javax.jms.JMSProducer;
+import javax.jms.MapMessage;
+import javax.jms.Message;
+import javax.jms.MessageFormatRuntimeException;
+import javax.jms.ObjectMessage;
+import javax.jms.TextMessage;
+
+import org.apache.activemq.util.JMSExceptionSupport;
+import org.apache.activemq.util.TypeConversionSupport;
+
+public class ActiveMQProducer implements JMSProducer {
+
+    private final ActiveMQContext activemqContext;
+    private final ActiveMQMessageProducer activemqMessageProducer;
+
+    // QoS override of defaults on a per-JMSProducer instance basis
+    private String correlationId = null;
+    private byte[] correlationIdBytes = null;
+    private Integer deliveryMode = null;
+    private Boolean disableMessageID = false;
+    private Boolean disableMessageTimestamp = false;
+    private Integer priority = null;
+    private Destination replyTo = null;
+    private Long timeToLive = null;
+    private String type = null;
+
+    // Properties applied to all messages on a per-JMS producer instance basis
+    private Map<String, Object> messageProperties = null;
+
+    ActiveMQProducer(ActiveMQContext activemqContext, ActiveMQMessageProducer activemqMessageProducer) {
+        this.activemqContext = activemqContext;
+        this.activemqMessageProducer = activemqMessageProducer;
+    }
+
+    @Override
+    public JMSProducer send(Destination destination, Message message) {
+        try {
+            if(this.correlationId != null) {
+                message.setJMSCorrelationID(this.correlationId);
+            }
+
+            if(this.correlationIdBytes != null) {
+                message.setJMSCorrelationIDAsBytes(this.correlationIdBytes);
+            }
+
+            if(this.replyTo != null) {
+                message.setJMSReplyTo(this.replyTo);
+            }
+
+            if(this.type != null) {
+                message.setJMSType(this.type);
+            }
+
+            if(messageProperties != null && !messageProperties.isEmpty()) {
+                for(Map.Entry<String, Object> propertyEntry : messageProperties.entrySet()) {
+                    message.setObjectProperty(propertyEntry.getKey(), propertyEntry.getValue());
+                }
+            }
+
+            activemqMessageProducer.send(destination, message, getDeliveryMode(), getPriority(), getTimeToLive(), getDisableMessageID(), getDisableMessageTimestamp(), null);
+        } catch (JMSException e) {
+            throw JMSExceptionSupport.convertToJMSRuntimeException(e);
+        }
+        return this;
+    }
+
+    @Override
+    public JMSProducer send(Destination destination, String body) {
+        TextMessage textMessage = activemqContext.createTextMessage(body);
+        send(destination, textMessage);
+        return this;
+    }
+
+    @Override
+    public JMSProducer send(Destination destination, Map<String, Object> body) {
+        MapMessage mapMessage = activemqContext.createMapMessage();
+
+        if (body != null) {
+            try {
+               for (Map.Entry<String, Object> mapEntry : body.entrySet()) {
+                  final String key = mapEntry.getKey();
+                  final Object value = mapEntry.getValue();
+                  final Class<?> valueObject = value.getClass();
+                  if (String.class.isAssignableFrom(valueObject)) {
+                      mapMessage.setString(key, String.class.cast(value));
+                  } else if (Integer.class.isAssignableFrom(valueObject)) {
+                      mapMessage.setInt(key, Integer.class.cast(value));
+                  } else if (Long.class.isAssignableFrom(valueObject)) {
+                      mapMessage.setLong(key, Long.class.cast(value));
+                  } else if (Double.class.isAssignableFrom(valueObject)) {
+                      mapMessage.setDouble(key, Double.class.cast(value));
+                  } else if (Boolean.class.isAssignableFrom(valueObject)) {
+                      mapMessage.setBoolean(key, Boolean.class.cast(value));
+                  } else if (Character.class.isAssignableFrom(valueObject)) {
+                      mapMessage.setChar(key, Character.class.cast(value));
+                  } else if (Short.class.isAssignableFrom(valueObject)) {
+                      mapMessage.setShort(key, Short.class.cast(value));
+                  } else if (Float.class.isAssignableFrom(valueObject)) {
+                      mapMessage.setFloat(key, Float.class.cast(value));
+                  } else if (Byte.class.isAssignableFrom(valueObject)) {
+                      mapMessage.setByte(key, Byte.class.cast(value));
+                  } else if (byte[].class.isAssignableFrom(valueObject)) {
+                      byte[] array = byte[].class.cast(value);
+                      mapMessage.setBytes(key, array, 0, array.length);
+                  } else {
+                      mapMessage.setObject(key, value);
+                  }
+               }
+            } catch (JMSException e) {
+               throw new MessageFormatRuntimeException(e.getMessage());
+            }
+         }
+         send(destination, mapMessage);
+         return this;
+    }
+
+    @Override
+    public JMSProducer send(Destination destination, byte[] body) {
+        BytesMessage bytesMessage = activemqContext.createBytesMessage();
+
+        try {
+            if(body != null) {
+                bytesMessage.writeBytes(body);
+            }
+            send(destination, bytesMessage);
+        } catch (JMSException e) {
+            throw JMSExceptionSupport.convertToJMSRuntimeException(e);
+        }
+        return this;
+    }
+
+    @Override
+    public JMSProducer send(Destination destination, Serializable body) {
+        ObjectMessage objectMessage = activemqContext.createObjectMessage(body);
+        send(destination, objectMessage);
+        return this;
+    }
+
+    @Override
+    public JMSProducer setDisableMessageID(boolean disableMessageID) {
+        this.disableMessageID = disableMessageID;
+        return this;
+    }
+
+    @Override
+    public boolean getDisableMessageID() {
+        if(this.disableMessageID != null) {
+            return this.disableMessageID;
+        }
+
+        try {
+            return this.activemqMessageProducer.getDisableMessageID();
+        } catch (JMSException e) {
+            throw JMSExceptionSupport.convertToJMSRuntimeException(e);
+        }
+    }
+
+    @Override
+    public JMSProducer setDisableMessageTimestamp(boolean disableMessageTimestamp) {
+        this.disableMessageTimestamp = disableMessageTimestamp;
+        return this;
+    }
+
+    @Override
+    public boolean getDisableMessageTimestamp() {
+        if(this.disableMessageTimestamp != null) {
+            return this.disableMessageTimestamp;
+        }
+
+        try {
+            return this.activemqMessageProducer.getDisableMessageTimestamp();
+        } catch (JMSException e) {
+            throw JMSExceptionSupport.convertToJMSRuntimeException(e);
+        }
+    }
+
+    @Override
+    public JMSProducer setDeliveryMode(int deliveryMode) {
+        if (deliveryMode != DeliveryMode.PERSISTENT && deliveryMode != DeliveryMode.NON_PERSISTENT) {
+            throw new IllegalStateRuntimeException("unknown delivery mode: " + deliveryMode);
+        }
+        this.deliveryMode = deliveryMode;
+        return this;
+    }
+
+    @Override
+    public int getDeliveryMode() {
+        if(deliveryMode != null) {
+            return deliveryMode;
+        }
+
+        try {
+            return this.activemqMessageProducer.getDeliveryMode();
+        } catch (JMSException e) {
+            throw JMSExceptionSupport.convertToJMSRuntimeException(e);
+        }
+    }
+
+    @Override
+    public JMSProducer setPriority(int priority) {
+        if (priority < 0 || priority > 9) {
+            throw new IllegalStateRuntimeException("default priority must be a value between 0 and 9");
+        }
+        this.priority = priority;
+        return this;
+    }
+
+    @Override
+    public int getPriority() {
+        if(priority != null) {
+            return priority;
+        }
+
+        try {
+            return this.activemqMessageProducer.getPriority();
+        } catch (JMSException e) {
+            throw JMSExceptionSupport.convertToJMSRuntimeException(e);
+        }
+    }
+
+    @Override
+    public JMSProducer setTimeToLive(long timeToLive) {
+        this.timeToLive = timeToLive;
+        return this;
+    }
+
+    @Override
+    public long getTimeToLive() {
+        if(timeToLive != null) {
+            return timeToLive;
+        }
+
+        try {
+            return this.activemqMessageProducer.getTimeToLive();
+        } catch (JMSException e) {
+            throw JMSExceptionSupport.convertToJMSRuntimeException(e);
+        }
+    }
+
+    @Override
+    public JMSProducer setDeliveryDelay(long deliveryDelay) {
+        throw new UnsupportedOperationException("setDeliveryDelay(long) is not supported");
+    }
+
+    @Override
+    public long getDeliveryDelay() {
+        throw new UnsupportedOperationException("getDeliveryDelay() is not supported");
+    }
+
+    @Override
+    public JMSProducer setAsync(CompletionListener completionListener) {
+        throw new UnsupportedOperationException("setAsync(CompletionListener) is not supported");
+    }
+
+    @Override
+    public CompletionListener getAsync() {
+        throw new UnsupportedOperationException("getAsync() is not supported");
+    }
+
+    @Override
+    public JMSProducer setProperty(String name, boolean value) {
+        checkProperty(name, value);
+        getCreatedMessageProperties().put(name, value);
+        return this;
+    }
+
+    @Override
+    public JMSProducer setProperty(String name, byte value) {
+        checkProperty(name, value);
+        getCreatedMessageProperties().put(name, value);
+        return this;
+    }
+
+    @Override
+    public JMSProducer setProperty(String name, short value) {
+        checkProperty(name, value);
+        getCreatedMessageProperties().put(name, value);
+        return this;
+    }
+
+    @Override
+    public JMSProducer setProperty(String name, int value) {
+        checkProperty(name, value);
+        getCreatedMessageProperties().put(name, value);
+        return this;
+    }
+
+    @Override
+    public JMSProducer setProperty(String name, long value) {
+        checkProperty(name, value);
+        getCreatedMessageProperties().put(name, value);
+        return this;
+    }
+
+    @Override
+    public JMSProducer setProperty(String name, float value) {
+        checkProperty(name, value);
+        getCreatedMessageProperties().put(name, value);
+        return this;
+    }
+
+    @Override
+    public JMSProducer setProperty(String name, double value) {
+        checkProperty(name, value);
+        getCreatedMessageProperties().put(name, value);
+        return this;
+    }
+
+    @Override
+    public JMSProducer setProperty(String name, String value) {
+        checkProperty(name, value);
+        getCreatedMessageProperties().put(name, value);
+        return this;
+    }
+
+    @Override
+    public JMSProducer setProperty(String name, Object value) {
+        checkProperty(name, value);
+        getCreatedMessageProperties().put(name, value);
+        return this;
+    }
+
+    @Override
+    public JMSProducer clearProperties() {
+        getCreatedMessageProperties().clear();
+        return this;
+    }
+
+    @Override
+    public boolean propertyExists(String name) {
+        if(name == null || name.isEmpty()) {
+            return false;
+        }
+        return getCreatedMessageProperties().containsKey(name);
+    }
+
+    @Override
+    public boolean getBooleanProperty(String name) {
+        Object value = getCreatedMessageProperties().get(name);
+        if (value == null) {
+            throw new NullPointerException("property " + name + " was null");
+        }
+        Boolean rc = (Boolean)TypeConversionSupport.convert(value, Boolean.class);
+        if (rc == null) {
+            throw new MessageFormatRuntimeException("Property " + name + " was a " + value.getClass().getName() + " and cannot be read as a boolean");
+        }
+        return rc.booleanValue();
+    }
+
+    @Override
+    public byte getByteProperty(String name) {
+        Object value = getCreatedMessageProperties().get(name);
+        if (value == null) {
+            throw new NullPointerException("property " + name + " was null");
+        }
+        Byte rc = (Byte)TypeConversionSupport.convert(value, Byte.class);
+        if (rc == null) {
+            throw new MessageFormatRuntimeException("Property " + name + " was a " + value.getClass().getName() + " and cannot be read as a byte");
+        }
+        return rc.byteValue();
+    }
+
+    @Override
+    public short getShortProperty(String name) {
+        Object value = getCreatedMessageProperties().get(name);
+        if (value == null) {
+            throw new NullPointerException("property " + name + " was null");
+        }
+        Short rc = (Short)TypeConversionSupport.convert(value, Short.class);
+        if (rc == null) {
+            throw new MessageFormatRuntimeException("Property " + name + " was a " + value.getClass().getName() + " and cannot be read as a short");
+        }
+        return rc.shortValue();
+    }
+
+    @Override
+    public int getIntProperty(String name) {
+        Object value = getCreatedMessageProperties().get(name);
+        if (value == null) {
+            throw new NullPointerException("property " + name + " was null");
+        }
+        Integer rc = (Integer)TypeConversionSupport.convert(value, Integer.class);
+        if (rc == null) {
+            throw new MessageFormatRuntimeException("Property " + name + " was a " + value.getClass().getName() + " and cannot be read as an integer");
+        }
+        return rc.intValue();
+    }
+
+    @Override
+    public long getLongProperty(String name) {
+        Object value = getCreatedMessageProperties().get(name);
+        if (value == null) {
+            throw new NullPointerException("property " + name + " was null");
+        }
+        Long rc = (Long)TypeConversionSupport.convert(value, Long.class);
+        if (rc == null) {
+            throw new MessageFormatRuntimeException("Property " + name + " was a " + value.getClass().getName() + " and cannot be read as a long");
+        }
+        return rc.longValue();
+    }
+
+    @Override
+    public float getFloatProperty(String name) {
+        Object value = getCreatedMessageProperties().get(name);
+        if (value == null) {
+            throw new NullPointerException("property " + name + " was null");
+        }
+        Float rc = (Float)TypeConversionSupport.convert(value, Float.class);
+        if (rc == null) {
+            throw new MessageFormatRuntimeException("Property " + name + " was a " + value.getClass().getName() + " and cannot be read as a float");
+        }
+        return rc.floatValue();
+    }
+
+    @Override
+    public double getDoubleProperty(String name) {
+        Object value = getCreatedMessageProperties().get(name);
+        if (value == null) {
+            throw new NullPointerException("property " + name + " was null");
+        }
+        Double rc = (Double)TypeConversionSupport.convert(value, Double.class);
+        if (rc == null) {
+            throw new MessageFormatRuntimeException("Property " + name + " was a " + value.getClass().getName() + " and cannot be read as a double");
+        }
+        return rc.doubleValue();
+    }
+
+    @Override
+    public String getStringProperty(String name) {
+        Object value = getCreatedMessageProperties().get(name);
+        if (value == null) {
+            throw new NullPointerException("property " + name + " was null");
+        }
+        String rc = (String)TypeConversionSupport.convert(value, String.class);
+        if (rc == null) {
+            throw new MessageFormatRuntimeException("Property " + name + " was a " + value.getClass().getName() + " and cannot be read as a string");
+        }
+        return rc;
+    }
+
+    @Override
+    public Object getObjectProperty(String name) {
+        if (name == null) {
+            throw new NullPointerException("Property name cannot be null");
+        }
+
+        // TODO: Update PropertyExpression to handle converting message headers to properties for JMSProducer.
+        //PropertyExpression expression = new PropertyExpression(name);
+        //return expression.evaluate(this);
+        return getCreatedMessageProperties().get(name);
+    }
+
+    @Override
+    public Set<String> getPropertyNames() {
+        return getCreatedMessageProperties().keySet();
+    }
+
+    @Override
+    public JMSProducer setJMSCorrelationIDAsBytes(byte[] correlationID) {
+        if(correlationID != null) {
+            this.correlationIdBytes = Arrays.copyOf(correlationID, correlationID.length);
+        }
+        return this;
+    }
+
+    @Override
+    public byte[] getJMSCorrelationIDAsBytes() {
+        return this.correlationIdBytes;
+    }
+
+    @Override
+    public JMSProducer setJMSCorrelationID(String correlationID) {
+        this.correlationId = correlationID;
+        return this;
+    }
+
+    @Override
+    public String getJMSCorrelationID() {
+        return this.correlationId;
+    }
+
+    @Override
+    public JMSProducer setJMSType(String type) {
+        this.type = type;
+        return this;
+    }
+
+    @Override
+    public String getJMSType() {
+        return this.type;
+    }
+
+    @Override
+    public JMSProducer setJMSReplyTo(Destination replyTo) {
+        this.replyTo = replyTo;
+        return this;
+    }
+
+    @Override
+    public Destination getJMSReplyTo() {
+        return replyTo;
+    }
+
+    private void checkProperty(String name, Object value) {
+        ActiveMQMessageProducerSupport.validateValidPropertyName(name);
+        ActiveMQMessageProducerSupport.validateValidPropertyValue(name, value);
+    }
+
+    private Map<String, Object> getCreatedMessageProperties() {
+        if(messageProperties == null) {
+            messageProperties = new HashMap<>();
+        }
+        return messageProperties;
+    }
+}
diff --git a/activemq-client/src/main/java/org/apache/activemq/ActiveMQSession.java b/activemq-client/src/main/java/org/apache/activemq/ActiveMQSession.java
index 2ba1fe1..66f7347 100644
--- a/activemq-client/src/main/java/org/apache/activemq/ActiveMQSession.java
+++ b/activemq-client/src/main/java/org/apache/activemq/ActiveMQSession.java
@@ -19,6 +19,8 @@
 import java.io.File;
 import java.io.InputStream;
 import java.io.Serializable;
+import java.lang.reflect.Method;
+import java.lang.reflect.Modifier;
 import java.net.URL;
 import java.util.Collections;
 import java.util.Iterator;
@@ -1394,12 +1396,14 @@
 
     @Override
     public MessageConsumer createDurableConsumer(Topic topic, String name) throws JMSException {
-        throw new UnsupportedOperationException("createDurableConsumer(Topic, name) is not supported");
+        checkClosed();
+        return createDurableSubscriber(topic, name, null, false);
     }
 
     @Override
     public MessageConsumer createDurableConsumer(Topic topic, String name, String messageSelector, boolean noLocal) throws JMSException {
-        throw new UnsupportedOperationException("createDurableConsumer(Topic, name, messageSelector, noLocal) is not supported");
+        checkClosed();
+        return createDurableSubscriber(topic, name, messageSelector, noLocal);
     }
 
     @Override
@@ -1939,6 +1943,26 @@
      */
     protected void send(ActiveMQMessageProducer producer, ActiveMQDestination destination, Message message, int deliveryMode, int priority, long timeToLive,
                         MemoryUsage producerWindow, int sendTimeout, AsyncCallback onComplete) throws JMSException {
+        send(producer, destination, message, deliveryMode, priority, timeToLive, producer.getDisableMessageID(), producer.getDisableMessageID(), producerWindow, sendTimeout, onComplete);
+    }
+
+    /**
+     * Sends the message for dispatch by the broker.
+     *
+     * @param producer - message producer.
+     * @param destination - message destination.
+     * @param message - message to be sent.
+     * @param deliveryMode - JMS message delivery mode.
+     * @param priority - message priority.
+     * @param timeToLive - message expiration.
+     * @param disableTimestamp - disable timestamp.
+     * @param disableMessageID - optionally, disable messageID.
+     * @param producerWindow
+     * @param onComplete
+     * @throws JMSException
+     */
+    protected void send(ActiveMQMessageProducer producer, ActiveMQDestination destination, Message message, int deliveryMode, int priority, long timeToLive,
+                        boolean disableMessageID, boolean disableMessageTimestamp, MemoryUsage producerWindow, int sendTimeout, AsyncCallback onComplete) throws JMSException {
 
         checkClosed();
         if (destination.isTemporary() && connection.isDeleted(destination)) {
@@ -1956,12 +1980,22 @@
             //Set the "JMS" header fields on the original message, see 1.1 spec section 3.4.11
             message.setJMSDeliveryMode(deliveryMode);
             long expiration = 0L;
-            if (!producer.getDisableMessageTimestamp()) {
-                long timeStamp = System.currentTimeMillis();
+            long timeStamp = System.currentTimeMillis();
+            if (timeToLive > 0) {
+                expiration = timeToLive + timeStamp;
+            }
+
+            // TODO: AMQ-8500 - update this when openwire supports JMSDeliveryTime
+            // ref: ActiveMQMessageTransformation#copyProperties
+            if(!(message instanceof ActiveMQMessage)) {
+                setForeignMessageDeliveryTime(message, timeStamp);
+            } else {
+                message.setJMSDeliveryTime(timeStamp);
+            }
+            if (!disableMessageTimestamp && !producer.getDisableMessageTimestamp()) {
                 message.setJMSTimestamp(timeStamp);
-                if (timeToLive > 0) {
-                    expiration = timeToLive + timeStamp;
-                }
+            } else {
+                message.setJMSTimestamp(0l);
             }
             message.setJMSExpiration(expiration);
             message.setJMSPriority(priority);
@@ -2287,4 +2321,22 @@
     protected ThreadPoolExecutor getConnectionExecutor() {
         return this.connectionExecutor;
     }
+
+    private static void setForeignMessageDeliveryTime(final Message foreignMessage, final long deliveryTime) throws JMSException {
+        // Check for JMS v2 message via presence of setJMSDeliveryTime
+        Method deliveryTimeMethod = null;
+        try {
+            Class<?> clazz = foreignMessage.getClass();
+            Method method = clazz.getMethod("setJMSDeliveryTime", new Class[] { long.class });
+            if (!Modifier.isAbstract(method.getModifiers())) {
+                deliveryTimeMethod = method;
+            }
+        } catch (NoSuchMethodException e) {
+            // non-JMS v2 message
+        }
+
+        if (deliveryTimeMethod != null) {
+            foreignMessage.setJMSDeliveryTime(deliveryTime);
+        }
+    }
 }
diff --git a/activemq-client/src/main/java/org/apache/activemq/util/JMSExceptionSupport.java b/activemq-client/src/main/java/org/apache/activemq/util/JMSExceptionSupport.java
index a73f01f..41711ca 100644
--- a/activemq-client/src/main/java/org/apache/activemq/util/JMSExceptionSupport.java
+++ b/activemq-client/src/main/java/org/apache/activemq/util/JMSExceptionSupport.java
@@ -16,10 +16,21 @@
  */
 package org.apache.activemq.util;
 
+import javax.jms.IllegalStateRuntimeException;
+import javax.jms.InvalidClientIDRuntimeException;
+import javax.jms.InvalidDestinationRuntimeException;
+import javax.jms.InvalidSelectorRuntimeException;
 import javax.jms.JMSException;
+import javax.jms.JMSRuntimeException;
 import javax.jms.JMSSecurityException;
+import javax.jms.JMSSecurityRuntimeException;
 import javax.jms.MessageEOFException;
 import javax.jms.MessageFormatException;
+import javax.jms.MessageFormatRuntimeException;
+import javax.jms.MessageNotWriteableRuntimeException;
+import javax.jms.ResourceAllocationRuntimeException;
+import javax.jms.TransactionInProgressRuntimeException;
+import javax.jms.TransactionRolledBackRuntimeException;
 
 import org.apache.activemq.MaxFrameSizeExceededException;
 
@@ -105,4 +116,38 @@
         exception.initCause(cause);
         return exception;
     }
+
+    public static JMSRuntimeException convertToJMSRuntimeException(JMSException e) {
+        if (e instanceof javax.jms.IllegalStateException) {
+            return new IllegalStateRuntimeException(e.getMessage(), e.getErrorCode(), e);
+        }
+        if (e instanceof javax.jms.InvalidClientIDException) {
+            return new InvalidClientIDRuntimeException(e.getMessage(), e.getErrorCode(), e);
+        }
+        if (e instanceof javax.jms.InvalidDestinationException) {
+            return new InvalidDestinationRuntimeException(e.getMessage(), e.getErrorCode(), e);
+        }
+        if (e instanceof javax.jms.InvalidSelectorException) {
+            return new InvalidSelectorRuntimeException(e.getMessage(), e.getErrorCode(), e);
+        }
+        if (e instanceof javax.jms.JMSSecurityException) {
+            return new JMSSecurityRuntimeException(e.getMessage(), e.getErrorCode(), e);
+        }
+        if (e instanceof javax.jms.MessageFormatException) {
+            return new MessageFormatRuntimeException(e.getMessage(), e.getErrorCode(), e);
+        }
+        if (e instanceof javax.jms.MessageNotWriteableException) {
+            return new MessageNotWriteableRuntimeException(e.getMessage(), e.getErrorCode(), e);
+        }
+        if (e instanceof javax.jms.ResourceAllocationException) {
+            return new ResourceAllocationRuntimeException(e.getMessage(), e.getErrorCode(), e);
+        }
+        if (e instanceof javax.jms.TransactionInProgressException) {
+            return new TransactionInProgressRuntimeException(e.getMessage(), e.getErrorCode(), e);
+        }
+        if (e instanceof javax.jms.TransactionRolledBackException) {
+            return new TransactionRolledBackRuntimeException(e.getMessage(), e.getErrorCode(), e);
+        }
+        return new JMSRuntimeException(e.getMessage(), e.getErrorCode(), e);
+    }
 }
diff --git a/activemq-unit-tests/pom.xml b/activemq-unit-tests/pom.xml
index 0770874..7a8ec71 100644
--- a/activemq-unit-tests/pom.xml
+++ b/activemq-unit-tests/pom.xml
@@ -323,9 +323,9 @@
             <link>http://junit.sourceforge.net/javadoc/</link>
           </links>
           <stylesheetfile>${basedir}/../etc/css/stylesheet.css</stylesheetfile>
-          <linksource>true</linksource>
           <maxmemory>256m</maxmemory>
           <source>${source-version}</source>
+          <noindex>true</noindex>
           <groups>
             <group>
               <title>JMS Client</title>
diff --git a/activemq-unit-tests/src/test/java/org/apache/activemq/jms2/ActiveMQJMS2AckModesTest.java b/activemq-unit-tests/src/test/java/org/apache/activemq/jms2/ActiveMQJMS2AckModesTest.java
new file mode 100644
index 0000000..25e45ec
--- /dev/null
+++ b/activemq-unit-tests/src/test/java/org/apache/activemq/jms2/ActiveMQJMS2AckModesTest.java
@@ -0,0 +1,162 @@
+/**
+ * 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.jms2;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
+
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.LinkedList;
+import java.util.List;
+
+import javax.jms.DeliveryMode;
+import javax.jms.Destination;
+import javax.jms.JMSConsumer;
+import javax.jms.JMSContext;
+import javax.jms.Message;
+
+import org.apache.activemq.ActiveMQSession;
+import org.apache.activemq.broker.jmx.DestinationViewMBean;
+import org.apache.activemq.command.ActiveMQDestination;
+import org.apache.activemq.util.Wait;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.junit.runners.Parameterized;
+
+@RunWith(value = Parameterized.class)
+public class ActiveMQJMS2AckModesTest extends ActiveMQJMS2TestBase {
+
+    private final String destinationName;
+    private final String destinationType;
+    private final int ackMode;
+    private final String messagePayload;
+
+    public ActiveMQJMS2AckModesTest(String destinationType, int ackMode) {
+        this.destinationName = "AMQ.JMS2.ACKMODE." + Integer.toString(ackMode) + destinationType.toUpperCase();
+        this.destinationType = destinationType;
+        this.ackMode = ackMode;
+        this.messagePayload = "Test message destType: " + destinationType + " ackMode: " + Integer.toString(ackMode);
+    }
+
+    @Parameterized.Parameters(name="destinationType={0}, ackMode={1}")
+    public static Collection<Object[]> data() {
+        return Arrays.asList(new Object[][] {
+                {"queue", ActiveMQSession.INDIVIDUAL_ACKNOWLEDGE },
+                {"queue", ActiveMQSession.AUTO_ACKNOWLEDGE },
+                {"queue", ActiveMQSession.CLIENT_ACKNOWLEDGE },
+                {"queue", ActiveMQSession.DUPS_OK_ACKNOWLEDGE },
+                {"queue", ActiveMQSession.SESSION_TRANSACTED },
+                {"topic", ActiveMQSession.INDIVIDUAL_ACKNOWLEDGE },
+                {"topic", ActiveMQSession.AUTO_ACKNOWLEDGE },
+                {"topic", ActiveMQSession.CLIENT_ACKNOWLEDGE },
+                {"topic", ActiveMQSession.DUPS_OK_ACKNOWLEDGE },
+                {"topic", ActiveMQSession.SESSION_TRANSACTED },
+                {"temp-queue", ActiveMQSession.INDIVIDUAL_ACKNOWLEDGE },
+                {"temp-queue", ActiveMQSession.AUTO_ACKNOWLEDGE },
+                {"temp-queue", ActiveMQSession.CLIENT_ACKNOWLEDGE },
+                {"temp-queue", ActiveMQSession.DUPS_OK_ACKNOWLEDGE },
+                {"temp-queue", ActiveMQSession.SESSION_TRANSACTED },
+                {"temp-topic", ActiveMQSession.INDIVIDUAL_ACKNOWLEDGE },
+                {"temp-topic", ActiveMQSession.AUTO_ACKNOWLEDGE },
+                {"temp-topic", ActiveMQSession.CLIENT_ACKNOWLEDGE },
+                {"temp-topic", ActiveMQSession.DUPS_OK_ACKNOWLEDGE },
+                {"temp-topic", ActiveMQSession.SESSION_TRANSACTED }
+        });
+    }
+
+    @Test
+    public void testAcknowledgementMode() {
+
+        try(JMSContext jmsContext = activemqConnectionFactory.createContext(DEFAULT_JMS_USER, DEFAULT_JMS_PASS, ackMode)) {
+            assertNotNull(jmsContext);
+            Destination destination = ActiveMQJMS2TestSupport.generateDestination(jmsContext, destinationType, destinationName);
+            assertNotNull(destination);
+            JMSConsumer jmsConsumer = jmsContext.createConsumer(destination);
+            assertNotNull(jmsConsumer);
+            jmsContext.start();
+
+            Message message = ActiveMQJMS2TestSupport.generateMessage(jmsContext, "text", messagePayload);
+
+            List<String> sentMessageIds = new LinkedList<>();
+            for(int deliveryMode : Arrays.asList(DeliveryMode.NON_PERSISTENT, DeliveryMode.PERSISTENT)) {
+                MessageData sendMessageData = new MessageData();
+                sendMessageData.setMessage(message).setDeliveryMode(deliveryMode);
+                sentMessageIds.add(ActiveMQJMS2TestSupport.sendMessage(jmsContext, destination, sendMessageData));
+            }
+
+            // For session transacted ack after all messages are sent
+            switch(ackMode) {
+            case ActiveMQSession.SESSION_TRANSACTED: jmsContext.commit(); break;
+            default: break;
+            }
+
+            Message recvMessage = null;
+            Message lastRecvMessage = null;
+            List<Message> recvMessages = new LinkedList<>();
+
+            recvMessage = jmsConsumer.receive(2000l);
+
+            while (recvMessage != null) {
+                recvMessages.add(recvMessage);
+                lastRecvMessage = recvMessage;
+                switch(ackMode) {
+                case ActiveMQSession.INDIVIDUAL_ACKNOWLEDGE: lastRecvMessage.acknowledge(); break;
+                default: break;
+                }
+
+                recvMessage = jmsConsumer.receive(500l);
+            }
+
+            assertEquals(Integer.valueOf(2), Integer.valueOf(recvMessages.size()));
+
+            switch(ackMode) {
+            case ActiveMQSession.CLIENT_ACKNOWLEDGE: lastRecvMessage.acknowledge(); break;
+            case ActiveMQSession.SESSION_TRANSACTED: jmsContext.commit(); break;
+            default: break;
+            }
+            jmsConsumer.close();
+
+            int foundCount = 0;
+            for(int validDeliveryMode : Arrays.asList(DeliveryMode.NON_PERSISTENT, DeliveryMode.PERSISTENT)) {
+                for(javax.jms.Message tmpMessage : recvMessages) {
+                    if(tmpMessage.getJMSDeliveryMode() == validDeliveryMode) {
+                        MessageData messageData = new MessageData();
+                        messageData.setMessageType("text").setMessagePayload(messagePayload).setDeliveryMode(validDeliveryMode);
+                        ActiveMQJMS2TestSupport.validateMessageData(tmpMessage, messageData);
+                        foundCount++;
+                    }
+                }
+            }
+            assertEquals(Integer.valueOf(2), Integer.valueOf(foundCount));
+
+            DestinationViewMBean destinationViewMBean = getDestinationViewMBean(destinationType, (ActiveMQDestination)destination);
+            assertTrue("QueueSize = 0 expected", Wait.waitFor(new Wait.Condition() {
+                @Override
+                public boolean isSatisified() throws Exception {
+                    return destinationViewMBean.getQueueSize() == 0l;
+                }
+            }, 5000l, 10l));
+        } catch (Exception e) {
+            fail(e.getMessage());
+        }
+    }
+
+
+}
diff --git a/activemq-unit-tests/src/test/java/org/apache/activemq/jms2/ActiveMQJMS2ContextTest.java b/activemq-unit-tests/src/test/java/org/apache/activemq/jms2/ActiveMQJMS2ContextTest.java
new file mode 100644
index 0000000..4891603
--- /dev/null
+++ b/activemq-unit-tests/src/test/java/org/apache/activemq/jms2/ActiveMQJMS2ContextTest.java
@@ -0,0 +1,360 @@
+/**
+ * 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.jms2;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertNull;
+import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
+
+import java.util.Enumeration;
+
+import javax.jms.CompletionListener;
+import javax.jms.Destination;
+import javax.jms.JMSConsumer;
+import javax.jms.JMSContext;
+import javax.jms.JMSException;
+import javax.jms.JMSProducer;
+import javax.jms.JMSRuntimeException;
+import javax.jms.Message;
+import javax.jms.Queue;
+import javax.jms.QueueBrowser;
+import javax.jms.Session;
+import javax.jms.TextMessage;
+import javax.jms.Topic;
+
+import org.apache.activemq.ActiveMQContext;
+import org.junit.Test;
+
+public class ActiveMQJMS2ContextTest extends ActiveMQJMS2TestBase {
+
+    @Test
+    public void testConnectionFactoryCreateContext() {
+        try(JMSContext jmsContext = activemqConnectionFactory.createContext()) {
+            assertNotNull(jmsContext);
+            jmsContext.start();
+            assertTrue(ActiveMQContext.class.isAssignableFrom(jmsContext.getClass()));
+            Destination destination = jmsContext.createQueue(methodNameDestinationName);
+            sendMessage(jmsContext, destination, "Test-" + methodNameDestinationName);
+            recvMessage(jmsContext, destination, "Test-" + methodNameDestinationName);
+        } catch (JMSException e) {
+            fail(e.getMessage());
+        }
+    }
+
+    @Test(expected = UnsupportedOperationException.class)
+    public void testConnectionFactoryCreateContextSession() {
+        activemqConnectionFactory.createContext(Session.AUTO_ACKNOWLEDGE);
+    }
+
+    @Test
+    public void testConnectionFactoryCreateContextUserPass() {
+        try(JMSContext jmsContext = activemqConnectionFactory.createContext(DEFAULT_JMS_USER, DEFAULT_JMS_PASS)) {
+            assertNotNull(jmsContext);
+            jmsContext.start();
+            assertTrue(ActiveMQContext.class.isAssignableFrom(jmsContext.getClass()));
+            Destination destination = jmsContext.createQueue(methodNameDestinationName);
+            sendMessage(jmsContext, destination, "Test-" + methodNameDestinationName);
+            recvMessage(jmsContext, destination, "Test-" + methodNameDestinationName);
+        } catch (JMSException e) {
+            fail(e.getMessage());
+        }
+    }
+
+    @Test
+    public void testConnectionFactoryCreateContextUserPassSession() {
+        try(JMSContext jmsContext = activemqConnectionFactory.createContext(DEFAULT_JMS_USER, DEFAULT_JMS_PASS, Session.AUTO_ACKNOWLEDGE)) {
+            assertNotNull(jmsContext);
+            assertTrue(ActiveMQContext.class.isAssignableFrom(jmsContext.getClass()));
+        }
+    }
+
+    @Test
+    public void testConnectionFactoryCreateContexMultiContext() {
+        JMSContext secondJMSContext = null;
+        JMSContext thirdJMSContext = null;
+
+        try(JMSContext jmsContext = activemqConnectionFactory.createContext(DEFAULT_JMS_USER, DEFAULT_JMS_PASS)) {
+            assertNotNull(jmsContext);
+            jmsContext.start();
+            assertTrue(ActiveMQContext.class.isAssignableFrom(jmsContext.getClass()));
+
+            Destination testDestination = jmsContext.createQueue(methodNameDestinationName);
+            sendMessage(jmsContext, testDestination, "Test-" + methodNameDestinationName);
+            recvMessage(jmsContext, testDestination, "Test-" + methodNameDestinationName);
+
+            secondJMSContext = jmsContext.createContext(Session.AUTO_ACKNOWLEDGE);
+        } catch (JMSException e) {
+            fail(e.getMessage());
+        }
+
+        // First context closed
+        String secondTestDestinationName = methodNameDestinationName + ".SECOND";
+        Destination secondTestDestination = secondJMSContext.createQueue(secondTestDestinationName);
+
+        try {
+            sendMessage(secondJMSContext, secondTestDestination, "Test-" + methodNameDestinationName);
+            recvMessage(secondJMSContext, secondTestDestination, "Test-" + methodNameDestinationName);
+        } catch (JMSException e) {
+            fail(e.getMessage());
+        } finally {
+            if(secondJMSContext != null) {
+                try { secondJMSContext.close(); } catch (JMSRuntimeException e) { fail(e.getMessage()); }
+            }
+        }
+
+        // Attempt to obtain a third context after all contexts have been closed
+        boolean caught = false;
+        try {
+            thirdJMSContext = secondJMSContext.createContext(Session.AUTO_ACKNOWLEDGE);
+            fail("JMSRuntimeException expected");
+        } catch (JMSRuntimeException e) {
+            assertNull(thirdJMSContext);
+            caught = true;
+            assertEquals("Context already closed", e.getMessage());
+        }
+        assertTrue(caught);
+    }
+
+    @Test
+    public void testConnectionFactoryCreateContextBrowse() {
+        try(JMSContext jmsContext = activemqConnectionFactory.createContext()) {
+            assertNotNull(jmsContext);
+            jmsContext.start();
+            assertTrue(ActiveMQContext.class.isAssignableFrom(jmsContext.getClass()));
+            Destination destination = jmsContext.createQueue(methodNameDestinationName);
+            sendMessage(jmsContext, destination, "Test-" + methodNameDestinationName);
+            browseMessage(jmsContext, destination, "Test-" + methodNameDestinationName, true);
+        } catch (JMSException e) {
+            fail(e.getMessage());
+        }
+    }
+
+    @Test
+    public void testConnectionFactoryCreateContextBrowseAutoStart() {
+        try(JMSContext jmsContext = activemqConnectionFactory.createContext()) {
+            assertNotNull(jmsContext);
+            assertTrue(ActiveMQContext.class.isAssignableFrom(jmsContext.getClass()));
+            Destination destination = jmsContext.createQueue(methodNameDestinationName);
+            sendMessage(jmsContext, destination, "Test-" + methodNameDestinationName);
+            browseMessage(jmsContext, destination, "Test-" + methodNameDestinationName, true);
+        } catch (JMSException e) {
+            fail(e.getMessage());
+        }
+    }
+
+    @Test
+    public void testConnectionFactoryCreateContextBrowseAutoStartFalse() {
+        try(JMSContext jmsContext = activemqConnectionFactory.createContext()) {
+            assertNotNull(jmsContext);
+            jmsContext.setAutoStart(false);
+            assertTrue(ActiveMQContext.class.isAssignableFrom(jmsContext.getClass()));
+            Destination destination = jmsContext.createQueue(methodNameDestinationName);
+            sendMessage(jmsContext, destination, "Test-" + methodNameDestinationName);
+            browseMessage(jmsContext, destination, "Test-" + methodNameDestinationName, false);
+        } catch (JMSException e) {
+            fail(e.getMessage());
+        }
+    }
+
+    @Test
+    public void testConnectionFactoryCreateContextBrowseAutoStartFalseStartDelayed() {
+        try(JMSContext jmsContext = activemqConnectionFactory.createContext()) {
+            assertNotNull(jmsContext);
+            jmsContext.setAutoStart(false);
+            assertTrue(ActiveMQContext.class.isAssignableFrom(jmsContext.getClass()));
+            Destination destination = jmsContext.createQueue(methodNameDestinationName);
+            sendMessage(jmsContext, destination, "Test-" + methodNameDestinationName);
+            jmsContext.start();
+            browseMessage(jmsContext, destination, "Test-" + methodNameDestinationName, true);
+        } catch (JMSException e) {
+            fail(e.getMessage());
+        }
+    }
+
+    @Test
+    public void testDisableMessageID() {
+        try(JMSContext jmsContext = activemqConnectionFactory.createContext()) {
+            assertNotNull(jmsContext);
+            jmsContext.setAutoStart(false);
+            assertTrue(ActiveMQContext.class.isAssignableFrom(jmsContext.getClass()));
+            jmsContext.start();
+            JMSProducer jmsProducer = jmsContext.createProducer();
+            jmsProducer.setDisableMessageID(true);
+            Destination destination = jmsContext.createQueue(methodNameDestinationName);
+            Message sendMessage = jmsContext.createTextMessage("Test-" + methodNameDestinationName);
+            jmsProducer.send(destination, sendMessage);
+            JMSConsumer jmsConsumer = jmsContext.createConsumer(destination);
+            Message recvMessage = jmsConsumer.receive(5000l);
+            assertNotNull(recvMessage);
+            // Verify messageID since ActiveMQ does not disableMessageID
+            assertEquals(sendMessage.getJMSMessageID(), recvMessage.getJMSMessageID());
+        } catch (JMSException e) {
+            fail(e.getMessage());
+        }
+    }
+
+    @Test(expected = UnsupportedOperationException.class)
+    public void testConnectionSharedConnectionConsumer() throws JMSException {
+        connection.createSharedConnectionConsumer(null, null, null, null, 10);
+    }
+
+    @Test(expected = UnsupportedOperationException.class)
+    public void testConnectionSharedDurableConnectionConsumer() throws JMSException {
+        connection.createSharedDurableConnectionConsumer(null, null, null, null, 10);
+    }
+
+    @Test(expected = UnsupportedOperationException.class)
+    public void testSessionAckMode() throws JMSException {
+        connection.createSession(Session.AUTO_ACKNOWLEDGE);
+    }
+
+    @Test
+    public void testSessionDurableConsumer() throws JMSException {
+        try(JMSContext jmsContext = activemqConnectionFactory.createContext()) {
+            assertNotNull(jmsContext);
+            jmsContext.setAutoStart(false);
+            jmsContext.setClientID(testName.getMethodName());
+            jmsContext.start();
+            assertTrue(ActiveMQContext.class.isAssignableFrom(jmsContext.getClass()));
+            Topic topic = jmsContext.createTopic(methodNameDestinationName);
+
+            JMSConsumer jmsConsumerRegisterDurableConsumer = jmsContext.createDurableConsumer(topic, testName.getMethodName());
+            jmsConsumerRegisterDurableConsumer.close();
+            sendMessage(jmsContext, topic, "Test-" + methodNameDestinationName);
+            recvMessageDurable(jmsContext, topic, testName.getMethodName(), null, false, "Test-" + methodNameDestinationName);
+        } catch (JMSException e) {
+            fail(e.getMessage());
+        }
+    }
+
+    @Test
+    public void testSessionDurableConsumerSelectorNoLocal() throws JMSException {
+        try(JMSContext jmsContext = activemqConnectionFactory.createContext()) {
+            assertNotNull(jmsContext);
+            jmsContext.setAutoStart(false);
+            jmsContext.setClientID(testName.getMethodName());
+            jmsContext.start();
+            assertTrue(ActiveMQContext.class.isAssignableFrom(jmsContext.getClass()));
+            Topic topic = jmsContext.createTopic(methodNameDestinationName);
+
+            JMSConsumer jmsConsumerRegisterDurableConsumer = jmsContext.createDurableConsumer(topic, testName.getMethodName(), "TRUE", false);
+            jmsConsumerRegisterDurableConsumer.close();
+            sendMessage(jmsContext, topic, "Test-" + methodNameDestinationName);
+            recvMessageDurable(jmsContext, topic, testName.getMethodName(), "TRUE", false, "Test-" + methodNameDestinationName);
+        } catch (JMSException e) {
+            fail(e.getMessage());
+        }
+    }
+
+    @Test(expected = UnsupportedOperationException.class)
+    public void testSessionSharedConsumer() throws JMSException {
+        session.createSharedConsumer(null, null);
+    }
+
+    @Test(expected = UnsupportedOperationException.class)
+    public void testSessionSharedConsumerSelector() throws JMSException {
+        session.createSharedConsumer(null, null, null);
+    }
+
+    @Test(expected = UnsupportedOperationException.class)
+    public void testSessionSharedDurableConsumer() throws JMSException {
+        session.createSharedDurableConsumer(null, null);
+    }
+
+    @Test(expected = UnsupportedOperationException.class)
+    public void testSessionSharedDurableConsumerSelector() throws JMSException {
+        session.createSharedDurableConsumer(null, null, null);
+    }
+
+    @Test(expected = UnsupportedOperationException.class)
+    public void testProducerDeliveryDelayGet() throws JMSException {
+        messageProducer.getDeliveryDelay();
+    }
+
+    @Test(expected = UnsupportedOperationException.class)
+    public void testProducerDeliveryDelaySet() throws JMSException {
+        messageProducer.setDeliveryDelay(1000l);
+    }
+
+    @Test(expected = UnsupportedOperationException.class)
+    public void testProducerSendMessageCompletionListener() throws JMSException {
+         messageProducer.send(session.createQueue(methodNameDestinationName), null, (CompletionListener)null);
+    }
+
+    @Test(expected = UnsupportedOperationException.class)
+    public void testProducerSendMessageQoSParamsCompletionListener() throws JMSException {
+         messageProducer.send(null, 1, 4, 0l, null);
+    }
+
+    @Test(expected = UnsupportedOperationException.class)
+    public void testProducerSendDestinationMessageCompletionListener() throws JMSException {
+         messageProducer.send(session.createQueue(methodNameDestinationName), null, null);
+    }
+
+    @Test(expected = UnsupportedOperationException.class)
+    public void testProducerSendDestinationMessageQosParamsCompletionListener() throws JMSException {
+         messageProducer.send(session.createQueue(methodNameDestinationName), null, 1, 4, 0l, null);
+    }
+
+    protected static void sendMessage(JMSContext jmsContext, Destination testDestination, String textBody) {
+        assertNotNull(jmsContext);
+        JMSProducer jmsProducer = jmsContext.createProducer();
+        jmsProducer.send(testDestination, textBody);
+    }
+
+    protected static void browseMessage(JMSContext jmsContext, Destination testDestination, String expectedTextBody, boolean expectFound) throws JMSException {
+        assertNotNull(jmsContext);
+        assertTrue(Queue.class.isAssignableFrom(testDestination.getClass()));
+        Queue testQueue = Queue.class.cast(testDestination);
+        try(QueueBrowser queueBrowser = jmsContext.createBrowser(testQueue)) {
+            Enumeration<?> messageEnumeration = queueBrowser.getEnumeration();
+            assertNotNull(messageEnumeration);
+
+            boolean found = false; 
+            while(!found && messageEnumeration.hasMoreElements()) {
+                javax.jms.Message message = (javax.jms.Message)messageEnumeration.nextElement();
+                assertNotNull(message);
+                assertTrue(TextMessage.class.isAssignableFrom(message.getClass()));
+                assertEquals(expectedTextBody, TextMessage.class.cast(message).getText());
+                found = true;
+            }
+            assertEquals(expectFound, found);
+        }
+    }
+
+    protected static void recvMessage(JMSContext jmsContext, Destination testDestination, String expectedTextBody) throws JMSException {
+        assertNotNull(jmsContext);
+        try(JMSConsumer jmsConsumer = jmsContext.createConsumer(testDestination)) {
+            javax.jms.Message message = jmsConsumer.receive(1000l);
+            assertNotNull(message);
+            assertTrue(TextMessage.class.isAssignableFrom(message.getClass()));
+            assertEquals(expectedTextBody, TextMessage.class.cast(message).getText());
+        }
+    }
+
+    protected static void recvMessageDurable(JMSContext jmsContext, Topic testTopic, String subscriptionName, String selector, boolean noLocal, String expectedTextBody) throws JMSException {
+        assertNotNull(jmsContext);
+        try(JMSConsumer jmsConsumer = jmsContext.createDurableConsumer(testTopic, subscriptionName, selector, noLocal)) {
+            javax.jms.Message message = jmsConsumer.receive(1000l);
+            assertNotNull(message);
+            assertTrue(TextMessage.class.isAssignableFrom(message.getClass()));
+            assertEquals(expectedTextBody, TextMessage.class.cast(message).getText());
+        }
+    }
+}
diff --git a/activemq-unit-tests/src/test/java/org/apache/activemq/jms2/ActiveMQJMS2MessageListenerTest.java b/activemq-unit-tests/src/test/java/org/apache/activemq/jms2/ActiveMQJMS2MessageListenerTest.java
new file mode 100644
index 0000000..917a26b
--- /dev/null
+++ b/activemq-unit-tests/src/test/java/org/apache/activemq/jms2/ActiveMQJMS2MessageListenerTest.java
@@ -0,0 +1,154 @@
+/**
+ * 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.jms2;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
+
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import javax.jms.DeliveryMode;
+import javax.jms.Destination;
+import javax.jms.JMSConsumer;
+import javax.jms.JMSContext;
+import javax.jms.JMSException;
+import javax.jms.Message;
+import javax.jms.MessageListener;
+import javax.jms.Session;
+
+import org.apache.activemq.ActiveMQSession;
+import org.apache.activemq.broker.jmx.QueueViewMBean;
+import org.apache.activemq.command.ActiveMQDestination;
+import org.apache.activemq.util.Wait;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.junit.runners.Parameterized;
+
+@RunWith(value = Parameterized.class)
+public class ActiveMQJMS2MessageListenerTest extends ActiveMQJMS2TestBase {
+
+    private final String destinationName;
+    private final String destinationType;
+    private final int ackMode;
+    private final String messagePayload;
+
+    public ActiveMQJMS2MessageListenerTest(String destinationType, int ackMode) {
+        this.destinationName = "AMQ.JMS2.ACKMODE." + Integer.toString(ackMode) + destinationType.toUpperCase();
+        this.destinationType = destinationType;
+        this.ackMode = ackMode;
+        this.messagePayload = "Test message destType: " + destinationType + " ackMode: " + Integer.toString(ackMode);
+    }
+
+    @Parameterized.Parameters(name="destinationType={0}, ackMode={1}")
+    public static Collection<Object[]> data() {
+        return Arrays.asList(new Object[][] {
+                {"queue", ActiveMQSession.INDIVIDUAL_ACKNOWLEDGE },
+                {"queue", Session.AUTO_ACKNOWLEDGE },
+                {"queue", Session.CLIENT_ACKNOWLEDGE },
+                {"queue", Session.DUPS_OK_ACKNOWLEDGE },
+                {"queue", Session.SESSION_TRANSACTED }
+        });
+    }
+
+    @Test
+    public void testMessageListener() {
+
+        try(JMSContext jmsContext = activemqConnectionFactory.createContext(DEFAULT_JMS_USER, DEFAULT_JMS_PASS, ackMode)) {
+            assertNotNull(jmsContext);
+            Destination destination = ActiveMQJMS2TestSupport.generateDestination(jmsContext, destinationType, destinationName);
+            assertNotNull(destination);
+            QueueViewMBean localQueueViewMBean = getQueueViewMBean((ActiveMQDestination)destination);
+            JMSConsumer jmsConsumer = jmsContext.createConsumer(destination);
+
+            AtomicInteger receivedMessageCount = new AtomicInteger();
+            AtomicInteger exceptionCount = new AtomicInteger();
+            CountDownLatch countDownLatch = new CountDownLatch(2);
+
+            jmsConsumer.setMessageListener(new MessageListener() {
+                @Override
+                public void onMessage(Message message) {
+                    countDownLatch.countDown();
+                    receivedMessageCount.incrementAndGet();
+                    try {
+                        switch(ackMode) {
+                        case Session.CLIENT_ACKNOWLEDGE: message.acknowledge(); break;
+                        case ActiveMQSession.INDIVIDUAL_ACKNOWLEDGE: message.acknowledge(); break;
+                        default: break;
+                        }
+                    } catch (JMSException e) {
+                        exceptionCount.incrementAndGet();
+                    }
+                }
+            });
+            jmsContext.start();
+
+            Message message = ActiveMQJMS2TestSupport.generateMessage(jmsContext, "text", messagePayload);
+
+            List<String> sentMessageIds = new LinkedList<>();
+            for(int deliveryMode : Arrays.asList(DeliveryMode.NON_PERSISTENT, DeliveryMode.PERSISTENT)) {
+                MessageData sendMessageData = new MessageData();
+                sendMessageData.setMessage(message).setDeliveryMode(deliveryMode);
+                sentMessageIds.add(ActiveMQJMS2TestSupport.sendMessage(jmsContext, destination, sendMessageData));
+            }
+
+            // For session transacted ack we ack after all messages are sent
+            switch(ackMode) {
+            case ActiveMQSession.SESSION_TRANSACTED:
+                assertEquals(Long.valueOf(0), Long.valueOf(localQueueViewMBean.getEnqueueCount()));
+                jmsContext.commit();
+                assertEquals(Long.valueOf(2), Long.valueOf(localQueueViewMBean.getEnqueueCount()));
+                break;
+            default: 
+                assertEquals(Long.valueOf(2), Long.valueOf(localQueueViewMBean.getEnqueueCount()));
+                break;
+            }
+
+            countDownLatch.await(5, TimeUnit.SECONDS);
+
+            assertEquals(Integer.valueOf(2), Integer.valueOf(receivedMessageCount.get()));
+            assertEquals(Integer.valueOf(0), Integer.valueOf(exceptionCount.get()));
+
+            // For session transacted we ack after all messages are received
+            switch(ackMode) {
+            case ActiveMQSession.SESSION_TRANSACTED:
+                assertEquals(Long.valueOf(0), Long.valueOf(localQueueViewMBean.getDequeueCount()));
+                jmsContext.commit();
+                break;
+            default: break;
+            }
+            jmsConsumer.close();
+
+            assertTrue("DequeueCount = 2 and QueueSize = 0 expected", Wait.waitFor(new Wait.Condition() {
+                @Override
+                public boolean isSatisified() throws Exception {
+                    return localQueueViewMBean.getDequeueCount() == 2l && localQueueViewMBean.getQueueSize() == 0l;
+                }
+            }, 5000l, 100l));
+
+        } catch (Exception e) {
+            fail(e.getMessage());
+        }
+    }
+}
diff --git a/activemq-unit-tests/src/test/java/org/apache/activemq/jms2/ActiveMQJMS2MessageTypesTest.java b/activemq-unit-tests/src/test/java/org/apache/activemq/jms2/ActiveMQJMS2MessageTypesTest.java
new file mode 100644
index 0000000..465c067
--- /dev/null
+++ b/activemq-unit-tests/src/test/java/org/apache/activemq/jms2/ActiveMQJMS2MessageTypesTest.java
@@ -0,0 +1,597 @@
+/**
+ * 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.jms2;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotEquals;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
+
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.Set;
+import java.util.UUID;
+import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import javax.jms.DeliveryMode;
+import javax.jms.Destination;
+import javax.jms.IllegalStateRuntimeException;
+import javax.jms.JMSConsumer;
+import javax.jms.JMSContext;
+import javax.jms.JMSException;
+import javax.jms.Message;
+import javax.jms.MessageListener;
+import javax.jms.Session;
+import javax.jms.Topic;
+
+import org.apache.activemq.util.Wait;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.junit.runners.Parameterized;
+
+@RunWith(value = Parameterized.class)
+public class ActiveMQJMS2MessageTypesTest extends ActiveMQJMS2TestBase {
+
+    private final Set<Integer> VALID_PRIORITIES = Set.of(0, 1, 2, 3, 4, 5, 6, 7, 8, 9);
+    private final String clientID;
+    private final String destinationType;
+    private final String messagePayload;
+    private final String messageType;
+
+    public ActiveMQJMS2MessageTypesTest(String destinationType, String messageType) {
+        this.clientID = destinationType + "-" + messageType;
+        this.destinationType = destinationType;
+        this.messagePayload = "Test message payload";
+        this.messageType = messageType;
+    }
+
+    @Parameterized.Parameters(name="destinationType={0}, messageType={1}")
+    public static Collection<Object[]> data() {
+        return Arrays.asList(new Object[][] {
+                {"queue", "bytes"},
+                {"queue", "map"},
+                {"queue", "object"},
+                {"queue", "stream"},
+                {"queue", "text"},
+                {"topic", "bytes"},
+                {"topic", "map"},
+                {"topic", "object"},
+                {"topic", "stream"},
+                {"topic", "text"},
+                {"temp-queue", "bytes"},
+                {"temp-queue", "map"},
+                {"temp-queue", "object"},
+                {"temp-queue", "stream"},
+                {"temp-queue", "text"},
+                {"temp-topic", "bytes"},
+                {"temp-topic", "map"},
+                {"temp-topic", "object"},
+                {"temp-topic", "stream"},
+                {"temp-topic", "text"},
+        });
+    }
+
+    @Test
+    public void testMessageDeliveryMode() {
+
+        try(JMSContext jmsContext = activemqConnectionFactory.createContext(DEFAULT_JMS_USER, DEFAULT_JMS_PASS, Session.AUTO_ACKNOWLEDGE)) {
+            assertNotNull(jmsContext);
+            Destination destination = ActiveMQJMS2TestSupport.generateDestination(jmsContext, destinationType, methodNameDestinationName);
+            assertNotNull(destination);
+            JMSConsumer jmsConsumer = jmsContext.createConsumer(destination);
+            assertNotNull(jmsConsumer);
+
+            Message message = ActiveMQJMS2TestSupport.generateMessage(jmsContext, messageType, messagePayload);
+
+            List<String> sentMessageIds = new LinkedList<>();
+            for(int deliveryMode : Arrays.asList(DeliveryMode.NON_PERSISTENT, DeliveryMode.PERSISTENT)) {
+                MessageData sendMessageData = new MessageData();
+                sendMessageData.setMessage(message).setDeliveryMode(deliveryMode);
+                sentMessageIds.add(ActiveMQJMS2TestSupport.sendMessage(jmsContext, destination, sendMessageData));
+            }
+
+            List<Message> recvMessages = new CopyOnWriteArrayList<>();
+            AtomicBoolean receivedExpected = new AtomicBoolean(false);
+            AtomicInteger exceptionCount = new AtomicInteger();
+
+            jmsConsumer.setMessageListener(new MessageListener() {
+                private boolean recvNonPersistent = false;
+                private boolean recvPersistent = false;
+
+                @Override
+                public void onMessage(Message message) {
+                    recvMessages.add(message);
+                    try {
+                        switch(message.getJMSDeliveryMode()) {
+                        case DeliveryMode.NON_PERSISTENT: recvNonPersistent = true; break;
+                        case DeliveryMode.PERSISTENT: recvPersistent = true; break;
+                        default: break;
+                        }
+
+                        if(recvNonPersistent && recvPersistent) {
+                            receivedExpected.set(true);
+                        }
+                    } catch (JMSException e) {
+                        exceptionCount.incrementAndGet();
+                    }
+                }
+            });
+            jmsContext.start();
+
+            assertTrue("Expected to receive both DeliveryModes", Wait.waitFor(new Wait.Condition() {
+                @Override
+                public boolean isSatisified() throws Exception {
+                    return receivedExpected.get();
+                }
+            }, 5000l, 100l));
+            jmsConsumer.close();
+            jmsContext.stop();
+
+            assertEquals(Integer.valueOf(2), Integer.valueOf(recvMessages.size()));
+        } catch (Exception e) {
+            fail(e.getMessage());
+        }
+    }
+
+    @Test
+    public void testMessageDeliveryModeInvalid() {
+
+        try(JMSContext jmsContext = activemqConnectionFactory.createContext(DEFAULT_JMS_USER, DEFAULT_JMS_PASS, Session.AUTO_ACKNOWLEDGE)) {
+            assertNotNull(jmsContext);
+            Destination destination = ActiveMQJMS2TestSupport.generateDestination(jmsContext, destinationType, methodNameDestinationName);
+            assertNotNull(destination);
+            jmsContext.start();
+
+            Message message = ActiveMQJMS2TestSupport.generateMessage(jmsContext, messageType, messagePayload);
+            MessageData sendMessageData = new MessageData();
+            sendMessageData.setMessage(message).setDeliveryMode(99);
+            boolean caught = false;
+            try {
+                ActiveMQJMS2TestSupport.sendMessage(jmsContext, destination, sendMessageData);
+                fail("IlegalStateRuntimeException expected");
+            } catch (IllegalStateRuntimeException e) {
+                assertEquals("unknown delivery mode: 99", e.getMessage());
+                caught = true;
+            }
+            assertTrue(caught);
+        } catch (Exception e) {
+            fail(e.getMessage());
+        }
+    }
+
+    @Test
+    public void testMessagePriority() {
+
+        try(JMSContext jmsContext = activemqConnectionFactory.createContext(DEFAULT_JMS_USER, DEFAULT_JMS_PASS, Session.AUTO_ACKNOWLEDGE)) {
+            assertNotNull(jmsContext);
+            Destination destination = ActiveMQJMS2TestSupport.generateDestination(jmsContext, destinationType, methodNameDestinationName);
+            assertNotNull(destination);
+            JMSConsumer jmsConsumer = jmsContext.createConsumer(destination);
+            assertNotNull(jmsConsumer);
+
+            Message message = ActiveMQJMS2TestSupport.generateMessage(jmsContext, messageType, messagePayload);
+            List<String> sentMessageIds = new LinkedList<>();
+            for(int priority : VALID_PRIORITIES) {
+                MessageData sendMessageData = new MessageData();
+                sendMessageData.setMessage(message).setPriority(priority);
+                sentMessageIds.add(ActiveMQJMS2TestSupport.sendMessage(jmsContext, destination, sendMessageData));
+            }
+
+            final List<Message> recvMessages = new CopyOnWriteArrayList<>();
+            final CountDownLatch countDownLatch = new CountDownLatch(10);
+
+            jmsConsumer.setMessageListener(new MessageListener() {
+                @Override
+                public void onMessage(Message message) {
+                    recvMessages.add(message);
+                    countDownLatch.countDown();
+                }
+            });
+            jmsContext.start();
+
+            assertTrue("Expected to receive 10 messages", countDownLatch.await(5, TimeUnit.SECONDS));
+            jmsConsumer.close();
+            jmsContext.stop();
+
+            // Check that exactly 10 messages have been received 
+            assertEquals(Integer.valueOf(10), Integer.valueOf(recvMessages.size()));
+
+            int validatedCount = 0;
+            for(int validatedPriority : VALID_PRIORITIES) {
+                for(javax.jms.Message tmpMessage : recvMessages) {
+                    if(tmpMessage.getJMSPriority() == validatedPriority) {
+                        MessageData messageData = new MessageData();
+                        messageData.setMessageType(messageType).setMessagePayload(messagePayload).setPriority(validatedPriority);
+                        ActiveMQJMS2TestSupport.validateMessageData(tmpMessage, messageData);
+                        validatedCount++;
+                    }
+                }
+            }
+            assertEquals(Integer.valueOf(10), Integer.valueOf(validatedCount));
+        } catch (Exception e) {
+            fail(e.getMessage());
+        }
+    }
+
+    @Test
+    public void testMessagePriorityInvalidLower() {
+
+        try(JMSContext jmsContext = activemqConnectionFactory.createContext(DEFAULT_JMS_USER, DEFAULT_JMS_PASS, Session.AUTO_ACKNOWLEDGE)) {
+            assertNotNull(jmsContext);
+            Destination destination = ActiveMQJMS2TestSupport.generateDestination(jmsContext, destinationType, methodNameDestinationName);
+            assertNotNull(destination);
+
+            Message message = ActiveMQJMS2TestSupport.generateMessage(jmsContext, messageType, messagePayload);
+            MessageData sendMessageData = new MessageData();
+            sendMessageData.setMessage(message).setPriority(-1);
+            boolean caught = false;
+            try {
+                ActiveMQJMS2TestSupport.sendMessage(jmsContext, destination, sendMessageData);
+                fail("IlegalStateRuntimeException expected");
+            } catch (IllegalStateRuntimeException e) {
+                assertEquals("default priority must be a value between 0 and 9", e.getMessage());
+                caught = true;
+            }
+            assertTrue(caught);
+        } catch (Exception e) {
+            fail(e.getMessage());
+        }
+    }
+
+    @Test
+    public void testMessagePriorityInvalidHigher() {
+
+        try(JMSContext jmsContext = activemqConnectionFactory.createContext(DEFAULT_JMS_USER, DEFAULT_JMS_PASS, Session.AUTO_ACKNOWLEDGE)) {
+            assertNotNull(jmsContext);
+            Destination destination = ActiveMQJMS2TestSupport.generateDestination(jmsContext, destinationType, methodNameDestinationName);
+            assertNotNull(destination);
+
+            Message message = ActiveMQJMS2TestSupport.generateMessage(jmsContext, messageType, messagePayload);
+            MessageData sendMessageData = new MessageData();
+            sendMessageData.setMessage(message).setPriority(10);
+            boolean caught = false;
+            try {
+                ActiveMQJMS2TestSupport.sendMessage(jmsContext, destination, sendMessageData);
+                fail("IlegalStateRuntimeException expected");
+            } catch (IllegalStateRuntimeException e) {
+                assertEquals("default priority must be a value between 0 and 9", e.getMessage());
+                caught = true;
+            }
+            assertTrue(caught);
+        } catch (Exception e) {
+            fail(e.getMessage());
+        }
+    }
+
+    @Test
+    public void testMessageTimestampTimeToLive() {
+
+        long timeToLive = 900000l;
+
+        try(JMSContext jmsContext = activemqConnectionFactory.createContext(DEFAULT_JMS_USER, DEFAULT_JMS_PASS, Session.AUTO_ACKNOWLEDGE)) {
+            assertNotNull(jmsContext);
+            Destination destination = ActiveMQJMS2TestSupport.generateDestination(jmsContext, destinationType, methodNameDestinationName);
+            assertNotNull(destination);
+            JMSConsumer jmsConsumer = jmsContext.createConsumer(destination);
+            assertNotNull(jmsConsumer);
+
+            Message message = ActiveMQJMS2TestSupport.generateMessage(jmsContext, messageType, messagePayload);
+
+            List<String> sentMessageIds = new LinkedList<>();
+            long messageExpiration = Long.MIN_VALUE;
+            long messageTimestamp = Long.MIN_VALUE;
+
+            MessageData sendMessageData = new MessageData();
+            sendMessageData.setMessage(message).setTimeToLive(timeToLive);
+            sentMessageIds.add(ActiveMQJMS2TestSupport.sendMessage(jmsContext, destination, sendMessageData));
+            messageExpiration = message.getJMSExpiration();
+            messageTimestamp = message.getJMSTimestamp();
+            assertNotEquals(Long.MIN_VALUE, messageExpiration);
+            assertNotEquals(Long.MIN_VALUE, messageTimestamp);
+
+            final List<Message> recvMessages = new CopyOnWriteArrayList<>();
+            final CountDownLatch countDownLatch = new CountDownLatch(1);
+
+            jmsConsumer.setMessageListener(new MessageListener() {
+                @Override
+                public void onMessage(Message message) {
+                    recvMessages.add(message);
+                    countDownLatch.countDown();
+                }
+            });
+            jmsContext.start();
+
+            assertTrue("Expected to receive 1 message", countDownLatch.await(5l, TimeUnit.SECONDS));
+            jmsConsumer.close();
+            jmsContext.stop();
+
+            // Check that exactly 1 messages have been received 
+            assertEquals(Integer.valueOf(1), Integer.valueOf(recvMessages.size()));
+
+            int validatedCount = 0;
+            for(javax.jms.Message tmpMessage : recvMessages) {
+                MessageData recvMessageData = new MessageData();
+                recvMessageData.setMessageType(messageType).setMessagePayload(messagePayload).setExpiration(messageExpiration).setTimestamp(messageTimestamp);
+                ActiveMQJMS2TestSupport.validateMessageData(tmpMessage, recvMessageData);
+                validatedCount++;
+            }
+            assertEquals(Integer.valueOf(1), Integer.valueOf(validatedCount));
+        } catch (Exception e) {
+            fail(e.getMessage());
+        }
+    }
+
+    @Test
+    public void testMessageDisableTimestamp() {
+
+        long timeToLive = 900000l;
+
+        try(JMSContext jmsContext = activemqConnectionFactory.createContext(DEFAULT_JMS_USER, DEFAULT_JMS_PASS, Session.AUTO_ACKNOWLEDGE)) {
+            assertNotNull(jmsContext);
+            Destination destination = ActiveMQJMS2TestSupport.generateDestination(jmsContext, destinationType, methodNameDestinationName);
+            assertNotNull(destination);
+            JMSConsumer jmsConsumer = jmsContext.createConsumer(destination);
+            assertNotNull(jmsConsumer);
+
+            Message message = ActiveMQJMS2TestSupport.generateMessage(jmsContext, messageType, messagePayload);
+
+            List<String> sentMessageIds = new LinkedList<>();
+            long messageExpiration = Long.MIN_VALUE;
+            long messageTimestamp = Long.MIN_VALUE;
+            MessageData sendMessageData = new MessageData();
+            sendMessageData.setMessage(message).setTimeToLive(timeToLive).setDisableMessageTimestamp(true);
+            sentMessageIds.add(ActiveMQJMS2TestSupport.sendMessage(jmsContext, destination, sendMessageData));
+            messageExpiration = message.getJMSExpiration();
+            messageTimestamp = message.getJMSTimestamp();
+            assertEquals(0l, messageTimestamp);
+            assertNotEquals(Long.MIN_VALUE, messageExpiration);
+
+            final List<Message> recvMessages = new CopyOnWriteArrayList<>();
+            final CountDownLatch countDownLatch = new CountDownLatch(1);
+
+            jmsConsumer.setMessageListener(new MessageListener() {
+                @Override
+                public void onMessage(Message message) {
+                    recvMessages.add(message);
+                    countDownLatch.countDown();
+                }
+            });
+            jmsContext.start();
+
+            assertTrue("Expected to receive 1 message", countDownLatch.await(5l, TimeUnit.SECONDS));
+            jmsConsumer.close();
+            jmsContext.stop();
+
+            for(javax.jms.Message tmpMessage : recvMessages) {
+                MessageData recvMessageData = new MessageData();
+                recvMessageData.setMessagePayload(messagePayload).setMessageType(messageType).setExpiration(messageExpiration).setTimestamp(messageTimestamp).setDisableMessageTimestamp(true);
+                ActiveMQJMS2TestSupport.validateMessageData(tmpMessage, recvMessageData);
+            }
+        } catch (Exception e) {
+            fail(e.getMessage());
+        }
+    }
+
+    @Test
+    public void testMessageNonQOSHeaders() {
+
+        String jmsCorrelationID = UUID.randomUUID().toString();
+
+        try(JMSContext jmsContext = activemqConnectionFactory.createContext(DEFAULT_JMS_USER, DEFAULT_JMS_PASS, Session.AUTO_ACKNOWLEDGE)) {
+            assertNotNull(jmsContext);
+            Destination destination = ActiveMQJMS2TestSupport.generateDestination(jmsContext, destinationType, methodNameDestinationName);
+            assertNotNull(destination);
+            JMSConsumer jmsConsumer = jmsContext.createConsumer(destination);
+            assertNotNull(jmsConsumer);
+
+            Message message = ActiveMQJMS2TestSupport.generateMessage(jmsContext, messageType, messagePayload);
+            Destination jmsReplyTo = ActiveMQJMS2TestSupport.generateDestination(jmsContext, destinationType, methodNameDestinationName + ".REPLYTO");
+            String jmsType = message.getClass().getName(); 
+
+            List<String> sentMessageIds = new LinkedList<>();
+            MessageData sendMessageData = new MessageData();
+            sendMessageData.setMessage(message).setCorrelationID(jmsCorrelationID).setReplyTo(jmsReplyTo).setJMSType(jmsType);
+            sentMessageIds.add(ActiveMQJMS2TestSupport.sendMessage(jmsContext, destination, sendMessageData));
+
+            final List<Message> recvMessages = new CopyOnWriteArrayList<>();
+            final CountDownLatch countDownLatch = new CountDownLatch(1);
+
+            jmsConsumer.setMessageListener(new MessageListener() {
+                @Override
+                public void onMessage(Message message) {
+                    recvMessages.add(message);
+                    countDownLatch.countDown();
+                }
+            });
+            jmsContext.start();
+
+            assertTrue("Expected to receive 1 message", countDownLatch.await(5l, TimeUnit.SECONDS));
+            jmsConsumer.close();
+            jmsContext.stop();
+
+            for(javax.jms.Message tmpMessage : recvMessages) {
+                MessageData recvMessageData = new MessageData();
+                recvMessageData.setMessagePayload(messagePayload).setMessageType(messageType).setCorrelationID(jmsCorrelationID).setReplyTo(jmsReplyTo).setJMSType(jmsType);
+                ActiveMQJMS2TestSupport.validateMessageData(tmpMessage, recvMessageData);
+            }
+        } catch (Exception e) {
+            fail(e.getMessage());
+        }
+    }   
+
+    @Test
+    public void testMessageDisableMessageID() {
+
+        try(JMSContext jmsContext = activemqConnectionFactory.createContext(DEFAULT_JMS_USER, DEFAULT_JMS_PASS, Session.AUTO_ACKNOWLEDGE)) {
+            assertNotNull(jmsContext);
+            Destination destination = ActiveMQJMS2TestSupport.generateDestination(jmsContext, destinationType, methodNameDestinationName);
+            assertNotNull(destination);
+            JMSConsumer jmsConsumer = jmsContext.createConsumer(destination);
+            assertNotNull(jmsConsumer);
+
+            Message message = ActiveMQJMS2TestSupport.generateMessage(jmsContext, messageType, messagePayload);
+            String jmsMessageID = (ActiveMQJMS2TestSupport.sendMessage(jmsContext, destination, message));
+
+            final List<Message> recvMessages = new CopyOnWriteArrayList<>();
+            final CountDownLatch countDownLatch = new CountDownLatch(1);
+
+            jmsConsumer.setMessageListener(new MessageListener() {
+                @Override
+                public void onMessage(Message message) {
+                    recvMessages.add(message);
+                    countDownLatch.countDown();
+                }
+            });
+            jmsContext.start();
+
+            assertTrue("Expected to receive 1 message", countDownLatch.await(5l, TimeUnit.SECONDS));
+            jmsConsumer.close();
+            jmsContext.stop();
+
+            for(javax.jms.Message tmpMessage : recvMessages) {
+                MessageData messageData = new MessageData();
+                messageData.setMessageType(messageType).setMessagePayload(messagePayload).setMessageID(jmsMessageID);
+                ActiveMQJMS2TestSupport.validateMessageData(tmpMessage, messageData);
+            }
+        } catch (Exception e) {
+            fail(e.getMessage());
+        }
+    }
+
+    @Test
+    public void testTopicDurableSubscriber() {
+        // Skip if this is not a topic test run
+        if(!"topic".equals(destinationType)) {
+            return;
+        }
+
+        try(JMSContext jmsContext = activemqConnectionFactory.createContext(DEFAULT_JMS_USER, DEFAULT_JMS_PASS, Session.AUTO_ACKNOWLEDGE)) {
+            assertNotNull(jmsContext);
+            jmsContext.setClientID(clientID);
+
+            Destination destination = ActiveMQJMS2TestSupport.generateDestination(jmsContext, destinationType, methodNameDestinationName);
+            assertNotNull(destination);
+            assertTrue(destination instanceof Topic);
+            Topic topic = Topic.class.cast(destination);
+
+            JMSConsumer jmsConsumerRegister = jmsContext.createDurableConsumer(topic, methodNameDestinationName);
+            assertNotNull(jmsConsumerRegister);
+            jmsContext.start();
+            jmsConsumerRegister.close();
+
+            Message message = ActiveMQJMS2TestSupport.generateMessage(jmsContext, messageType, messagePayload);
+            String jmsMessageID = (ActiveMQJMS2TestSupport.sendMessage(jmsContext, destination, message));
+
+            JMSConsumer jmsConsumerDurable = jmsContext.createDurableConsumer(topic, methodNameDestinationName);
+            assertNotNull(jmsConsumerDurable);
+
+            final List<Message> recvMessages = new CopyOnWriteArrayList<>();
+            final CountDownLatch countDownLatch = new CountDownLatch(1);
+            jmsConsumerDurable.setMessageListener(new MessageListener() {
+                @Override
+                public void onMessage(Message message) {
+                    recvMessages.add(message);
+                    countDownLatch.countDown();
+                }
+            });
+
+            assertTrue("Expected to receive 1 message", countDownLatch.await(5l, TimeUnit.SECONDS));
+            jmsConsumerDurable.close();
+            jmsContext.stop();
+
+            for(javax.jms.Message tmpMessage : recvMessages) {
+                MessageData messageData = new MessageData();
+                messageData.setMessageType(messageType).setMessagePayload(messagePayload).setMessageID(jmsMessageID);
+                ActiveMQJMS2TestSupport.validateMessageData(tmpMessage, messageData);
+            }
+        } catch (Exception e) {
+            fail(e.getMessage());
+        }
+    }
+
+    @Test
+    public void testTopicDurableSubscriberSelector() {
+        // Skip if this is not a topic test run
+        if(!"topic".equals(destinationType)) {
+            return;
+        }
+
+        try(JMSContext jmsContext = activemqConnectionFactory.createContext(DEFAULT_JMS_USER, DEFAULT_JMS_PASS, Session.AUTO_ACKNOWLEDGE)) {
+            assertNotNull(jmsContext);
+            jmsContext.setClientID(clientID);
+
+            Destination destination = ActiveMQJMS2TestSupport.generateDestination(jmsContext, destinationType, methodNameDestinationName);
+            assertNotNull(destination);
+            assertTrue(destination instanceof Topic);
+            Topic topic = Topic.class.cast(destination);
+
+            JMSConsumer jmsConsumerRegister = jmsContext.createDurableConsumer(topic, methodNameDestinationName, "JMSPriority=7", false);
+            assertNotNull(jmsConsumerRegister);
+            assertEquals("JMSPriority=7", jmsConsumerRegister.getMessageSelector());
+            jmsContext.start();
+
+            jmsConsumerRegister.close();
+
+            Message message = ActiveMQJMS2TestSupport.generateMessage(jmsContext, messageType, messagePayload);
+
+            List<String> sentMessageIds = new LinkedList<>();
+            String matchMessageId = null;
+            for(int priority=0; priority<10; priority++) {
+                MessageData sendMessageData = new MessageData();
+                sendMessageData.setMessage(message).setPriority(priority);
+                String sentMessageId = ActiveMQJMS2TestSupport.sendMessage(jmsContext, destination, sendMessageData);
+                sentMessageIds.add(sentMessageId);
+                if(priority == 7) {
+                    matchMessageId = sentMessageId;
+                }
+            }
+
+            JMSConsumer jmsConsumerDurable = jmsContext.createDurableConsumer(topic, methodNameDestinationName, "JMSPriority=7", false);
+            assertNotNull(jmsConsumerDurable);
+            assertEquals("JMSPriority=7", jmsConsumerDurable.getMessageSelector());
+
+            final List<Message> recvMessages = new CopyOnWriteArrayList<>();
+            final CountDownLatch countDownLatch = new CountDownLatch(1);
+            jmsConsumerDurable.setMessageListener(new MessageListener() {
+                @Override
+                public void onMessage(Message message) {
+                    recvMessages.add(message);
+                    countDownLatch.countDown();
+                }
+            });
+
+            assertTrue("Expected to receive 1 message", countDownLatch.await(5l, TimeUnit.SECONDS));
+            jmsConsumerDurable.close();
+            jmsContext.stop();
+
+            for(javax.jms.Message tmpMessage : recvMessages) {
+                MessageData recvMessageData = new MessageData();
+                recvMessageData.setMessageType(messageType).setMessagePayload(messagePayload).setMessageID(matchMessageId);
+                ActiveMQJMS2TestSupport.validateMessageData(tmpMessage, recvMessageData);
+            }
+        } catch (Exception e) {
+            fail(e.getMessage());
+        }
+    }
+}
diff --git a/activemq-unit-tests/src/test/java/org/apache/activemq/jms2/ActiveMQJMS2ProducerTest.java b/activemq-unit-tests/src/test/java/org/apache/activemq/jms2/ActiveMQJMS2ProducerTest.java
new file mode 100644
index 0000000..84bc8c4
--- /dev/null
+++ b/activemq-unit-tests/src/test/java/org/apache/activemq/jms2/ActiveMQJMS2ProducerTest.java
@@ -0,0 +1,192 @@
+/**
+ * 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.jms2;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotEquals;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
+
+import java.time.ZonedDateTime;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Set;
+
+import javax.jms.Destination;
+import javax.jms.JMSConsumer;
+import javax.jms.JMSContext;
+import javax.jms.JMSException;
+import javax.jms.JMSProducer;
+import javax.jms.Session;
+import javax.jms.TextMessage;
+
+import org.apache.activemq.ActiveMQMessageProducerSupport;
+import org.junit.Test;
+
+public class ActiveMQJMS2ProducerTest extends ActiveMQJMS2TestBase {
+
+    private static final String PROPERTY_NAME_VALID="ValidName";
+    private static final String PROPERTY_VALUE_VALID="valid value";
+
+    private static final Set<String> PROPERTY_NAMES_INVALID = Set.of("contains+plus+sign", "123startswithnumber", "contains blank space", "has-dash", "with&amp");
+    private static final Set<Object> PROPERTY_VALUES_INVALID = Set.of(new HashSet<String>(), new HashMap<Integer, String>(), ZonedDateTime.now());
+
+    @Test
+    public void testInvalidPropertyNameNullEmpty() {
+        try(JMSContext jmsContext = activemqConnectionFactory.createContext(DEFAULT_JMS_USER, DEFAULT_JMS_PASS, Session.AUTO_ACKNOWLEDGE)) {
+            assertNotNull(jmsContext);
+            JMSProducer jmsProducer = jmsContext.createProducer();
+            jmsContext.start();
+
+            boolean caught = false;
+            try {
+                jmsProducer.setProperty(null, PROPERTY_VALUE_VALID);
+            } catch (Exception e) {
+                assertEquals("Invalid JMS property name must not be null or empty", e.getMessage());
+                caught = true;
+            }
+            assertTrue(caught);
+            caught = false;
+
+            try {
+                jmsProducer.setProperty("", PROPERTY_VALUE_VALID);
+            } catch (Exception e) {
+                assertEquals("Invalid JMS property name must not be null or empty", e.getMessage());
+                caught = true;
+            }
+            assertTrue(caught);
+            caught = false;
+        } catch (Exception e) {
+            fail(e.getMessage());
+        }
+    }
+
+    @Test
+    public void testInvalidPropertyNameKeyword() {
+        try(JMSContext jmsContext = activemqConnectionFactory.createContext(DEFAULT_JMS_USER, DEFAULT_JMS_PASS, Session.AUTO_ACKNOWLEDGE)) {
+            assertNotNull(jmsContext);
+            JMSProducer jmsProducer = jmsContext.createProducer();
+            jmsContext.start();
+
+            for(String propertyName : ActiveMQMessageProducerSupport.JMS_PROPERTY_NAMES_DISALLOWED) {
+                boolean caught = false;
+                try {
+                    jmsProducer.setProperty(propertyName, PROPERTY_VALUE_VALID);
+                } catch (Exception e) {
+                    assertEquals("Invalid JMS property: " + propertyName + " name is in disallowed list", e.getMessage());
+                    caught = true;
+                }
+                assertTrue("Expected exception for propertyName: " + propertyName, caught);
+            }
+        } catch (Exception e) {
+            fail(e.getMessage());
+        }
+    }
+
+    @Test
+    public void testInvalidPropertyNameInvalidJavaIdentifier() {
+        try(JMSContext jmsContext = activemqConnectionFactory.createContext(DEFAULT_JMS_USER, DEFAULT_JMS_PASS, Session.AUTO_ACKNOWLEDGE)) {
+            assertNotNull(jmsContext);
+            JMSProducer jmsProducer = jmsContext.createProducer();
+            jmsContext.start();
+
+            for(String propertyName : PROPERTY_NAMES_INVALID) {
+                boolean caught = false;
+                try {
+                    jmsProducer.setProperty(propertyName, PROPERTY_VALUE_VALID);
+                } catch (Exception e) {
+                    assertTrue(e.getMessage().startsWith("Invalid JMS property: " + propertyName + " name"));
+                    assertTrue(e.getMessage().contains("invalid character"));
+                    caught = true;
+                }
+                assertTrue("Expected exception for propertyName: " + propertyName, caught);
+            }
+        } catch (Exception e) {
+            fail(e.getMessage());
+        }
+    }
+
+    @Test
+    public void testInvalidPropertyNameInvalidValueClassType() {
+        try(JMSContext jmsContext = activemqConnectionFactory.createContext(DEFAULT_JMS_USER, DEFAULT_JMS_PASS, Session.AUTO_ACKNOWLEDGE)) {
+            assertNotNull(jmsContext);
+            JMSProducer jmsProducer = jmsContext.createProducer();
+            jmsContext.start();
+
+            for(Object propertyValue : PROPERTY_VALUES_INVALID) {
+                boolean caught = false;
+                try {
+                    jmsProducer.setProperty(PROPERTY_NAME_VALID, propertyValue);
+                } catch (Exception e) {
+                    assertEquals("Invalid JMS property: " + PROPERTY_NAME_VALID + " value class: " + propertyValue.getClass().getName() + " is not permitted by specification", e.getMessage());
+                    caught = true;
+                }
+                assertTrue("Expected exception for propertyValue: " + propertyValue, caught);
+            }
+        } catch (Exception e) {
+            fail(e.getMessage());
+        }
+    }
+
+    @Test
+    public void testDisableTimestamp() {
+        try(JMSContext jmsContext = activemqConnectionFactory.createContext(DEFAULT_JMS_USER, DEFAULT_JMS_PASS, Session.AUTO_ACKNOWLEDGE)) {
+            assertNotNull(jmsContext);
+            Destination destination = jmsContext.createQueue(methodNameDestinationName);
+            JMSProducer jmsProducer1 = jmsContext.createProducer();
+            jmsProducer1.setDisableMessageTimestamp(true);
+            jmsContext.start();
+
+            String textBody = "Test-" + methodNameDestinationName;
+            jmsProducer1.send(destination, textBody);
+            verifyTimestamp(jmsContext, destination, textBody, true);
+
+            // Re-enable timestatmp
+            jmsProducer1.setDisableMessageTimestamp(false);
+            jmsProducer1.send(destination, textBody);
+            verifyTimestamp(jmsContext, destination, textBody, false);
+
+            // Create second producer from same Context
+            JMSProducer jmsProducer2 = jmsContext.createProducer();
+            jmsProducer2.setDisableMessageTimestamp(true);
+            jmsProducer2.send(destination, textBody);
+            verifyTimestamp(jmsContext, destination, textBody, true);
+
+            // Re-confirm jmsProducer1 setting remains
+            jmsProducer1.send(destination, textBody);
+            verifyTimestamp(jmsContext, destination, textBody, false);
+        } catch (Exception e) {
+            fail(e.getMessage());
+        }
+    }
+
+    protected static void verifyTimestamp(JMSContext jmsContext, Destination destination, String expectedTextBody, boolean expectTimestampZero) throws JMSException {
+        try(JMSConsumer jmsConsumer = jmsContext.createConsumer(destination)) {
+            javax.jms.Message message = jmsConsumer.receive(1000l);
+            assertNotNull(message);
+            assertTrue(TextMessage.class.isAssignableFrom(message.getClass()));
+            assertEquals(expectedTextBody, TextMessage.class.cast(message).getText());
+
+            if(expectTimestampZero) {
+                assertEquals(Long.valueOf(0), Long.valueOf(message.getJMSTimestamp()));
+            } else {
+                assertNotEquals(Long.valueOf(0), Long.valueOf(message.getJMSTimestamp()));
+            }
+        }
+    }
+}
diff --git a/activemq-unit-tests/src/test/java/org/apache/activemq/jms2/ActiveMQJMS2TestBase.java b/activemq-unit-tests/src/test/java/org/apache/activemq/jms2/ActiveMQJMS2TestBase.java
new file mode 100644
index 0000000..d15af01
--- /dev/null
+++ b/activemq-unit-tests/src/test/java/org/apache/activemq/jms2/ActiveMQJMS2TestBase.java
@@ -0,0 +1,141 @@
+/**
+ * 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.jms2;
+
+import java.lang.management.ManagementFactory;
+import java.util.LinkedList;
+import java.util.List;
+import javax.jms.Connection;
+import javax.jms.MessageProducer;
+import javax.jms.Session;
+import javax.management.JMX;
+import javax.management.MBeanServer;
+import org.apache.activemq.ActiveMQConnectionFactory;
+import org.apache.activemq.broker.jmx.BrokerMBeanSupport;
+import org.apache.activemq.broker.jmx.DestinationViewMBean;
+import org.apache.activemq.broker.jmx.QueueViewMBean;
+import org.apache.activemq.broker.jmx.TopicViewMBean;
+import org.apache.activemq.command.ActiveMQDestination;
+import org.junit.After;
+import org.junit.AfterClass;
+import org.junit.Before;
+import org.junit.BeforeClass;
+import org.junit.Rule;
+import org.junit.rules.TestName;
+
+public abstract class ActiveMQJMS2TestBase {
+
+    public static final String DEFAULT_JMX_DOMAIN_NAME = "org.apache.activemq";
+    public static final String DEFAULT_JMX_BROKER_NAME = "localhost";
+
+    public static final String DEFAULT_JMS_USER = "admin";
+    public static final String DEFAULT_JMS_PASS = "admin";
+
+    protected static ActiveMQConnectionFactory activemqConnectionFactory = null;
+
+    @Rule public TestName testName = new TestName();
+
+    // Control session
+    protected Connection connection = null;
+    protected Session session = null;
+    protected MessageProducer messageProducer = null;
+
+    protected String methodNameDestinationName = null;
+    protected MBeanServer mbeanServer = null;
+
+    @BeforeClass
+    public static void setUpClass() {
+        activemqConnectionFactory = new ActiveMQConnectionFactory("vm://localhost?marshal=false&broker.persistent=false");
+        List<String> newTrustedPackages = new LinkedList<>();
+        newTrustedPackages.addAll(activemqConnectionFactory.getTrustedPackages());
+        newTrustedPackages.add(ActiveMQJMS2TestBase.class.getPackageName());
+        activemqConnectionFactory.setTrustedPackages(newTrustedPackages);
+    }
+
+    @AfterClass
+    public static void tearDownClass() {
+        activemqConnectionFactory = null;
+    }
+
+    @Before
+    public void setUp() throws Exception {
+        connection = activemqConnectionFactory.createConnection();
+        connection.start();
+        session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
+        methodNameDestinationName = "AMQ.JMS2." + cleanParameterizedMethodName(testName.getMethodName().toUpperCase());
+        messageProducer = session.createProducer(session.createQueue(methodNameDestinationName));
+        mbeanServer = ManagementFactory.getPlatformMBeanServer();
+    }
+
+    @After
+    public void tearDown() {
+        if(messageProducer != null) {
+            try { messageProducer.close(); } catch (Exception e) { } finally { messageProducer = null; }
+        }
+
+        if(session != null) {
+            try { session.close(); } catch (Exception e) { } finally { session = null; }
+        }
+
+        if(connection != null) {
+            try { connection.close(); } catch (Exception e) { } finally { connection = null; }
+        }
+
+        methodNameDestinationName = null;
+    }
+
+    protected DestinationViewMBean getDestinationViewMBean(String destinationType, ActiveMQDestination destination) throws Exception {
+        switch(destinationType) {
+        case "queue": return getQueueViewMBean(destination);
+        case "topic": return getTopicViewMBean(destination);
+        case "temp-queue": return getTempQueueViewMBean(destination);
+        case "temp-topic": return getTempTopicViewMBean(destination);
+        default: throw new IllegalStateException("Unsupported destinationType: " + destinationType);
+        }
+    }
+
+    protected QueueViewMBean getQueueViewMBean(ActiveMQDestination destination) throws Exception {
+        return JMX.newMBeanProxy(mbeanServer, BrokerMBeanSupport.createDestinationName(BrokerMBeanSupport.createBrokerObjectName(DEFAULT_JMX_DOMAIN_NAME, DEFAULT_JMX_BROKER_NAME).toString(), destination), QueueViewMBean.class);
+    }
+
+    protected TopicViewMBean getTopicViewMBean(ActiveMQDestination destination) throws Exception {
+        return JMX.newMBeanProxy(mbeanServer, BrokerMBeanSupport.createDestinationName(BrokerMBeanSupport.createBrokerObjectName(DEFAULT_JMX_DOMAIN_NAME, DEFAULT_JMX_BROKER_NAME).toString(), destination), TopicViewMBean.class);
+    }
+
+    protected TopicViewMBean getTempQueueViewMBean(ActiveMQDestination destination) throws Exception {
+        return JMX.newMBeanProxy(mbeanServer, BrokerMBeanSupport.createDestinationName(BrokerMBeanSupport.createBrokerObjectName(DEFAULT_JMX_DOMAIN_NAME, DEFAULT_JMX_BROKER_NAME).toString(), destination), TopicViewMBean.class);
+    }
+
+    protected TopicViewMBean getTempTopicViewMBean(ActiveMQDestination destination) throws Exception {
+        return JMX.newMBeanProxy(mbeanServer, BrokerMBeanSupport.createDestinationName(BrokerMBeanSupport.createBrokerObjectName(DEFAULT_JMX_DOMAIN_NAME, DEFAULT_JMX_BROKER_NAME).toString(), destination), TopicViewMBean.class);
+    }
+
+    private static String cleanParameterizedMethodName(String methodName) {
+        // clean up parameterized method string: TESTMESSAGETIMESTAMPTIMETOLIVE[DESTINATIONTYPE=QUEUE, MESSAGETYPE=BYTES]
+        // returns: TESTMESSAGETIMESTAMPTIMETOLIVE.QUEUE.BYTES
+
+        if(methodName == null || (!methodName.contains("[") && !methodName.contains("]"))) {
+            return methodName;
+        }
+
+        String[] step1 = methodName.split("\\[", 2);
+        String[] step2 = step1[1].split("\\]", 2);
+        String[] step3 = step2[0].split(",", 16);
+
+        return step1[0] + "." + step3[0].split("=", 2)[1] + "." + step3[1].split("=", 2)[1];
+    }
+}
diff --git a/activemq-unit-tests/src/test/java/org/apache/activemq/jms2/ActiveMQJMS2TestObjectMessagePayload.java b/activemq-unit-tests/src/test/java/org/apache/activemq/jms2/ActiveMQJMS2TestObjectMessagePayload.java
new file mode 100644
index 0000000..dc3bd86
--- /dev/null
+++ b/activemq-unit-tests/src/test/java/org/apache/activemq/jms2/ActiveMQJMS2TestObjectMessagePayload.java
@@ -0,0 +1,34 @@
+/**
+ * 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.jms2;
+
+import java.io.Serializable;
+
+public class ActiveMQJMS2TestObjectMessagePayload implements Serializable {
+
+    private static final long serialVersionUID = 1L;
+
+    private String payload;
+
+    public ActiveMQJMS2TestObjectMessagePayload(String payload) {
+        this.payload = payload;
+    }
+
+    public String getPayload() {
+        return payload;
+    }
+}
diff --git a/activemq-unit-tests/src/test/java/org/apache/activemq/jms2/ActiveMQJMS2TestSupport.java b/activemq-unit-tests/src/test/java/org/apache/activemq/jms2/ActiveMQJMS2TestSupport.java
new file mode 100644
index 0000000..6008f97
--- /dev/null
+++ b/activemq-unit-tests/src/test/java/org/apache/activemq/jms2/ActiveMQJMS2TestSupport.java
@@ -0,0 +1,307 @@
+/**
+ * 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.jms2;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
+
+import java.nio.charset.StandardCharsets;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.Enumeration;
+import java.util.HashSet;
+import java.util.Set;
+
+import javax.jms.BytesMessage;
+import javax.jms.Destination;
+import javax.jms.JMSContext;
+import javax.jms.JMSException;
+import javax.jms.JMSProducer;
+import javax.jms.MapMessage;
+import javax.jms.Message;
+import javax.jms.ObjectMessage;
+import javax.jms.QueueBrowser;
+import javax.jms.StreamMessage;
+import javax.jms.TextMessage;
+
+public class ActiveMQJMS2TestSupport {
+
+    protected static final Set<String> PROPERTY_NAMES = Collections.unmodifiableSet(new HashSet<>(Arrays.asList("JMS2_BOOLEAN_MIN", "JMS2_BOOLEAN_MAX", "JMS2_BYTE_MIN", "JMS2_BYTE_MAX",
+                                                        "JMS2_DOUBLE_MIN", "JMS2_DOUBLE_MAX", "JMS2_INT_MIN", "JMS2_INT_MAX", "JMS2_FLOAT_MIN", "JMS2_FLOAT_MAX", "JMS2_LONG_MIN",
+                                                        "JMS2_LONG_MAX", "JMS2_SHORT_MIN", "JMS2_SHORT_MAX", "JMS2_STRING_VAL")));
+
+    private ActiveMQJMS2TestSupport() {}
+
+    protected static Destination generateDestination(JMSContext jmsContext, String destinationType, String destinationName) throws JMSException {
+        Destination destination = null;
+        switch(destinationType) {
+        case "queue":
+            destination = jmsContext.createQueue(destinationName);
+            break;
+        case "topic":
+            destination = jmsContext.createTopic(destinationName);
+            break;
+        case "temp-queue":
+            destination = jmsContext.createTemporaryQueue();
+            break;
+        case "temp-topic":
+            destination = jmsContext.createTemporaryTopic();
+            break;
+        default:
+            fail("Unsupported destinationType:" + destinationType);
+        }
+       assertNotNull(destination);
+       return destination;
+    }
+
+    protected static Message generateMessage(JMSContext jmsContext, String messageType, String payload) throws JMSException {
+        assertNotNull(messageType);
+        Message tmpMessage = null;
+        switch(messageType) {
+        case "bytes":
+            BytesMessage bytesMessage = jmsContext.createBytesMessage();
+            bytesMessage.writeBytes(payload.getBytes(StandardCharsets.UTF_8));
+            tmpMessage = bytesMessage;
+            break;
+        case "map":
+            MapMessage mapMessage = jmsContext.createMapMessage();
+            mapMessage.setString("payload", payload);
+            tmpMessage = mapMessage;
+            break;
+        case "object":
+            tmpMessage = jmsContext.createObjectMessage(new ActiveMQJMS2TestObjectMessagePayload(payload));
+            break;
+        case "stream":
+            StreamMessage streamMessage = jmsContext.createStreamMessage();
+            streamMessage.writeString(payload);
+            tmpMessage = streamMessage;
+            break;
+        case "text":
+            tmpMessage = jmsContext.createTextMessage(payload);
+            break;
+        default:
+            fail("Unsupported messageType:" + messageType);
+        }
+        return tmpMessage;
+    }
+
+    protected static void populateJMSHeaders(javax.jms.Message message, String correlationID, Destination replyTo, String jmsType) throws JMSException {
+        assertNotNull(message);
+        message.setJMSCorrelationID(null);
+        message.setJMSReplyTo(null);
+        message.setJMSType(null);
+    }
+
+    protected static String sendMessage(JMSContext jmsContext, Destination destination, Message message) throws JMSException {
+        MessageData messageData = new MessageData();
+        messageData.setMessage(message);
+        return sendMessage(jmsContext, destination, messageData);
+    }
+
+    protected static String sendMessage(JMSContext jmsContext, Destination destination, MessageData messageData) throws JMSException {
+        assertNotNull(jmsContext);
+        assertNotNull(messageData);
+        assertNotNull(messageData.getMessage());
+
+        JMSProducer jmsProducer = jmsContext.createProducer();
+
+        if(messageData.getDeliveryDelay() != null) {
+            jmsProducer.setDeliveryDelay(messageData.getDeliveryDelay());
+        }
+        if(messageData.getDeliveryMode() != null) {
+            jmsProducer.setDeliveryMode(messageData.getDeliveryMode());
+        } 
+        if(messageData.getDisableMessageID() != null) {
+            jmsProducer.setDisableMessageID(messageData.getDisableMessageID());
+        }
+        if(messageData.getDisableMessageTimestamp() != null) {
+            jmsProducer.setDisableMessageTimestamp(messageData.getDisableMessageTimestamp());
+        }
+        if(messageData.getCorrelationID() != null) {
+            jmsProducer.setJMSCorrelationID(messageData.getCorrelationID());
+        }
+        if(messageData.getReplyTo() != null) {
+            jmsProducer.setJMSReplyTo(messageData.getReplyTo());
+        }
+        if(messageData.getJmsType() != null) {
+            jmsProducer.setJMSType(messageData.getJmsType());
+        }
+        if(messageData.getPriority() != null) {
+            jmsProducer.setPriority(messageData.getPriority());
+        }
+        if(messageData.getTimeToLive() != null) {
+            jmsProducer.setTimeToLive(messageData.getTimeToLive());
+        }
+        populateJMSProperties(jmsProducer);
+        validateJMSProperties(jmsProducer);
+        jmsProducer.send(destination, messageData.getMessage());
+
+        return messageData.getMessage().getJMSMessageID();
+    }
+
+    protected static void browseMessage(JMSContext jmsContext, String testDestinationName, String expectedTextBody, boolean expectFound) throws JMSException {
+        assertNotNull(jmsContext);
+        try(QueueBrowser queueBrowser = jmsContext.createBrowser(jmsContext.createQueue(testDestinationName))) {
+            Enumeration<?> messageEnumeration = queueBrowser.getEnumeration();
+            assertNotNull(messageEnumeration);
+            
+            boolean found = false; 
+            while(!found && messageEnumeration.hasMoreElements()) {
+                javax.jms.Message message = (Message)messageEnumeration.nextElement();
+                assertNotNull(message);
+                assertTrue(TextMessage.class.isAssignableFrom(message.getClass()));
+                assertEquals(expectedTextBody, TextMessage.class.cast(message).getText());
+                found = true;
+            }
+            assertEquals(expectFound, found);
+        }
+    }
+
+    protected static void validateMessageData(javax.jms.Message message, MessageData messageData) throws JMSException {
+        assertNotNull(message);
+        assertNotNull(messageData.getMessageType());
+        assertNotNull(messageData.getMessagePayload());
+        validateJMSHeaders(message, messageData);
+        validateJMSProperties(message);
+
+        switch(messageData.getMessageType()) {
+        case "bytes":
+            assertTrue(message instanceof BytesMessage);
+            BytesMessage bytesMessage = BytesMessage.class.cast(message);
+            byte[] payload = new byte[(int)bytesMessage.getBodyLength()];
+            bytesMessage.readBytes(payload);
+            assertEquals(messageData.getMessagePayload(), new String(payload, StandardCharsets.UTF_8));
+            break;
+        case "map":
+            assertTrue(message instanceof MapMessage);
+            MapMessage mapMessage = MapMessage.class.cast(message);
+            String mapPayload = mapMessage.getString("payload");
+            assertEquals(messageData.getMessagePayload(), mapPayload);
+            break;
+        case "object":
+            assertTrue(message instanceof ObjectMessage);
+            ObjectMessage objectMessage = ObjectMessage.class.cast(message);
+            Object tmpObject = objectMessage.getObject();
+            assertNotNull(tmpObject);
+            assertTrue(tmpObject instanceof ActiveMQJMS2TestObjectMessagePayload);
+            assertEquals(messageData.getMessagePayload(), ActiveMQJMS2TestObjectMessagePayload.class.cast(tmpObject).getPayload());
+            break;
+        case "stream":
+            assertTrue(message instanceof StreamMessage);
+            StreamMessage streamMessage = StreamMessage.class.cast(message);
+            assertEquals(messageData.getMessagePayload(), streamMessage.readString());
+            break;
+        case "text":
+            assertTrue(message instanceof TextMessage);
+            assertEquals(messageData.getMessagePayload(), TextMessage.class.cast(message).getText());
+            break;
+        default:
+            fail("Unsupported messageType:" + messageData.getMessageType());
+        }
+    }
+
+    private static void validateJMSHeaders(javax.jms.Message message, MessageData messageData) throws JMSException {
+        assertNotNull(message);
+        assertEquals(messageData.getCorrelationID(), message.getJMSCorrelationID());
+        if(messageData.getDeliveryMode() != null) {
+            assertEquals(messageData.getDeliveryMode(), Integer.valueOf(message.getJMSDeliveryMode()));
+        }
+        if(messageData.getDeliveryTime() != null) {
+            assertEquals(messageData.getDeliveryTime(), Long.valueOf(message.getJMSDeliveryTime()));
+        }
+        if(messageData.getExpiration() != null) {
+            assertEquals(messageData.getExpiration(), Long.valueOf(message.getJMSExpiration()));
+        }
+        if(messageData.getMessageID() != null) {
+            assertEquals(messageData.getMessageID(), message.getJMSMessageID());
+        }
+        if(messageData.getPriority() != null) {
+            assertEquals(messageData.getPriority(), Integer.valueOf(message.getJMSPriority()));
+        }
+        assertEquals(messageData.getReplyTo(), message.getJMSReplyTo());
+        if(messageData.getTimestamp() != null) {
+            assertEquals(messageData.getTimestamp(), Long.valueOf(message.getJMSTimestamp()));
+        }
+        if(Boolean.TRUE.equals(messageData.getDisableMessageTimestamp())) {
+            assertEquals(Long.valueOf(0l), Long.valueOf(message.getJMSTimestamp()));
+        }
+        assertEquals(messageData.getJmsType(), message.getJMSType());
+    }
+
+    private static void populateJMSProperties(JMSProducer jmsProducer) throws JMSException {
+        jmsProducer.setProperty("JMS2_BOOLEAN_MIN", false);
+        jmsProducer.setProperty("JMS2_BOOLEAN_MAX", true);
+        jmsProducer.setProperty("JMS2_BYTE_MIN", Byte.MIN_VALUE);
+        jmsProducer.setProperty("JMS2_BYTE_MAX", Byte.MAX_VALUE);
+        jmsProducer.setProperty("JMS2_DOUBLE_MIN", Double.MIN_VALUE);
+        jmsProducer.setProperty("JMS2_DOUBLE_MAX", Double.MAX_VALUE);
+        jmsProducer.setProperty("JMS2_INT_MIN", Integer.MIN_VALUE);
+        jmsProducer.setProperty("JMS2_INT_MAX", Integer.MAX_VALUE);
+        jmsProducer.setProperty("JMS2_FLOAT_MIN", Float.MIN_VALUE);
+        jmsProducer.setProperty("JMS2_FLOAT_MAX", Float.MAX_VALUE);
+        jmsProducer.setProperty("JMS2_LONG_MIN", Long.MIN_VALUE);
+        jmsProducer.setProperty("JMS2_LONG_MAX", Long.MAX_VALUE);
+        jmsProducer.setProperty("JMS2_SHORT_MIN", Short.MIN_VALUE);
+        jmsProducer.setProperty("JMS2_SHORT_MAX", Short.MAX_VALUE);
+        jmsProducer.setProperty("JMS2_STRING_VAL", "Hello World");
+    }
+
+    private static void validateJMSProperties(JMSProducer jmsProducer) throws JMSException {
+        assertNotNull(jmsProducer);
+        assertNotNull(jmsProducer.getPropertyNames());
+        assertEquals(Integer.valueOf(PROPERTY_NAMES.size()), Integer.valueOf(jmsProducer.getPropertyNames().size()));
+        for(String propertyName : PROPERTY_NAMES) {
+            assertTrue(jmsProducer.propertyExists(propertyName));
+        }
+        assertEquals(Boolean.FALSE, Boolean.valueOf(jmsProducer.getBooleanProperty("JMS2_BOOLEAN_MIN")));
+        assertEquals(Boolean.TRUE, Boolean.valueOf(jmsProducer.getBooleanProperty("JMS2_BOOLEAN_MAX")));
+        assertEquals(Byte.valueOf(Byte.MIN_VALUE), Byte.valueOf(jmsProducer.getByteProperty("JMS2_BYTE_MIN")));
+        assertEquals(Byte.valueOf(Byte.MAX_VALUE), Byte.valueOf(jmsProducer.getByteProperty("JMS2_BYTE_MAX")));
+        assertEquals(Double.valueOf(Double.MIN_VALUE), Double.valueOf(jmsProducer.getDoubleProperty("JMS2_DOUBLE_MIN")));
+        assertEquals(Double.valueOf(Double.MAX_VALUE), Double.valueOf(jmsProducer.getDoubleProperty("JMS2_DOUBLE_MAX")));
+        assertEquals(Integer.valueOf(Integer.MIN_VALUE), Integer.valueOf(jmsProducer.getIntProperty("JMS2_INT_MIN")));
+        assertEquals(Integer.valueOf(Integer.MAX_VALUE), Integer.valueOf(jmsProducer.getIntProperty("JMS2_INT_MAX")));
+        assertEquals(Float.valueOf(Float.MIN_VALUE), Float.valueOf(jmsProducer.getFloatProperty("JMS2_FLOAT_MIN")));
+        assertEquals(Float.valueOf(Float.MAX_VALUE), Float.valueOf(jmsProducer.getFloatProperty("JMS2_FLOAT_MAX")));
+        assertEquals(Long.valueOf(Long.MIN_VALUE), Long.valueOf(jmsProducer.getLongProperty("JMS2_LONG_MIN")));
+        assertEquals(Long.valueOf(Long.MAX_VALUE), Long.valueOf(jmsProducer.getLongProperty("JMS2_LONG_MAX")));
+        assertEquals(Short.valueOf(Short.MIN_VALUE), Short.valueOf(jmsProducer.getShortProperty("JMS2_SHORT_MIN")));
+        assertEquals(Short.valueOf(Short.MAX_VALUE), Short.valueOf(jmsProducer.getShortProperty("JMS2_SHORT_MAX")));
+        assertEquals("Hello World", jmsProducer.getStringProperty("JMS2_STRING_VAL"));
+    }
+
+    private static void validateJMSProperties(javax.jms.Message message) throws JMSException {
+        assertNotNull(message);
+        assertEquals(Boolean.FALSE, Boolean.valueOf(message.getBooleanProperty("JMS2_BOOLEAN_MIN")));
+        assertEquals(Boolean.TRUE, Boolean.valueOf(message.getBooleanProperty("JMS2_BOOLEAN_MAX")));
+        assertEquals(Byte.valueOf(Byte.MIN_VALUE), Byte.valueOf(message.getByteProperty("JMS2_BYTE_MIN")));
+        assertEquals(Byte.valueOf(Byte.MAX_VALUE), Byte.valueOf(message.getByteProperty("JMS2_BYTE_MAX")));
+        assertEquals(Double.valueOf(Double.MIN_VALUE), Double.valueOf(message.getDoubleProperty("JMS2_DOUBLE_MIN")));
+        assertEquals(Double.valueOf(Double.MAX_VALUE), Double.valueOf(message.getDoubleProperty("JMS2_DOUBLE_MAX")));
+        assertEquals(Integer.valueOf(Integer.MIN_VALUE), Integer.valueOf(message.getIntProperty("JMS2_INT_MIN")));
+        assertEquals(Integer.valueOf(Integer.MAX_VALUE), Integer.valueOf(message.getIntProperty("JMS2_INT_MAX")));
+        assertEquals(Float.valueOf(Float.MIN_VALUE), Float.valueOf(message.getFloatProperty("JMS2_FLOAT_MIN")));
+        assertEquals(Float.valueOf(Float.MAX_VALUE), Float.valueOf(message.getFloatProperty("JMS2_FLOAT_MAX")));
+        assertEquals(Long.valueOf(Long.MIN_VALUE), Long.valueOf(message.getLongProperty("JMS2_LONG_MIN")));
+        assertEquals(Long.valueOf(Long.MAX_VALUE), Long.valueOf(message.getLongProperty("JMS2_LONG_MAX")));
+        assertEquals(Short.valueOf(Short.MIN_VALUE), Short.valueOf(message.getShortProperty("JMS2_SHORT_MIN")));
+        assertEquals(Short.valueOf(Short.MAX_VALUE), Short.valueOf(message.getShortProperty("JMS2_SHORT_MAX")));
+        assertEquals("Hello World", message.getStringProperty("JMS2_STRING_VAL"));
+    }
+}
diff --git a/activemq-unit-tests/src/test/java/org/apache/activemq/jms2/MessageData.java b/activemq-unit-tests/src/test/java/org/apache/activemq/jms2/MessageData.java
new file mode 100644
index 0000000..4e84bef
--- /dev/null
+++ b/activemq-unit-tests/src/test/java/org/apache/activemq/jms2/MessageData.java
@@ -0,0 +1,124 @@
+/**
+ * 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.jms2;
+
+import javax.jms.Destination;
+
+public class MessageData {
+
+    private javax.jms.Message message = null;
+    private String messageType = null;
+    private String messagePayload = null; 
+    private Long deliveryDelay = null;
+    private Integer deliveryMode = null;
+    private Boolean disableMessageID = null;
+    private Boolean disableMessageTimestamp = null;
+    private String correlationID = null;
+    private Destination replyTo = null;
+    private String jmsType = null;
+    private Integer priority = null;
+    private Long expiration = null;
+    private Long timestamp = null;
+    private Long deliveryTime = null;
+    private String messageID = null;
+    private Long timeToLive = null;
+
+    MessageData() {}
+
+    // builder-style setters
+    public MessageData setMessage(javax.jms.Message message) { this.message = message; return this; }
+    public MessageData setMessageType(String messageType) { this.messageType = messageType; return this; }
+    public MessageData setMessagePayload(String messagePayload) { this.messagePayload = messagePayload; return this; } 
+    public MessageData setDeliveryDelay(Long deliveryDelay) { this.deliveryDelay = deliveryDelay; return this; }
+    public MessageData setDeliveryMode(Integer deliveryMode) { this.deliveryMode = deliveryMode; return this; }
+    public MessageData setDisableMessageID(Boolean disableMessageID) { this.disableMessageID = disableMessageID; return this; }
+    public MessageData setDisableMessageTimestamp(Boolean disableMessageTimestamp) { this.disableMessageTimestamp = disableMessageTimestamp; return this; }
+    public MessageData setCorrelationID(String correlationID) { this.correlationID = correlationID; return this; }
+    public MessageData setReplyTo(Destination replyTo) { this.replyTo = replyTo; return this; }
+    public MessageData setJMSType(String jmsType) { this.jmsType = jmsType; return this; }
+    public MessageData setPriority(Integer priority) { this.priority = priority; return this; }
+    public MessageData setExpiration(Long expiration) { this.expiration = expiration; return this; }
+    public MessageData setTimestamp(Long timestamp) { this.timestamp = timestamp; return this; }
+    public MessageData setDeliveryTime(Long deliveryTime) { this.deliveryTime = deliveryTime; return this; }
+    public MessageData setMessageID(String messageID) { this.messageID = messageID; return this; }
+    public MessageData setTimeToLive(Long timeToLive) { this.timeToLive = timeToLive; return this; }
+
+    // standard getters
+    public String getMessagePayload() {
+        return messagePayload;
+    }
+
+    public Long getDeliveryDelay() {
+        return deliveryDelay;
+    }
+
+    public Integer getDeliveryMode() {
+        return deliveryMode;
+    }
+
+    public Boolean getDisableMessageID() {
+        return disableMessageID;
+    }
+
+    public Boolean getDisableMessageTimestamp() {
+        return disableMessageTimestamp;
+    }
+
+    public String getCorrelationID() {
+        return correlationID;
+    }
+
+    public Destination getReplyTo() {
+        return replyTo;
+    }
+
+    public String getJmsType() {
+        return jmsType;
+    }
+
+    public Integer getPriority() {
+        return priority;
+    }
+
+    public Long getExpiration() {
+        return expiration;
+    }
+
+    public Long getTimestamp() {
+        return timestamp;
+    }
+
+    public Long getDeliveryTime() {
+        return deliveryTime;
+    }
+
+    public String getMessageID() {
+        return messageID;
+    }
+
+    public javax.jms.Message getMessage() {
+        return message;
+    }
+
+    public String getMessageType() {
+        return messageType;
+    }
+
+    public Long getTimeToLive() {
+        return timeToLive;
+    }
+}
diff --git a/pom.xml b/pom.xml
index c552711..f01d967 100644
--- a/pom.xml
+++ b/pom.xml
@@ -1402,6 +1402,7 @@
             <exclude>**/kahadb/**/*.data</exclude>
             <exclude>**/resources/*.xsd</exclude>
             <exclude>**/src/test/resources/keystore</exclude>
+            <exclude>**/*.ts</exclude>
             <!-- web, web-console, web-demo files -->
             <exclude>**/webapp/mqtt/mqttws31.js</exclude>
             <exclude>**/webapp/js/dojo.js</exclude>
@@ -1461,9 +1462,9 @@
             <link>http://logging.apache.org/log4j/docs/api/</link>
           </links>
           <stylesheetfile>${basedir}/../etc/css/stylesheet.css</stylesheetfile>
-          <linksource>true</linksource>
           <maxmemory>2048m</maxmemory>
           <source>${source-version}</source>
+          <noindex>true</noindex>
           <additionalJOption>-J-Xmx2048m</additionalJOption>
           <!-- necessary for now under the javadocs can be fixed because jdk8 is much stricter -->
           <additionalJOption>${javadoc.options}</additionalJOption>
@@ -1592,6 +1593,9 @@
           <plugin>
             <groupId>org.apache.maven.plugins</groupId>
             <artifactId>maven-javadoc-plugin</artifactId>
+            <configuration>
+              <noindex>true</noindex>
+            </configuration>
             <executions>
               <execution>
                 <id>attach-javadocs</id>