Merge branch 'trunk' into qpid-2920

Conflicts:
	qpid/cpp/src/cluster.mk
	qpid/java/systests/src/main/java/org/apache/qpid/test/client/destination/AddressBasedDestinationTest.java

git-svn-id: https://svn.apache.org/repos/asf/qpid/branches/qpid-2920@1086439 13f79535-47bb-0310-9956-ffa450edef68

13 files changed, 456 insertions, 76 deletions

diff --git a/qpid/java/broker/src/main/java/org/apache/qpid/server/queue/SimpleAMQQueue.java b/qpid/java/broker/src/main/java/org/apache/qpid/server/queue/SimpleAMQQueue.java
index b003152db6..0e3f7b2625 100644
--- a/qpid/java/broker/src/main/java/org/apache/qpid/server/queue/SimpleAMQQueue.java
+++ b/qpid/java/broker/src/main/java/org/apache/qpid/server/queue/SimpleAMQQueue.java
@@ -1808,14 +1808,40 @@ public class SimpleAMQQueue implements AMQQueue, Subscription.StateListener
     }
 
 
+    /**
+     * Used by queue Runners to asynchronously deliver messages to consumers.
+     *
+     * A queue Runner is started whenever a state change occurs, e.g when a new
+     * message arrives on the queue and cannot be immediately delivered to a
+     * subscription (i.e. asynchronous delivery is required). Unless there are
+     * SubFlushRunners operating (due to subscriptions unsuspending) which are
+     * capable of accepting/delivering all messages then these messages would
+     * otherwise remain on the queue.
+     *
+     * processQueue should be running while there are messages on the queue AND
+     * there are subscriptions that can deliver them. If there are no
+     * subscriptions capable of delivering the remaining messages on the queue
+     * then processQueue should stop to prevent spinning.
+     *
+     * Since processQueue is runs in a fixed size Executor, it should not run
+     * indefinitely to prevent starving other tasks of CPU (e.g jobs to process
+     * incoming messages may not be able to be scheduled in the thread pool
+     * because all threads are working on clearing down large queues). To solve
+     * this problem, after an arbitrary number of message deliveries the
+     * processQueue job stops iterating, resubmits itself to the executor, and
+     * ends the current instance
+     *
+     * @param runner the Runner to schedule
+     * @throws AMQException
+     */
     private void processQueue(Runnable runner) throws AMQException
     {
         long stateChangeCount;
         long previousStateChangeCount = Long.MIN_VALUE;
         boolean deliveryIncomplete = true;
 
-        int extraLoops = 1;
-        long iterations = MAX_ASYNC_DELIVERIES;
+        boolean lastLoop = false;
+        int iterations = MAX_ASYNC_DELIVERIES;
 
         _asynchronousRunner.compareAndSet(runner, null);
 
@@ -1832,12 +1858,14 @@ public class SimpleAMQQueue implements AMQQueue, Subscription.StateListener
 
             if (previousStateChangeCount != stateChangeCount)
             {
-                extraLoops = 1;
+                //further asynchronous delivery is required since the
+                //previous loop. keep going if iteration slicing allows.
+                lastLoop = false;
             }
 
             previousStateChangeCount = stateChangeCount;
-            deliveryIncomplete = _subscriptionList.size() != 0;
-            boolean done;
+            boolean allSubscriptionsDone = true;
+            boolean subscriptionDone;
 
             SubscriptionList.SubscriptionNodeIterator subscriptionIter = _subscriptionList.iterator();
             //iterate over the subscribers and try to advance their pointer
@@ -1847,30 +1875,25 @@ public class SimpleAMQQueue implements AMQQueue, Subscription.StateListener
                 sub.getSendLock();
                 try
                 {
-
-                    done = attemptDelivery(sub);
-
-                    if (done)
+                    //attempt delivery. returns true if no further delivery currently possible to this sub
+                    subscriptionDone = attemptDelivery(sub);
+                    if (subscriptionDone)
                     {
-                        if (extraLoops == 0)
+                        //close autoClose subscriptions if we are not currently intent on continuing
+                        if (lastLoop && sub.isAutoClose())
                         {
-                            deliveryIncomplete = false;
-                            if (sub.isAutoClose())
-                            {
-                                unregisterSubscription(sub);
+                            unregisterSubscription(sub);
 
-                                sub.confirmAutoClose();
-                            }
-                        }
-                        else
-                        {
-                            extraLoops--;
+                            sub.confirmAutoClose();
                         }
                     }
                     else
                     {
+                        //this subscription can accept additional deliveries, so we must 
+                        //keep going after this (if iteration slicing allows it)
+                        allSubscriptionsDone = false;
+                        lastLoop = false;
                         iterations--;
-                        extraLoops = 1;
                     }
                 }
                 finally
@@ -1878,10 +1901,34 @@ public class SimpleAMQQueue implements AMQQueue, Subscription.StateListener
                     sub.releaseSendLock();
                 }
             }
