diff --git a/client/src/main/java/org/apache/qpid/client/BasicMessageConsumer.java b/client/src/main/java/org/apache/qpid/client/BasicMessageConsumer.java
index adb79cf..d30f516 100644
--- a/client/src/main/java/org/apache/qpid/client/BasicMessageConsumer.java
+++ b/client/src/main/java/org/apache/qpid/client/BasicMessageConsumer.java
@@ -781,6 +781,7 @@
     protected void preDeliver(AbstractJMSMessage msg)
     {
         _session.setInRecovery(false);
+        msg.setAMQSession(_session);
 
         switch (_acknowledgeMode)
         {
@@ -793,9 +794,6 @@
                 _session.addUnacknowledgedMessage(msg.getDeliveryTag());
                 break;
             case Session.CLIENT_ACKNOWLEDGE:
-                // we set the session so that when the user calls acknowledge() it can call the method on session
-                // to send out the appropriate frame
-                msg.setAMQSession(_session);
                 _session.addUnacknowledgedMessage(msg.getDeliveryTag());
                 _session.markDirty();
                 break;
diff --git a/client/src/main/java/org/apache/qpid/client/message/AMQMessageDelegate_0_8.java b/client/src/main/java/org/apache/qpid/client/message/AMQMessageDelegate_0_8.java
index cb61770..8a9aec8 100644
--- a/client/src/main/java/org/apache/qpid/client/message/AMQMessageDelegate_0_8.java
+++ b/client/src/main/java/org/apache/qpid/client/message/AMQMessageDelegate_0_8.java
@@ -34,14 +34,18 @@
 import javax.jms.MessageNotWriteableException;
 import javax.jms.Queue;
 
+import org.apache.qpid.QpidException;
+import org.apache.qpid.client.AMQAnyDestination;
 import org.apache.qpid.client.AMQDestination;
 import org.apache.qpid.client.AMQQueue;
 import org.apache.qpid.client.AMQSession;
 import org.apache.qpid.client.AMQSession_0_8;
 import org.apache.qpid.client.AMQTopic;
 import org.apache.qpid.client.CustomJMSXProperty;
+import org.apache.qpid.exchange.ExchangeDefaults;
 import org.apache.qpid.framing.AMQShortString;
 import org.apache.qpid.framing.BasicContentHeaderProperties;
+import org.apache.qpid.messaging.Address;
 import org.apache.qpid.url.AMQBindingURL;
 import org.apache.qpid.url.BindingURL;
 
@@ -297,15 +301,17 @@
                     else if(replyToEncoding.contains("/"))
                     {
                         String[] parts = replyToEncoding.split("/",2);
-                        dest = new NonBURLReplyToDestination(parts[0], parts[1]);
-
-
+                        dest = new NonBURLReplyToDestination(ExchangeDefaults.DIRECT_EXCHANGE_CLASS, parts[0], parts[1]);
                     }
                     else
                     {
-                        if(getAMQSession().isQueueBound(replyToEncoding, null, null))
+                        if (getAMQSession().isQueueBound(null, replyToEncoding, null))
                         {
-                            dest = new NonBURLReplyToDestination(replyToEncoding, "");
+                            dest = new NonBURLReplyToDestination(ExchangeDefaults.DIRECT_EXCHANGE_CLASS, "", replyToEncoding);
+                        }
+                        else if (isExchangeExist(replyToEncoding))
+                        {
+                            dest = new NonBURLReplyToDestination(ExchangeDefaults.DIRECT_EXCHANGE_CLASS, replyToEncoding, "");
                         }
                         else
                         {
@@ -322,6 +328,32 @@
         }
     }
 
+    private boolean isExchangeExist(final String replyToEncoding)
+    {
+        try
+        {
+            AMQDestination amqDestination = new AMQDestination()
+            {
+                @Override
+                public String getAddressName()
+                {
+                    return replyToEncoding;
+                }
+
+                @Override
+                public boolean isNameRequired()
+                {
+                    return false;
+                }
+            };
+            return getAMQSession().isExchangeExist(amqDestination, false);
+        }
+        catch (Exception e)
+        {
+            return false;
+        }
+    }
+
     public void setJMSReplyTo(Destination destination) throws JMSException
     {
         if (destination == null)
@@ -660,7 +692,7 @@
         public DefaultRouterDestination(final String replyToEncoding)
         {
             super("",
-                  "direct",
+                  ExchangeDefaults.DIRECT_EXCHANGE_CLASS,
                   replyToEncoding,
                   replyToEncoding);
         }
@@ -682,10 +714,12 @@
     {
         private static final long serialVersionUID = 122897705932489259L;
 
-        public NonBURLReplyToDestination(final String exchange, final String routingKey)
+        public NonBURLReplyToDestination(final String exchangeClass,
+                                         final String exchange,
+                                         final String routingKey)
         {
             super(exchange,
-                  null,
+                  exchangeClass,
                   routingKey,
                   routingKey);
         }
diff --git a/client/src/main/java/org/apache/qpid/client/message/AbstractJMSMessage.java b/client/src/main/java/org/apache/qpid/client/message/AbstractJMSMessage.java
index 44681ac..d2cd007 100644
--- a/client/src/main/java/org/apache/qpid/client/message/AbstractJMSMessage.java
+++ b/client/src/main/java/org/apache/qpid/client/message/AbstractJMSMessage.java
@@ -34,8 +34,6 @@
 {
 
 
-    /** If the acknowledge mode is CLIENT_ACKNOWLEDGE the session is required */
-
     private AMQMessageDelegate _delegate;
     private boolean _redelivered;
     private boolean _receivedFromServer;
@@ -398,12 +396,6 @@
 
 
 
-    /**
-     * The session is set when CLIENT_ACKNOWLEDGE mode is used so that the CHANNEL ACK can be sent when the user calls
-     * acknowledge()
-     *
-     * @param s the AMQ session that delivered this message
-     */
     public void setAMQSession(AMQSession s)
     {
         _delegate.setAMQSession(s);
diff --git a/client/src/test/java/org/apache/qpid/client/message/AbstractJMSMessageTest.java b/client/src/test/java/org/apache/qpid/client/message/AbstractJMSMessageTest.java
index 325e2b7..2bde82b 100644
--- a/client/src/test/java/org/apache/qpid/client/message/AbstractJMSMessageTest.java
+++ b/client/src/test/java/org/apache/qpid/client/message/AbstractJMSMessageTest.java
@@ -21,17 +21,49 @@
 package org.apache.qpid.client.message;
 
 
-import org.apache.qpid.test.utils.QpidTestCase;
+import static org.mockito.Matchers.anyBoolean;
+import static org.mockito.Matchers.argThat;
+import static org.mockito.Matchers.eq;
+import static org.mockito.Matchers.isNull;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+import java.util.ArrayList;
+import java.util.List;
 
 import javax.jms.JMSException;
 
+import org.mockito.ArgumentMatcher;
+
+import org.apache.qpid.QpidException;
+import org.apache.qpid.client.AMQDestination;
+import org.apache.qpid.client.AMQSession;
+import org.apache.qpid.framing.BasicContentHeaderProperties;
+import org.apache.qpid.framing.ContentBody;
+import org.apache.qpid.framing.ContentHeaderBody;
+import org.apache.qpid.test.utils.QpidTestCase;
+
 public class AbstractJMSMessageTest extends QpidTestCase
 {
 
+    private final AMQSession<?,?> _session = mock(AMQSession.class);
+    private final MessageFactoryRegistry _messageFactoryRegistry = MessageFactoryRegistry.newDefaultRegistry(_session);
+
+    public void testIncoming08ReplyTo() throws Exception
+    {
+        when(_session.isQueueBound(isNull(String.class), eq("knownQueue"), isNull(String.class))).thenReturn(true);
+        when(_session.isExchangeExist(isDestinationWithAddress("knownExchange"), anyBoolean())).thenReturn(true);
+
+        doReplyToTest("direct://amq.direct/knownQueue?routingkey='knownQueue'", "direct://amq.direct/knownQueue/knownQueue?routingkey='knownQueue'");
+        doReplyToTest("knownQueue", "direct:///knownQueue/knownQueue?routingkey='knownQueue'");
+        doReplyToTest("knownExchange", "direct://knownExchange//?routingkey=''");
+        doReplyToTest("news-service/sports", "direct://news-service/sports/sports?routingkey='sports'");
+    }
+
     public void testSetNullJMSReplyTo08() throws JMSException
     {
         JMSTextMessage message = new JMSTextMessage(AMQMessageDelegateFactory.FACTORY_0_8);
-        try 
+        try
         {
             message.setJMSReplyTo(null);
         }
@@ -44,7 +76,7 @@
     public void testSetNullJMSReplyTo10() throws JMSException
     {
         JMSTextMessage message = new JMSTextMessage(AMQMessageDelegateFactory.FACTORY_0_10);
-        try 
+        try
         {
             message.setJMSReplyTo(null);
         }
@@ -54,4 +86,37 @@
         }
     }
 
+    private void doReplyToTest(final String headerReplyTo, final String expectedReplyToAddress)
+            throws QpidException, JMSException
+    {
+        final ContentHeaderBody contentHeader = new ContentHeaderBody(new BasicContentHeaderProperties());
+        contentHeader.getProperties().setReplyTo(headerReplyTo);
+
+        final List<ContentBody> contentBodies = new ArrayList<>();
+        final AbstractJMSMessage message = _messageFactoryRegistry.createMessage(0, false,
+                                                                                 "amq.direct",
+                                                                                 "routingKey",
+                                                                                 contentHeader,
+                                                                                 contentBodies,
+                                                                                 null,
+                                                                                 null,
+                                                                                 0);
+        message.setAMQSession(_session);
+
+        assertNotNull(message.getJMSReplyTo());
+        assertEquals(expectedReplyToAddress, message.getJMSReplyTo().toString());
+    }
+
+    private AMQDestination isDestinationWithAddress(final String expectedAddress)
+    {
+        return argThat( new ArgumentMatcher<AMQDestination>()
+        {
+            @Override
+            public boolean matches(Object argument)
+            {
+                AMQDestination actual = (AMQDestination) argument;
+                return actual.getAddressName().equals(expectedAddress);
+            }
+        });
+    }
 }