+
+            if(allSubscriptionsDone && lastLoop)
+            {
+                //We have done an extra loop already and there are again
+                //again no further delivery attempts possible, only
+                //keep going if state change demands it.
+                deliveryIncomplete = false;
+            }
+            else if(allSubscriptionsDone)
+            {
+                //All subscriptions reported being done, but we have to do
+                //an extra loop if the iterations are not exhausted and
+                //there is still any work to be done
+                deliveryIncomplete = _subscriptionList.size() != 0;
+                lastLoop = true;
+            }
+            else
+            {
+                //some subscriptions can still accept more messages,
+                //keep going if iteration count allows.
+                lastLoop = false;
+                deliveryIncomplete = true;
+            }
+
             _asynchronousRunner.set(null);
         }
 
-        // If deliveries == 0 then the limitting factor was the time-slicing rather than available messages or credit
+        // If iterations == 0 then the limiting factor was the time-slicing rather than available messages or credit
         // therefore we should schedule this runner again (unless someone beats us to it :-) ).
         if (iterations == 0 && _asynchronousRunner.compareAndSet(null, runner))
         {
diff --git a/qpid/java/broker/src/main/java/org/apache/qpid/server/transport/ServerConnection.java b/qpid/java/broker/src/main/java/org/apache/qpid/server/transport/ServerConnection.java
index 75bd50e3a2..54cd709af3 100644
--- a/qpid/java/broker/src/main/java/org/apache/qpid/server/transport/ServerConnection.java
+++ b/qpid/java/broker/src/main/java/org/apache/qpid/server/transport/ServerConnection.java
@@ -88,8 +88,18 @@ public class ServerConnection extends Connection implements AMQConnectionModel,
                 _onOpenTask.run();    
             }
             _actor.message(ConnectionMessages.OPEN(getClientId(), "0-10", true, true));
+
+            getVirtualHost().getConnectionRegistry().registerConnection(this);
         }
-        
+
+        if (state == State.CLOSE_RCVD || state == State.CLOSED || state == State.CLOSING)
+        {
+            if(_virtualHost != null)
+            {
+                _virtualHost.getConnectionRegistry().deregisterConnection(this);
+            }
+        }
+
         if (state == State.CLOSED)
         {
             logClosed();
@@ -126,7 +136,6 @@ public class ServerConnection extends Connection implements AMQConnectionModel,
     public void setVirtualHost(VirtualHost virtualHost)
     {
         _virtualHost = virtualHost;
-        _virtualHost.getConnectionRegistry().registerConnection(this);
         
         initialiseStatistics();
     }
@@ -253,7 +262,6 @@ public class ServerConnection extends Connection implements AMQConnectionModel,
             // Ignore
         }
         close(replyCode, message);
-        getVirtualHost().getConnectionRegistry().deregisterConnection(this);
     }
 
     @Override
diff --git a/qpid/java/client/src/main/java/org/apache/qpid/client/AMQSession.java b/qpid/java/client/src/main/java/org/apache/qpid/client/AMQSession.java
index 5fc8d43e94..d518d7b2c4 100644
--- a/qpid/java/client/src/main/java/org/apache/qpid/client/AMQSession.java
+++ b/qpid/java/client/src/main/java/org/apache/qpid/client/AMQSession.java
@@ -3461,4 +3461,9 @@ public abstract class AMQSession<C extends BasicMessageConsumer, P extends Basic
     {
         return _closing.get()|| _connection.isClosing();
     }
+    
+    public boolean isDeclareExchanges()
+    {
+    	return DECLARE_EXCHANGES;
+    }
 }
diff --git a/qpid/java/client/src/main/java/org/apache/qpid/client/BasicMessageProducer_0_10.java b/qpid/java/client/src/main/java/org/apache/qpid/client/BasicMessageProducer_0_10.java
index 36f2dfb66f..62d1d1698c 100644
--- a/qpid/java/client/src/main/java/org/apache/qpid/client/BasicMessageProducer_0_10.java
+++ b/qpid/java/client/src/main/java/org/apache/qpid/client/BasicMessageProducer_0_10.java
@@ -72,12 +72,15 @@ public class BasicMessageProducer_0_10 extends BasicMessageProducer
     {
         if (destination.getDestSyntax() == DestSyntax.BURL)
         {
-            String name = destination.getExchangeName().toString();
-            ((AMQSession_0_10) getSession()).getQpidSession().exchangeDeclare
-                (name,
-                 destination.getExchangeClass().toString(),
-                 null, null,
-                 name.startsWith("amq.") ? Option.PASSIVE : Option.NONE);
+        	if (getSession().isDeclareExchanges())
+        	{
+	            String name = destination.getExchangeName().toString();
+	            ((AMQSession_0_10) getSession()).getQpidSession().exchangeDeclare
+	                (name,
+	                 destination.getExchangeClass().toString(),
+	                 null, null,
+	                 name.startsWith("amq.") ? Option.PASSIVE : Option.NONE);
+        	}
         }
         else
         {       
diff --git a/qpid/java/client/src/main/java/org/apache/qpid/client/messaging/address/AddressHelper.java b/qpid/java/client/src/main/java/org/apache/qpid/client/messaging/address/AddressHelper.java
index e454a8eee4..a53558757e 100644
--- a/qpid/java/client/src/main/java/org/apache/qpid/client/messaging/address/AddressHelper.java
+++ b/qpid/java/client/src/main/java/org/apache/qpid/client/messaging/address/AddressHelper.java
@@ -232,14 +232,9 @@ public class AddressHelper
     
     private boolean getDurability(Map map)
     {
-        if (map != null && map.get(DURABLE) != null)
-        {
-            return Boolean.parseBoolean((String)map.get(DURABLE));
-        }
-        else
-        {
-            return false;
-        }
+        Accessor access = new MapAccessor(map);
+        Boolean result = access.getBoolean(DURABLE);
+        return (result == null) ? false : result.booleanValue();
     }
 
     /**
diff --git a/qpid/java/common/src/main/java/org/apache/qpid/util/FileUtils.java b/qpid/java/common/src/main/java/org/apache/qpid/util/FileUtils.java
index 516204fbd3..1a57af9bf7 100644
--- a/qpid/java/common/src/main/java/org/apache/qpid/util/FileUtils.java
+++ b/qpid/java/common/src/main/java/org/apache/qpid/util/FileUtils.java
@@ -339,7 +339,7 @@ public class FileUtils
         }
 
         //else we have a source directory
-        if (!dst.isDirectory() && !dst.mkdir())
+        if (!dst.isDirectory() && !dst.mkdirs())
         {
             throw new UnableToCopyException("Unable to create destination directory");
         }
diff --git a/qpid/java/module.xml b/qpid/java/module.xml
index 80f577b018..6ff807d377 100644
--- a/qpid/java/module.xml
+++ b/qpid/java/module.xml
@@ -323,6 +323,8 @@
   </condition>
 
   <property name="jvm.args" value=""/>
+  <property name="broker.existing.qpid.work" value=""/>
+
   <target name="test" depends="build,compile-tests" if="module.test.src.exists"
          unless="${dontruntest}" description="execute unit tests">
 
@@ -346,6 +348,7 @@
       <sysproperty key="broker" value="${broker}"/>
       <sysproperty key="broker.clean" value="${broker.clean}"/>
       <sysproperty key="broker.clean.between.tests" value="${broker.clean.between.tests}"/>
+      <sysproperty key="broker.existing.qpid.work" value="${broker.existing.qpid.work}"/>
       <sysproperty key="broker.persistent" value="${broker.persistent}"/>
       <sysproperty key="broker.version" value="${broker.version}"/>
       <sysproperty key="broker.ready" value="${broker.ready}" />
diff --git a/qpid/java/systests/src/main/java/org/apache/qpid/server/queue/MultipleTransactedBatchProducerTest.java b/qpid/java/systests/src/main/java/org/apache/qpid/server/queue/MultipleTransactedBatchProducerTest.java
new file mode 100644
index 0000000000..460270e188
--- /dev/null
+++ b/qpid/java/systests/src/main/java/org/apache/qpid/server/queue/MultipleTransactedBatchProducerTest.java
@@ -0,0 +1,246 @@
+/*
+ *
+ * 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.qpid.server.queue;
+
+import java.util.Random;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+
+import javax.jms.Connection;
+import javax.jms.JMSException;
+import javax.jms.Message;
+import javax.jms.MessageConsumer;
+import javax.jms.MessageListener;
+import javax.jms.Session;
+
+import org.apache.log4j.Logger;
+import org.apache.qpid.test.utils.QpidBrokerTestCase;
+
+public class MultipleTransactedBatchProducerTest extends QpidBrokerTestCase
+{
+    private static final Logger _logger = Logger.getLogger(MultipleTransactedBatchProducerTest.class);
+
+    private static final int MESSAGE_COUNT = 1000;
+    private static final int BATCH_SIZE = 50;
+    private static final int NUM_PRODUCERS = 2;
+    private static final int NUM_CONSUMERS = 3;
+    private static final Random RANDOM = new Random();
+
+    private CountDownLatch _receivedLatch;
+    private String _queueName;
+
+    private volatile String _failMsg;
+
+    public void setUp() throws Exception
+    {
+        //debug level logging often makes this test pass artificially, turn the level down to info.
+        setSystemProperty("amqj.server.logging.level", "INFO");
+        _receivedLatch = new CountDownLatch(MESSAGE_COUNT * NUM_PRODUCERS);
+        setConfigurationProperty("management.enabled", "true");
+        super.setUp();
+        _queueName = getTestQueueName();
+        _failMsg = null;
+    }
+
+    /**
+     * When there are multiple producers submitting batches of messages to a given
+     * queue using transacted sessions, it is highly probable that concurrent
+     * enqueue() activity will occur and attempt delivery of their message to the
+     * same subscription. In this scenario it is likely that one of the attempts
+     * will succeed and the other will result in use of the deliverAsync() method
+     * to start a queue Runner and ensure delivery of the message.
+     *
+     * A defect within the processQueue() method used by the Runner would mean that
+     * delivery of these messages may not occur, should the Runner stop before all
+     * messages have been processed. Such a defect was discovered and found to be
+     * most visible when Selectors are used such that one and only one subscription
+     * can/will accept any given message, but multiple subscriptions are present,
+     * and one of the earlier subscriptions receives more messages than the others.
+     *
+     * This test is to validate that the processQueue() method is able to correctly
+     * deliver all of the messages present for asynchronous delivery to subscriptions,
+     * by utilising multiple batch transacted producers to create the scenario and
+     * ensure all messages are received by a consumer.
+     */
+    public void testMultipleBatchedProducersWithMultipleConsumersUsingSelectors() throws Exception
+    {
+        String selector1 = ("(\"" + _queueName +"\" % " + NUM_CONSUMERS + ") = 0");
+        String selector2 = ("(\"" + _queueName +"\" % " + NUM_CONSUMERS + ") = 1");
+        String selector3 = ("(\"" + _queueName +"\" % " + NUM_CONSUMERS + ") = 2");
+
+        //create consumers
+        Connection conn1 = getConnection();
+        conn1.setExceptionListener(new ExceptionHandler("conn1"));
+        Session sess1 = conn1.createSession(true, Session.SESSION_TRANSACTED);
+        MessageConsumer cons1 = sess1.createConsumer(sess1.createQueue(_queueName), selector1);
+        cons1.setMessageListener(new Cons(sess1,"consumer1"));
+
+        Connection conn2 = getConnection();
+        conn2.setExceptionListener(new ExceptionHandler("conn2"));
+        Session sess2 = conn2.createSession(true, Session.SESSION_TRANSACTED);
+        MessageConsumer cons2 = sess2.createConsumer(sess2.createQueue(_queueName), selector2);
+        cons2.setMessageListener(new Cons(sess2,"consumer2"));
+
+        Connection conn3 = getConnection();
+        conn3.setExceptionListener(new ExceptionHandler("conn3"));
+        Session sess3 = conn3.createSession(true, Session.SESSION_TRANSACTED);
+        MessageConsumer cons3 = sess3.createConsumer(sess3.createQueue(_queueName), selector3);
+        cons3.setMessageListener(new Cons(sess3,"consumer3"));
+
+        conn1.start();
+        conn2.start();
+        conn3.start();
+
+        //create producers
+        Connection connA = getConnection();
+        connA.setExceptionListener(new ExceptionHandler("connA"));
+        Connection connB = getConnection();
+        connB.setExceptionListener(new ExceptionHandler("connB"));
+        Thread producer1 = new Thread(new ProducerThread(connA, _queueName, "producer1"));
+        Thread producer2 = new Thread(new ProducerThread(connB, _queueName, "producer2"));
+
+        producer1.start();
+        Thread.sleep(10);
+        producer2.start();
+
+        //await delivery of the messages
+        boolean result = _receivedLatch.await(75, TimeUnit.SECONDS);
+
+        assertNull("Test failed because: " + String.valueOf(_failMsg), _failMsg);
+        assertTrue("Some of the messages were not all recieved in the given timeframe, remaining count was: "+_receivedLatch.getCount(),
+                   result);
+
+    }
+
+    @Override
+    public Message createNextMessage(Session session, int msgCount) throws JMSException
+    {
+        Message message = super.createNextMessage(session,msgCount);
+
+        //bias at least 50% of the messages to the first consumers selector because
+        //the issue presents itself primarily when an earlier subscription completes
+        //delivery after the later subscriptions
+        int val;
+        if (msgCount % 2 == 0)
+        {
+            val = 0;
+        }
+        else
+        {
+            val = RANDOM.nextInt(Integer.MAX_VALUE);
+        }
+
+        message.setIntProperty(_queueName, val);
+
+        return message;
+    }
+
+    private class Cons implements MessageListener
+    {
+        private Session _sess;
+        private String _desc;
+
+        public Cons(Session sess, String desc)
+        {
+            _sess = sess;
+            _desc = desc;
+        }
+
+        public void onMessage(Message message)
+        {
+            _receivedLatch.countDown();
+            int msgCount = 0;
+            int msgID = 0;
+            try
+            {
+                msgCount = message.getIntProperty(INDEX);
+                msgID = message.getIntProperty(_queueName);
+            }
+            catch (JMSException e)
+            {
+                _logger.error(_desc + " received exception: " + e.getMessage(), e);
+                failAsyncTest(e.getMessage());
+            }
+
+            _logger.info("Consumer received message:"+ msgCount + " with ID: " + msgID);
+
+            try
+            {
+                _sess.commit();
+            }
+            catch (JMSException e)
+            {
+                _logger.error(_desc + " received exception: " + e.getMessage(), e);
+                failAsyncTest(e.getMessage());
+            }
+        }
+    }
+
+    private class ProducerThread implements Runnable
+    {
+        private Connection _conn;
+        private String _dest;
+        private String _desc;
+
+        public ProducerThread(Connection conn, String dest, String desc)
+        {
+            _conn = conn;
+            _dest = dest;
+            _desc = desc;
+        }
+
+        public void run()
+        {
+            try
+            {
+                Session session = _conn.createSession(true, Session.SESSION_TRANSACTED);
+                sendMessage(session, session.createQueue(_dest), MESSAGE_COUNT, BATCH_SIZE);
+            }
+            catch (Exception e)
+            {
+                _logger.error(_desc + " received exception: " + e.getMessage(), e);
+                failAsyncTest(e.getMessage());
+            }
+        }
+    }
+
+    private class ExceptionHandler implements javax.jms.ExceptionListener
+    {
+        private String _desc;
+
+        public ExceptionHandler(String description)
+        {
+            _desc = description;
+        }
+
+        public void onException(JMSException e)
+        {
+            _logger.error(_desc + " received exception: " + e.getMessage(), e);
+            failAsyncTest(e.getMessage());
+        }
+    }
+
+    private void failAsyncTest(String msg)
+    {
+        _logger.error("Failing test because: " + msg);
+        _failMsg = msg;
+    }
+}
\ No newline at end of file
diff --git a/qpid/java/systests/src/main/java/org/apache/qpid/test/client/destination/AddressBasedDestinationTest.java b/qpid/java/systests/src/main/java/org/apache/qpid/test/client/destination/AddressBasedDestinationTest.java
index 30dc30cd81..a737781e30 100644
--- a/qpid/java/systests/src/main/java/org/apache/qpid/test/client/destination/AddressBasedDestinationTest.java
+++ b/qpid/java/systests/src/main/java/org/apache/qpid/test/client/destination/AddressBasedDestinationTest.java
@@ -474,39 +474,6 @@ public class AddressBasedDestinationTest extends QpidBrokerTestCase
     }
 
     /**
-    * Test goal: Verifies that and address based destination can be used successfully
-    *            as a reply to.
-    */
-    public void testAddressBasedReplyTo() throws Exception
-    {
-        Session jmsSession = _connection.createSession(false,Session.AUTO_ACKNOWLEDGE);
-
-        String addr = "ADDR:amq.direct/x512; {create: receiver, " +
-                      "link : {name : 'MY.RESP.QUEUE', " +
-                      "x-declare : { auto-delete: true, exclusive: true, " +
-                                   "arguments : {'qpid.max_size': 1000, 'qpid.policy_type': ring }} } }";
-
-        Destination replyTo = new AMQAnyDestination(addr);
-        Destination dest =new AMQAnyDestination("ADDR:amq.direct/Hello");
-
-        MessageConsumer cons = jmsSession.createConsumer(dest);
-        MessageProducer prod = jmsSession.createProducer(dest);
-        Message m = jmsSession.createTextMessage("Hello");
-        m.setJMSReplyTo(replyTo);
-        prod.send(m);
-
-        Message msg = cons.receive(1000);
-        assertNotNull("consumer should have received the message",msg);
-
-        MessageConsumer replyToCons = jmsSession.createConsumer(replyTo);
-        MessageProducer replyToProd = jmsSession.createProducer(msg.getJMSReplyTo());
-        replyToProd.send(jmsSession.createTextMessage("reply"));
-
-        Message replyToMsg = replyToCons.receive(1000);
-        assertNotNull("The reply to consumer should have got the message",replyToMsg);
-    }
-
-    /**
      * Test goal: Verifies that session.createQueue method
      *            works as expected both with the new and old addressing scheme.
      */
@@ -1020,4 +987,38 @@ public class AddressBasedDestinationTest extends QpidBrokerTestCase
         prod.close();
         cons.close();
     }
+
+    public void testReplyToWithNamelessExchange() throws Exception
+    {
+    	System.setProperty("qpid.declare_exchanges","false");
+    	replyToTest("ADDR:my-queue;{create: always}");
+    	System.setProperty("qpid.declare_exchanges","true");
+    }
+
+    public void testReplyToWithCustomExchange() throws Exception
+    {
+    	replyToTest("ADDR:hello;{create:always,node:{type:topic}}");
+    }
+
+    private void replyToTest(String replyTo) throws Exception
+    {
+		Session session = _connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
+		Destination replyToDest = AMQDestination.createDestination(replyTo);
+	    MessageConsumer replyToCons = session.createConsumer(replyToDest);
+
+		Destination dest = session.createQueue("amq.direct/test");
+
+		MessageConsumer cons = session.createConsumer(dest);
+		MessageProducer prod = session.createProducer(dest);
+		Message m = session.createTextMessage("test");
+		m.setJMSReplyTo(replyToDest);
+		prod.send(m);
+
+		Message msg = cons.receive();
+		MessageProducer prodR = session.createProducer(msg.getJMSReplyTo());
+		prodR.send(session.createTextMessage("x"));
+
+		Message m1 = replyToCons.receive();
+		assertNotNull("The reply to consumer should have received the messsage",m1);
+    }
 }
diff --git a/qpid/java/systests/src/main/java/org/apache/qpid/test/unit/message/JMSPropertiesTest.java b/qpid/java/systests/src/main/java/org/apache/qpid/test/unit/message/JMSPropertiesTest.java
index 830421a01f..8caeaa55c0 100644
--- a/qpid/java/systests/src/main/java/org/apache/qpid/test/unit/message/JMSPropertiesTest.java
+++ b/qpid/java/systests/src/main/java/org/apache/qpid/test/unit/message/JMSPropertiesTest.java
@@ -31,6 +31,7 @@ import org.apache.qpid.test.utils.QpidBrokerTestCase;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import javax.jms.Connection;
 import javax.jms.Destination;
 import javax.jms.Message;
 import javax.jms.MessageConsumer;
@@ -39,7 +40,11 @@ import javax.jms.MessageProducer;
 import javax.jms.ObjectMessage;
 import javax.jms.Queue;
 import javax.jms.Session;
+import javax.jms.Topic;
+
 import java.util.Enumeration;
+import java.util.HashMap;
+import java.util.Map;
 
 /**
  * @author Apache Software Foundation
@@ -163,4 +168,37 @@ public class JMSPropertiesTest extends QpidBrokerTestCase
         con.close();
     }
 
+    /**
+     * Test Goal : test if the message properties can be retrieved properly with out an error
+     *             and also test if unsupported properties are filtered out. See QPID-2930.
+     */
+    public void testGetPropertyNames() throws Exception
+    {
+        Connection con = getConnection("guest", "guest");
+        Session ssn = (AMQSession) con.createSession(false, Session.CLIENT_ACKNOWLEDGE);
+        con.start();
+        
+        Topic topic = ssn.createTopic("test");
+        MessageConsumer consumer = ssn.createConsumer(topic);
+        MessageProducer prod = ssn.createProducer(topic);
+        Message m = ssn.createMessage();
+        m.setObjectProperty("x-amqp-0-10.routing-key", "routing-key".getBytes());
+        m.setObjectProperty("routing-key", "routing-key");
+        prod.send(m);
+        
+        Message msg = consumer.receive(1000);
+        assertNotNull(msg);
+        
+    	Enumeration<String> enu = msg.getPropertyNames();
+    	Map<String,String> map = new HashMap<String,String>();
+    	while (enu.hasMoreElements()) 
+    	{    		
+    		String name = enu.nextElement(); 
+    		String value = msg.getStringProperty(name);
+    		map.put(name, value);
+       } 
+    	
+       assertFalse("Property 'x-amqp-0-10.routing-key' should have been filtered out",map.containsKey("x-amqp-0-10.routing-key"));
+       assertTrue("Property routing-key should be present",map.containsKey("routing-key"));      
+    }
 }
diff --git a/qpid/java/systests/src/main/java/org/apache/qpid/test/utils/QpidBrokerTestCase.java b/qpid/java/systests/src/main/java/org/apache/qpid/test/utils/QpidBrokerTestCase.java
index ae38a75e7a..d7da5d8a78 100644
--- a/qpid/java/systests/src/main/java/org/apache/qpid/test/utils/QpidBrokerTestCase.java
+++ b/qpid/java/systests/src/main/java/org/apache/qpid/test/utils/QpidBrokerTestCase.java
@@ -67,6 +67,7 @@ import org.apache.qpid.server.registry.ApplicationRegistry;
 import org.apache.qpid.server.registry.ConfigurationFileApplicationRegistry;
 import org.apache.qpid.server.store.DerbyMessageStore;
 import org.apache.qpid.url.URLSyntaxException;
+import org.apache.qpid.util.FileUtils;
 import org.apache.qpid.util.LogMonitor;
 
 /**
@@ -109,6 +110,7 @@ public class QpidBrokerTestCase extends QpidTestCase
     private static final String BROKER = "broker";
     private static final String BROKER_CLEAN = "broker.clean";
     private static final String BROKER_CLEAN_BETWEEN_TESTS = "broker.clean.between.tests";
+    private static final String BROKER_EXISTING_QPID_WORK = "broker.existing.qpid.work";
     private static final String BROKER_VERSION = "broker.version";
     protected static final String BROKER_READY = "broker.ready";
     private static final String BROKER_STOPPED = "broker.stopped";
@@ -283,6 +285,19 @@ public class QpidBrokerTestCase extends QpidTestCase
             fail("Unable to test without config file:" + _configFile);
         }
 
+        if(_brokerCleanBetweenTests)
+        {
+            cleanBroker();
+
+            String existingQpidWorkPath = System.getProperty(BROKER_EXISTING_QPID_WORK);
+            if(existingQpidWorkPath != null && !existingQpidWorkPath.equals(""))
+            {
+                File existing = new File(existingQpidWorkPath);
+                File qpidWork = new File(getQpidWork(_broker, getPort()));
+                FileUtils.copyRecursive(existing, qpidWork);
+            }
+        }
+
         startBroker();
     }
 
@@ -490,7 +505,7 @@ public class QpidBrokerTestCase extends QpidTestCase
             // DON'T change PNAME, qpid.stop needs this value.
             env.put("QPID_PNAME", "-DPNAME=QPBRKR -DTNAME=\"" + _testName + "\"");
             // Add the port to QPID_WORK to ensure unique working dirs for multi broker tests
-            env.put("QPID_WORK", System.getProperty("QPID_WORK")+ "/" + port);
+            env.put("QPID_WORK", getQpidWork(_broker, port));
 
 
             // Use the environment variable to set amqj.logging.level for the broker
@@ -578,6 +593,20 @@ public class QpidBrokerTestCase extends QpidTestCase
         _brokers.put(port, process);
     }
 
+    private String getQpidWork(String broker, int port)
+    {
+        if (broker.equals(VM))
+        {
+            return System.getProperty("QPID_WORK");
+        }
+        else if (!broker.equals(EXTERNAL))
+        {
+            return System.getProperty("QPID_WORK")+ "/" + port;
+        }
+
+        return System.getProperty("QPID_WORK");
+    }
+
     public String getTestConfigFile()
     {
         String path = _output == null ? System.getProperty("java.io.tmpdir") : _output;
@@ -1190,7 +1219,8 @@ public class QpidBrokerTestCase extends QpidTestCase
 
         MessageProducer producer = session.createProducer(destination);
 
-        for (int i = offset; i < (count + offset); i++)
+        int i = offset;
+        for (; i < (count + offset); i++)
         {
             Message next = createNextMessage(session, i);
 
@@ -1213,7 +1243,7 @@ public class QpidBrokerTestCase extends QpidTestCase
         // we have no batchSize or
         // our count is not divible by batchSize.
         if (session.getTransacted() &&
-            ( batchSize == 0 || count % batchSize != 0))
+            ( batchSize == 0 || (i-1) % batchSize != 0))
         {
             session.commit();
         }
diff --git a/qpid/java/test-profiles/08StandaloneExcludes b/qpid/java/test-profiles/08StandaloneExcludes
index b482a14c6d..43eb1f8ee5 100644
--- a/qpid/java/test-profiles/08StandaloneExcludes
+++ b/qpid/java/test-profiles/08StandaloneExcludes
@@ -37,3 +37,5 @@ org.apache.qpid.test.unit.message.UTF8Test#*
 org.apache.qpid.client.MessageListenerTest#testSynchronousReceiveNoWait
 
 org.apache.qpid.test.unit.client.connection.ConnectionTest#testUnsupportedSASLMechanism
+
+org.apache.qpid.test.unit.message.JMSPropertiesTest#testGetPropertyNames
diff --git a/qpid/java/test-profiles/JavaInVMExcludes b/qpid/java/test-profiles/JavaInVMExcludes
index 7e7da4302e..7960b28d81 100644
--- a/qpid/java/test-profiles/JavaInVMExcludes
+++ b/qpid/java/test-profiles/JavaInVMExcludes
@@ -41,3 +41,5 @@ org.apache.qpid.test.unit.ack.RecoverTest#testOderingWithSyncConsumer
 
 //The VM broker does not export the logging management JMX MBean
 org.apache.qpid.server.security.acl.ExternalAdminACLTest#*
+
+org.apache.qpid.test.unit.message.JMSPropertiesTest#testGetPropertyNames