QPID-3346: merge **from** trunk **to** qpid-3346 dev branch

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

32 files changed, 1449 insertions, 917 deletions

diff --git a/qpid/java/systests/src/main/java/org/apache/qpid/client/failover/FailoverBehaviourTest.java b/qpid/java/systests/src/main/java/org/apache/qpid/client/failover/FailoverBehaviourTest.java
new file mode 100644
index 0000000000..aa62106c46
--- /dev/null
+++ b/qpid/java/systests/src/main/java/org/apache/qpid/client/failover/FailoverBehaviourTest.java
@@ -0,0 +1,924 @@
+/*
+ * 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.client.failover;
+
+import java.text.MessageFormat;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import javax.jms.Connection;
+import javax.jms.Destination;
+import javax.jms.ExceptionListener;
+import javax.jms.JMSException;
+import javax.jms.Message;
+import javax.jms.MessageConsumer;
+import javax.jms.MessageListener;
+import javax.jms.MessageProducer;
+import javax.jms.Session;
+import javax.jms.TextMessage;
+import javax.jms.TransactionRolledBackException;
+
+import org.apache.qpid.client.AMQConnection;
+import org.apache.qpid.jms.ConnectionListener;
+import org.apache.qpid.test.utils.FailoverBaseCase;
+
+/**
+ * Test suite to test all possible failover corner cases
+ */
+public class FailoverBehaviourTest extends FailoverBaseCase implements ConnectionListener, ExceptionListener
+{
+    private static final String TEST_MESSAGE_FORMAT = "test message {0}";
+
+    /** Indicates whether tests are run against clustered broker */
+    private static boolean CLUSTERED = Boolean.getBoolean("profile.clustered");
+
+    /** Default number of messages to send before failover */
+    private static final int DEFAULT_NUMBER_OF_MESSAGES = 10;
+
+    /** Actual number of messages to send before failover */
+    protected int _messageNumber = Integer.getInteger("profile.failoverMsgCount", DEFAULT_NUMBER_OF_MESSAGES);
+
+    /** Test connection */
+    protected Connection _connection;
+
+    /**
+     * Failover completion latch is used to wait till connectivity to broker is
+     * restored
+     */
+    private CountDownLatch _failoverComplete;
+
+    /**
+     * Consumer session
+     */
+    private Session _consumerSession;
+
+    /**
+     * Test destination
+     */
+    private Destination _destination;
+
+    /**
+     * Consumer
+     */
+    private MessageConsumer _consumer;
+
+    /**
+     * Producer session
+     */
+    private Session _producerSession;
+
+    /**
+     * Producer
+     */
+    private MessageProducer _producer;
+
+    /**
+     * Holds exception sent into {@link ExceptionListener} on failover
+     */
+    private JMSException _exceptionListenerException;
+
+    @Override
+    protected void setUp() throws Exception
+    {
+        super.setUp();
+
+        _connection = getConnection();
+        _connection.setExceptionListener(this);
+        ((AMQConnection) _connection).setConnectionListener(this);
+        _failoverComplete = new CountDownLatch(1);
+    }
+
+    /**
+     * Test whether MessageProducer can successfully publish messages after
+     * failover and rollback transaction
+     */
+    public void testMessageProducingAndRollbackAfterFailover() throws Exception
+    {
+        init(Session.SESSION_TRANSACTED, true);
+        produceMessages();
+        causeFailure();
+
+        assertFailoverException();
+        // producer should be able to send messages after failover
+        _producer.send(_producerSession.createTextMessage("test message " + _messageNumber));
+
+        // rollback after failover
+        _producerSession.rollback();
+
+        // tests whether sending and committing is working after failover
+        produceMessages();
+        _producerSession.commit();
+
+        // tests whether receiving and committing is working after failover
+        consumeMessages();
+        _consumerSession.commit();
+    }
+
+    /**
+     * Test whether {@link TransactionRolledBackException} is thrown on commit
+     * of dirty transacted session after failover.
+     * <p>
+     * Verifies whether second after failover commit is successful.
+     */
+    public void testTransactionRolledBackExceptionThrownOnCommitAfterFailoverOnProducingMessages() throws Exception
+    {
+        init(Session.SESSION_TRANSACTED, true);
+        produceMessages();
+        causeFailure();
+
+        assertFailoverException();
+
+        // producer should be able to send messages after failover
+        _producer.send(_producerSession.createTextMessage("test message " + _messageNumber));
+
+        try
+        {
+            _producerSession.commit();
+            fail("TransactionRolledBackException is expected on commit after failover with dirty session!");
+        }
+        catch (JMSException t)
+        {
+            assertTrue("Expected TransactionRolledBackException but thrown " + t,
+                    t instanceof TransactionRolledBackException);
+        }
+
+        // simulate process of user replaying the transaction
+        produceMessages("replayed test message {0}", _messageNumber, false);
+
+        // no exception should be thrown
+        _producerSession.commit();
+
+        // only messages sent after rollback should be received
+        consumeMessages("replayed test message {0}", _messageNumber);
+
+        // no exception should be thrown
+        _consumerSession.commit();
+    }
+
+    /**
+     * Tests JMSException is not thrown on commit with a clean session after
+     * failover
+     */
+    public void testNoJMSExceptionThrownOnCommitAfterFailoverWithCleanProducerSession() throws Exception
+    {
+        init(Session.SESSION_TRANSACTED, true);
+
+        causeFailure();
+
+        assertFailoverException();
+
+        // should not throw an exception for a clean session
+        _producerSession.commit();
+
+        // tests whether sending and committing is working after failover
+        produceMessages();
+        _producerSession.commit();
+
+        // tests whether receiving and committing is working after failover
+        consumeMessages();
+        _consumerSession.commit();
+    }
+
+    /**
+     * Tests {@link TransactionRolledBackException} is thrown on commit of dirty
+     * transacted session after failover.
+     * <p>
+     * Verifies whether second after failover commit is successful.
+     */
+    public void testTransactionRolledBackExceptionThrownOnCommitAfterFailoverOnMessageReceiving() throws Exception
+    {
+        init(Session.SESSION_TRANSACTED, true);
+        produceMessages();
+        _producerSession.commit();
+
+        // receive messages but do not commit
+        consumeMessages();
+
+        causeFailure();
+
+        assertFailoverException();
+
+        try
+        {
+            // should throw TransactionRolledBackException
+            _consumerSession.commit();
+            fail("TransactionRolledBackException is expected on commit after failover");
+        }
+        catch (Exception t)
+        {
+            assertTrue("Expected TransactionRolledBackException but thrown " + t,
+                    t instanceof TransactionRolledBackException);
+        }
+
+        resendMessagesIfNecessary();
+
+        // consume messages successfully
+        consumeMessages();
+        _consumerSession.commit();
+    }
+
+    /**
+     * Tests JMSException is not thrown on commit with a clean session after failover
+     */
+    public void testNoJMSExceptionThrownOnCommitAfterFailoverWithCleanConsumerSession() throws Exception
+    {
+        init(Session.SESSION_TRANSACTED, true);
+        produceMessages();
+        _producerSession.commit();
+
+        consumeMessages();
+        _consumerSession.commit();
+
+        causeFailure();
+
+        assertFailoverException();
+
+        // should not throw an exception with a clean consumer session
+        _consumerSession.commit();
+    }
+
+    /**
+     * Test that TransactionRolledBackException is thrown on commit of
+     * dirty session in asynchronous consumer after failover.
+     */
+    public void testTransactionRolledBackExceptionThrownOnCommitAfterFailoverOnReceivingMessagesAsynchronously()
+    throws Exception
+    {
+        init(Session.SESSION_TRANSACTED, false);
+        FailoverTestMessageListener ml = new FailoverTestMessageListener();
+        _consumer.setMessageListener(ml);
+
+        _connection.start();
+
+        produceMessages();
+        _producerSession.commit();
+
+        // wait for message receiving
+        ml.awaitForEnd();
+
+        assertEquals("Received unexpected number of messages!", _messageNumber, ml.getMessageCounter());
+
+        // assert messages
+        int counter = 0;
+        for (Message message : ml.getReceivedMessages())
+        {
+            assertReceivedMessage(message, TEST_MESSAGE_FORMAT, counter++);
+        }
+        ml.reset();
+
+        causeFailure();
+        assertFailoverException();
+
+
+        try
+        {
+            _consumerSession.commit();
+            fail("TransactionRolledBackException should be thrown!");
+        }
+        catch (TransactionRolledBackException e)
+        {
+            // that is what is expected
+        }
+
+        resendMessagesIfNecessary();
+
+        // wait for message receiving
+        ml.awaitForEnd();
+
+        assertEquals("Received unexpected number of messages!", _messageNumber, ml.getMessageCounter());
+
+        // assert messages
+        counter = 0;
+        for (Message message : ml.getReceivedMessages())
+        {
+            assertReceivedMessage(message, TEST_MESSAGE_FORMAT, counter++);
+        }
+
+        // commit again. It should be successful
+        _consumerSession.commit();
+    }
+
+    /**
+     * Test that {@link Session#rollback()} does not throw exception after failover
+     * and that we are able to consume messages.
+     */
+    public void testRollbackAfterFailover() throws Exception
+    {
+        init(Session.SESSION_TRANSACTED, true);
+
+        produceMessages();
+        _producerSession.commit();
+
+        consumeMessages();
+
+        causeFailure();
+
+        assertFailoverException();
+
+        _consumerSession.rollback();
+
+        resendMessagesIfNecessary();
+
+        // tests whether receiving and committing is working after failover
+        consumeMessages();
+        _consumerSession.commit();
+    }
+
+    /**
+     * Test that {@link Session#rollback()} does not throw exception after receiving further messages
+     * after failover, and we can receive published messages after rollback.
+     */
+    public void testRollbackAfterReceivingAfterFailover() throws Exception
+    {
+        init(Session.SESSION_TRANSACTED, true);
+
+        produceMessages();
+        _producerSession.commit();
+
+        consumeMessages();
+        causeFailure();
+
+        assertFailoverException();
+
+        resendMessagesIfNecessary();
+
+        consumeMessages();
+
+        _consumerSession.rollback();
+
+        // tests whether receiving and committing is working after failover
+        consumeMessages();
+        _consumerSession.commit();
+    }
+
+    /**
+     * Test that {@link Session#recover()} does not throw an exception after failover
+     * and that we can consume messages after recover.
+     */
+    public void testRecoverAfterFailover() throws Exception
+    {
+        init(Session.CLIENT_ACKNOWLEDGE, true);
+
+        produceMessages();
+
+        // consume messages but do not acknowledge them
+        consumeMessages();
+
+        causeFailure();
+
+        assertFailoverException();
+
+        _consumerSession.recover();
+
+        resendMessagesIfNecessary();
+
+        // tests whether receiving and acknowledgment is working after recover
+        Message lastMessage = consumeMessages();
+        lastMessage.acknowledge();
+    }
+
+    /**
+     * Test that receiving more messages after failover and then calling
+     * {@link Session#recover()} does not throw an exception
+     * and that we can consume messages after recover.
+     */
+    public void testRecoverWithConsumedMessagesAfterFailover() throws Exception
+    {
+        init(Session.CLIENT_ACKNOWLEDGE, true);
+
+        produceMessages();
+
+        // consume messages but do not acknowledge them
+        consumeMessages();
+
+        causeFailure();
+
+        assertFailoverException();
+
+        // publishing should work after failover
+        resendMessagesIfNecessary();
+
+        // consume messages again on a dirty session
+        consumeMessages();
+
+        // recover should successfully restore session
+        _consumerSession.recover();
+
+        // tests whether receiving and acknowledgment is working after recover
+        Message lastMessage = consumeMessages();
+        lastMessage.acknowledge();
+    }
+
+    /**
+     * Test that first call to {@link Message#acknowledge()} after failover
+     * throws a JMSEXception if session is dirty.
+     */
+    public void testAcknowledgeAfterFailover() throws Exception
+    {
+        init(Session.CLIENT_ACKNOWLEDGE, true);
+
+        produceMessages();
+
+        // consume messages but do not acknowledge them
+        Message lastMessage = consumeMessages();
+        causeFailure();
+
+        assertFailoverException();
+
+        try
+        {
+            // an implicit recover performed when acknowledge throws an exception due to failover 
+            lastMessage.acknowledge();
+            fail("JMSException should be thrown");
+        }
+        catch (JMSException t)
+        {
+            // TODO: assert error code and/or expected exception type
+        }
+
+        resendMessagesIfNecessary();
+
+        // tests whether receiving and acknowledgment is working after recover
+        lastMessage = consumeMessages();
+        lastMessage.acknowledge();
+    }
+
+    /**
+     * Test that receiving of messages after failover prior to calling
+     * {@link Message#acknowledge()} still results in acknowledge throwing an exception.
+     */
+    public void testAcknowledgeAfterMessageReceivingAfterFailover() throws Exception
+    {
+        init(Session.CLIENT_ACKNOWLEDGE, true);
+
+        produceMessages();
+
+        // consume messages but do not acknowledge them
+        consumeMessages();
+        causeFailure();
+
+        assertFailoverException();
+
+        resendMessagesIfNecessary();
+
+        // consume again on dirty session
+        Message lastMessage = consumeMessages();
+        try
+        {
+            // an implicit recover performed when acknowledge throws an exception due to failover 
+            lastMessage.acknowledge();
+            fail("JMSException should be thrown");
+        }
+        catch (JMSException t)
+        {
+            // TODO: assert error code and/or expected exception type
+        }
+
+        // tests whether receiving and acknowledgment is working on a clean session
+        lastMessage = consumeMessages();
+        lastMessage.acknowledge();
+    }
+
+    /**
+     * Tests that call to {@link Message#acknowledge()} after failover throws an exception in asynchronous consumer
+     * and we can consume messages after acknowledge.
+     */
+    public void testAcknowledgeAfterFailoverForAsynchronousConsumer() throws Exception
+    {
+        init(Session.CLIENT_ACKNOWLEDGE, false);
+        FailoverTestMessageListener ml = new FailoverTestMessageListener();
+        _consumer.setMessageListener(ml);
+        _connection.start();
+
+        produceMessages();
+
+        // wait for message receiving
+        ml.awaitForEnd();
+
+        assertEquals("Received unexpected number of messages!", _messageNumber, ml.getMessageCounter());
+
+        // assert messages
+        int counter = 0;
+        Message currentMessage = null;
+        for (Message message : ml.getReceivedMessages())
+        {
+            assertReceivedMessage(message, TEST_MESSAGE_FORMAT, counter++);
+            currentMessage = message;
+        }
+        ml.reset();
+
+        causeFailure();
+        assertFailoverException();
+
+
+        try
+        {
+            currentMessage.acknowledge();
+            fail("JMSException should be thrown!");
+        }
+        catch (JMSException e)
+        {
+            // TODO: assert error code and/or expected exception type
+        }
+
+        resendMessagesIfNecessary();
+
+        // wait for message receiving
+        ml.awaitForEnd();
+
+        assertEquals("Received unexpected number of messages!", _messageNumber, ml.getMessageCounter());
+
+        // assert messages
+        counter = 0;
+        for (Message message : ml.getReceivedMessages())
+        {
+            assertReceivedMessage(message, TEST_MESSAGE_FORMAT, counter++);
+            currentMessage = message;
+        }
+
+        // acknowledge again. It should be successful
+        currentMessage.acknowledge();
+    }
+
+    /**
+     * Test whether {@link Session#recover()} works as expected after failover
+     * in AA mode.
+     */
+    public void testRecoverAfterFailoverInAutoAcknowledgeMode() throws Exception
+    {
+        init(Session.AUTO_ACKNOWLEDGE, true);
+
+        produceMessages();
+
+        // receive first message in order to start a dispatcher thread
+        Message receivedMessage = _consumer.receive(1000l);
+        assertReceivedMessage(receivedMessage, TEST_MESSAGE_FORMAT, 0);
+
+        causeFailure();
+
+        assertFailoverException();
+
+        _consumerSession.recover();
+
+        resendMessagesIfNecessary();
+
+        // tests whether receiving is working after recover
+        consumeMessages();
+    }
+
+    public void testClientAcknowledgedSessionCloseAfterFailover() throws Exception
+    {
+        sessionCloseAfterFailoverImpl(Session.CLIENT_ACKNOWLEDGE);
+    }
+
+    public void testTransactedSessionCloseAfterFailover() throws Exception
+    {
+        sessionCloseAfterFailoverImpl(Session.SESSION_TRANSACTED);
+    }
+
+    public void testAutoAcknowledgedSessionCloseAfterFailover() throws Exception
+    {
+        sessionCloseAfterFailoverImpl(Session.AUTO_ACKNOWLEDGE);
+    }
+
+    /**
+     * Tests {@link Session#close()} for session with given acknowledge mode 
+     * to ensure that close works after failover.
+     *
+     * @param acknowledgeMode session acknowledge mode
+     * @throws JMSException
+     */
+    private void sessionCloseAfterFailoverImpl(int acknowledgeMode) throws JMSException
+    {
+        init(acknowledgeMode, true);
+        produceMessages(TEST_MESSAGE_FORMAT, _messageNumber, false);
+        if (acknowledgeMode == Session.SESSION_TRANSACTED)
+        {
+            _producerSession.commit();
+        }
+
+        // intentionally receive message but do not commit or acknowledge it in
+        // case of transacted or CLIENT_ACK session
+        Message receivedMessage = _consumer.receive(1000l);
+        assertReceivedMessage(receivedMessage, TEST_MESSAGE_FORMAT, 0);
+
+        causeFailure();
+
+        assertFailoverException();
+
+        // for transacted/client_ack session
+        // no exception should be thrown but transaction should be automatically
+        // rolled back
+        _consumerSession.close();
+    }
+
+    /**
+     * A helper method to instantiate produce and consumer sessions, producer
+     * and consumer.
+     *
+     * @param acknowledgeMode
+     *            acknowledge mode
+     * @param startConnection
+     *            indicates whether connection should be started
+     * @throws JMSException
+     */
+    private void init(int acknowledgeMode, boolean startConnection) throws JMSException
+    {
+        boolean isTransacted = acknowledgeMode == Session.SESSION_TRANSACTED ? true : false;
+
+        _consumerSession = _connection.createSession(isTransacted, acknowledgeMode);
+        _destination = _consumerSession.createQueue(getTestQueueName() + "_" + System.currentTimeMillis());
+        _consumer = _consumerSession.createConsumer(_destination);
+
+        if (startConnection)
+        {
+            _connection.start();
+        }
+
+        _producerSession = _connection.createSession(isTransacted, acknowledgeMode);
+        _producer = _producerSession.createProducer(_destination);
+
+    }
+
+    /**
+     * Resends messages if reconnected to a non-clustered broker
+     *
+     * @throws JMSException
+     */
+    private void resendMessagesIfNecessary() throws JMSException
+    {
+        if (!CLUSTERED)
+        {
+            // assert that a new broker does not have messages on a queue
+            if (_consumer.getMessageListener() == null)
+            {
+                Message message = _consumer.receive(100l);
+                assertNull("Received a message after failover with non-clustered broker!", message);
+            }
+            // re-sending messages if reconnected to a non-clustered broker
+            produceMessages(true);
+        }
+    }
+
+    /**
+     * Produces a default number of messages with default text content into test
+     * queue
+     *
+     * @throws JMSException
+     */
+    private void produceMessages() throws JMSException
+    {
+        produceMessages(false);
+    }
+
+    private void produceMessages(boolean seperateProducer) throws JMSException
+    {
+        produceMessages(TEST_MESSAGE_FORMAT, _messageNumber, seperateProducer);
+    }
+
+    /**
+     * Consumes a default number of messages and asserts their content.
+     *
+     * @return last consumed message
+     * @throws JMSException
+     */
+    private Message consumeMessages() throws JMSException
+    {
+        return consumeMessages(TEST_MESSAGE_FORMAT, _messageNumber);
+    }
+
+    /**
+     * Produces given number of text messages with content matching given
+     * content pattern
+     *
+     * @param messagePattern message content pattern
+     * @param messageNumber  number of messages to send
+     * @param standaloneProducer whether to use the existing producer or a new one.
+     * @throws JMSException
+     */
+    private void produceMessages(String messagePattern, int messageNumber, boolean standaloneProducer) throws JMSException
+    {
+        Session producerSession;
+        MessageProducer producer;
+
+        if(standaloneProducer)
+        {
+            producerSession = _connection.createSession(true, Session.SESSION_TRANSACTED);
+            producer = producerSession.createProducer(_destination);
+        }
+        else
+        {
+            producerSession = _producerSession;
+            producer = _producer;
+        }
+
+        for (int i = 0; i < messageNumber; i++)
+        {
+            String text = MessageFormat.format(messagePattern, i);
+            Message message = producerSession.createTextMessage(text);
+            producer.send(message);
+        }
+
+        if(standaloneProducer)
+        {
+            producerSession.commit();
+        }
+    }
+
+    /**
+     * Consumes given number of text messages and asserts that their content
+     * matches given pattern
+     *
+     * @param messagePattern
+     *            messages content pattern
+     * @param messageNumber
+     *            message number to received
+     * @return last consumed message
+     * @throws JMSException
+     */
+    private Message consumeMessages(String messagePattern, int messageNumber) throws JMSException
+    {
+        Message receivedMesssage = null;
+        for (int i = 0; i < messageNumber; i++)
+        {
+            receivedMesssage = _consumer.receive(1000l);
+            assertReceivedMessage(receivedMesssage, messagePattern, i);
+        }
+        return receivedMesssage;
+    }
+
+    /**
+     * Asserts received message
+     *
+     * @param receivedMessage
+     *            received message
+     * @param messagePattern
+     *            messages content pattern
+     * @param messageIndex
+     *            message index
+     */
+    private void assertReceivedMessage(Message receivedMessage, String messagePattern, int messageIndex)
+    {
+        assertNotNull("Expected message [" + messageIndex + "] is not received!", receivedMessage);
+        assertTrue("Failure to receive message [" + messageIndex + "], expected TextMessage but received "
+                + receivedMessage, receivedMessage instanceof TextMessage);
+        String expectedText = MessageFormat.format(messagePattern, messageIndex);
+        String receivedText = null;
+        try
+        {
+            receivedText = ((TextMessage) receivedMessage).getText();
+        }
+        catch (JMSException e)
+        {
+            fail("JMSException occured while getting message text:" + e.getMessage());
+        }
+        assertEquals("Failover is broken! Expected [" + expectedText + "] but got [" + receivedText + "]",
+                expectedText, receivedText);
+    }
+
+    /**
+     * Causes failover and waits till connection is re-established.
+     */
+    private void causeFailure()
+    {
+        causeFailure(getFailingPort(), DEFAULT_FAILOVER_TIME * 2);
+    }
+
+    /**
+     * Causes failover by stopping broker on given port and waits till
+     * connection is re-established during given time interval.
+     *
+     * @param port
+     *            broker port
+     * @param delay
+     *            time interval to wait for connection re-establishement
+     */
+    private void causeFailure(int port, long delay)
+    {
+        failBroker(port);
+
+        awaitForFailoverCompletion(delay);
+    }
+
+    private void awaitForFailoverCompletion(long delay)
+    {
+        _logger.info("Awaiting Failover completion..");
+        try
+        {
+            if (!_failoverComplete.await(delay, TimeUnit.MILLISECONDS))
+            {
+                fail("Failover did not complete");
+            }
+        }
+        catch (InterruptedException e)
+        {
+            fail("Test was interrupted:" + e.getMessage());
+        }
+    }
+
+    private void assertFailoverException()
+    {
+        // TODO: assert exception is received (once implemented)
+        // along with error code and/or expected exception type
+    }
+
+    @Override
+    public void bytesSent(long count)
+    {
+    }
+
+    @Override
+    public void bytesReceived(long count)
+    {
+    }
+
+    @Override
+    public boolean preFailover(boolean redirect)
+    {
+        return true;
+    }
+
+    @Override
+    public boolean preResubscribe()
+    {
+        return true;
+    }
+
+    @Override
+    public void failoverComplete()
+    {
+        _failoverComplete.countDown();
+    }
+
+    @Override
+    public void onException(JMSException e)
+    {
+        _exceptionListenerException = e;
+    }
+
+    private class FailoverTestMessageListener implements MessageListener
+    {
+        // message counter
+        private AtomicInteger _counter = new AtomicInteger();
+
+        private List<Message> _receivedMessage = new ArrayList<Message>();
+
+        private volatile CountDownLatch _endLatch;
+
+        public FailoverTestMessageListener() throws JMSException
+        {
+            _endLatch = new CountDownLatch(1);
+        }
+
+        @Override
+        public void onMessage(Message message)
+        {
+            _receivedMessage.add(message);
+            if (_counter.incrementAndGet() % _messageNumber == 0)
+            {
+                _endLatch.countDown();
+            }
+        }
+
+        public void reset()
+        {
+            _receivedMessage.clear();
+            _endLatch = new CountDownLatch(1);
+            _counter.set(0);
+        }
+
+        public List<Message> getReceivedMessages()
+        {
+            return _receivedMessage;
+        }
+
+        public Object awaitForEnd() throws InterruptedException
+        {
+            return _endLatch.await((long) _messageNumber, TimeUnit.SECONDS);
+        }
+
+        public int getMessageCounter()
+        {
+            return _counter.get();
+        }
+    }
+}
diff --git a/qpid/java/systests/src/main/java/org/apache/qpid/client/ssl/SSLTest.java b/qpid/java/systests/src/main/java/org/apache/qpid/client/ssl/SSLTest.java
index 471ebb16fc..d754979ab9 100644
--- a/qpid/java/systests/src/main/java/org/apache/qpid/client/ssl/SSLTest.java
+++ b/qpid/java/systests/src/main/java/org/apache/qpid/client/ssl/SSLTest.java
@@ -32,9 +32,9 @@ import org.apache.qpid.transport.Connection;
 
 public class SSLTest extends QpidBrokerTestCase
 {
-    private static final String KEYSTORE = TEST_RESOURCES_DIR + "/ssl/java_client_keystore.jks";
+    private static final String KEYSTORE = "test-profiles/test_resources/ssl/java_client_keystore.jks";
     private static final String KEYSTORE_PASSWORD = "password";
-    private static final String TRUSTSTORE = TEST_RESOURCES_DIR + "/ssl/java_client_truststore.jks";
+    private static final String TRUSTSTORE = "test-profiles/test_resources/ssl/java_client_truststore.jks";
     private static final String TRUSTSTORE_PASSWORD = "password";
     private static final String CERT_ALIAS_APP1 = "app1";
     private static final String CERT_ALIAS_APP2 = "app2";
diff --git a/qpid/java/systests/src/main/java/org/apache/qpid/management/jmx/ManagementActorLoggingTest.java b/qpid/java/systests/src/main/java/org/apache/qpid/management/jmx/ManagementActorLoggingTest.java
index 2864d8e994..12a1682212 100644
--- a/qpid/java/systests/src/main/java/org/apache/qpid/management/jmx/ManagementActorLoggingTest.java
+++ b/qpid/java/systests/src/main/java/org/apache/qpid/management/jmx/ManagementActorLoggingTest.java
@@ -106,7 +106,7 @@ public class ManagementActorLoggingTest extends AbstractTestLogging
         });
 
         //Remove the connection close from any 0-10 connections
-        _monitor.reset();
+        _monitor.markDiscardPoint();
 
         // Get a managedConnection
         ManagedConnection mangedConnection = _jmxUtils.getManagedObject(ManagedConnection.class, "org.apache.qpid:type=VirtualHost.Connection,*");
@@ -147,7 +147,7 @@ public class ManagementActorLoggingTest extends AbstractTestLogging
      */
     public void testCreateExchangeDirectTransientViaManagementConsole() throws IOException, JMException
     {
-        _monitor.reset();
+        _monitor.markDiscardPoint();
 
         _jmxUtils.createExchange("test", getName(), "direct", false);
 
@@ -171,7 +171,7 @@ public class ManagementActorLoggingTest extends AbstractTestLogging
     public void testCreateExchangeTopicTransientViaManagementConsole() throws IOException, JMException
     {
         //Remove any previous exchange declares
-        _monitor.reset();
+        _monitor.markDiscardPoint();
 
         _jmxUtils.createExchange("test", getName(), "topic", false);
 
@@ -196,7 +196,7 @@ public class ManagementActorLoggingTest extends AbstractTestLogging
     public void testCreateExchangeFanoutTransientViaManagementConsole() throws IOException, JMException
     {
         //Remove any previous exchange declares
-        _monitor.reset();
+        _monitor.markDiscardPoint();
 
         _jmxUtils.createExchange("test", getName(), "fanout", false);
 
@@ -221,7 +221,7 @@ public class ManagementActorLoggingTest extends AbstractTestLogging
     public void testCreateExchangeHeadersTransientViaManagementConsole() throws IOException, JMException
     {
         //Remove any previous exchange declares
-        _monitor.reset();
+        _monitor.markDiscardPoint();
 
         _jmxUtils.createExchange("test", getName(), "headers", false);
 
@@ -265,7 +265,7 @@ public class ManagementActorLoggingTest extends AbstractTestLogging
     public void testCreateQueueTransientViaManagementConsole() throws IOException, JMException
     {
         //Remove any previous queue declares
-        _monitor.reset();
+        _monitor.markDiscardPoint();
 
         _jmxUtils.createQueue("test", getName(), null, false);
 
@@ -308,7 +308,7 @@ public class ManagementActorLoggingTest extends AbstractTestLogging
     public void testQueueDeleteViaManagementConsole() throws IOException, JMException
     {
         //Remove any previous queue declares
-        _monitor.reset();
+        _monitor.markDiscardPoint();
 
         _jmxUtils.createQueue("test", getName(), null, false);
 
@@ -354,7 +354,7 @@ public class ManagementActorLoggingTest extends AbstractTestLogging
     public void testBindingCreateOnDirectViaManagementConsole() throws IOException, JMException
     {
         //Remove any previous queue declares
-        _monitor.reset();
+        _monitor.markDiscardPoint();
 
         _jmxUtils.createQueue("test", getName(), null, false);
 
@@ -381,7 +381,7 @@ public class ManagementActorLoggingTest extends AbstractTestLogging
     public void testBindingCreateOnTopicViaManagementConsole() throws IOException, JMException
     {
         //Remove any previous queue declares
-        _monitor.reset();
+        _monitor.markDiscardPoint();
 
         _jmxUtils.createQueue("test", getName(), null, false);
 
@@ -408,7 +408,7 @@ public class ManagementActorLoggingTest extends AbstractTestLogging
     public void testBindingCreateOnFanoutViaManagementConsole() throws IOException, JMException
     {
         //Remove any previous queue declares
-        _monitor.reset();
+        _monitor.markDiscardPoint();
 
         _jmxUtils.createQueue("test", getName(), null, false);
 
@@ -455,7 +455,7 @@ public class ManagementActorLoggingTest extends AbstractTestLogging
     public void testUnRegisterExchangeViaManagementConsole() throws IOException, JMException
     {
         //Remove any previous queue declares
-        _monitor.reset();
+        _monitor.markDiscardPoint();
 
         _jmxUtils.createExchange("test", getName(), "direct", false);
 
diff --git a/qpid/java/systests/src/main/java/org/apache/qpid/server/logging/AbstractTestLogging.java b/qpid/java/systests/src/main/java/org/apache/qpid/server/logging/AbstractTestLogging.java
index f56f428f0b..484c2afeb5 100644
--- a/qpid/java/systests/src/main/java/org/apache/qpid/server/logging/AbstractTestLogging.java
+++ b/qpid/java/systests/src/main/java/org/apache/qpid/server/logging/AbstractTestLogging.java
@@ -349,7 +349,7 @@ public class AbstractTestLogging extends QpidBrokerTestCase
 
     public boolean waitForMessage(String message, long wait) throws FileNotFoundException, IOException
     {
-        return _monitor.waitForMessage(message, wait, true);
+        return _monitor.waitForMessage(message, wait);
     }
 
     /**
diff --git a/qpid/java/systests/src/main/java/org/apache/qpid/server/logging/AlertingTest.java b/qpid/java/systests/src/main/java/org/apache/qpid/server/logging/AlertingTest.java
index a2487b49bf..aef98b8a2a 100644
--- a/qpid/java/systests/src/main/java/org/apache/qpid/server/logging/AlertingTest.java
+++ b/qpid/java/systests/src/main/java/org/apache/qpid/server/logging/AlertingTest.java
@@ -136,7 +136,7 @@ public class AlertingTest extends AbstractTestLogging
         stopBroker();
 
         // Rest the monitoring clearing the current output file.
-        _monitor.reset();
+        _monitor.markDiscardPoint();
         startBroker();
         wasAlertFired();
     }
@@ -169,7 +169,7 @@ public class AlertingTest extends AbstractTestLogging
 
         stopBroker();
 
-        _monitor.reset();
+        _monitor.markDiscardPoint();
 
         // Change max message count to 5, start broker and make sure that that's triggered at the right time
         setConfigurationProperty("virtualhosts.virtualhost." + VIRTUALHOST + ".queues.maximumMessageCount", "5");
diff --git a/qpid/java/systests/src/main/java/org/apache/qpid/server/logging/BindingLoggingTest.java b/qpid/java/systests/src/main/java/org/apache/qpid/server/logging/BindingLoggingTest.java
index 97914f84a5..be2da128bc 100644
--- a/qpid/java/systests/src/main/java/org/apache/qpid/server/logging/BindingLoggingTest.java
+++ b/qpid/java/systests/src/main/java/org/apache/qpid/server/logging/BindingLoggingTest.java
@@ -55,7 +55,7 @@ public class BindingLoggingTest extends AbstractTestLogging
     {
         super.setUp();
         //Ignore broker startup messages
-        _monitor.reset();
+        _monitor.markDiscardPoint();
 
         _connection = getConnection();
 
diff --git a/qpid/java/systests/src/main/java/org/apache/qpid/server/logging/BrokerLoggingTest.java b/qpid/java/systests/src/main/java/org/apache/qpid/server/logging/BrokerLoggingTest.java
index e901903eb4..7969ffc059 100644
--- a/qpid/java/systests/src/main/java/org/apache/qpid/server/logging/BrokerLoggingTest.java
+++ b/qpid/java/systests/src/main/java/org/apache/qpid/server/logging/BrokerLoggingTest.java
@@ -246,7 +246,7 @@ public class BrokerLoggingTest extends AbstractTestLogging
         if (isJavaBroker() && isExternalBroker())
         {
             // Get custom -l value used during testing for the broker startup
-            String customLog4j = _brokerCommand.substring(_brokerCommand.indexOf("-l") + 2);
+            String customLog4j = _brokerCommand.substring(_brokerCommand.indexOf("-l") + 2).trim();
 
             String TESTID = "BRK-1007";
 
diff --git a/qpid/java/systests/src/main/java/org/apache/qpid/server/logging/DerbyMessageStoreLoggingTest.java b/qpid/java/systests/src/main/java/org/apache/qpid/server/logging/DerbyMessageStoreLoggingTest.java
index 16c529316a..d45bde2d98 100644
--- a/qpid/java/systests/src/main/java/org/apache/qpid/server/logging/DerbyMessageStoreLoggingTest.java
+++ b/qpid/java/systests/src/main/java/org/apache/qpid/server/logging/DerbyMessageStoreLoggingTest.java
@@ -496,7 +496,7 @@ public class DerbyMessageStoreLoggingTest extends MemoryMessageStoreLoggingTest
             stopBroker();
 
             // Clear our monitor
-            _monitor.reset();
+            _monitor.markDiscardPoint();
 
             startBroker();
 
diff --git a/qpid/java/systests/src/main/java/org/apache/qpid/server/logging/DurableQueueLoggingTest.java b/qpid/java/systests/src/main/java/org/apache/qpid/server/logging/DurableQueueLoggingTest.java
index 32adc49521..602bdb66b5 100644
--- a/qpid/java/systests/src/main/java/org/apache/qpid/server/logging/DurableQueueLoggingTest.java
+++ b/qpid/java/systests/src/main/java/org/apache/qpid/server/logging/DurableQueueLoggingTest.java
@@ -58,7 +58,7 @@ public class DurableQueueLoggingTest extends AbstractTestLogging
     {
         super.setUp();
         //Ensure we only have logs from our test
-        _monitor.reset();
+        _monitor.markDiscardPoint();
 
         _connection = getConnection();
         _session = _connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
diff --git a/qpid/java/systests/src/main/java/org/apache/qpid/server/logging/ExchangeLoggingTest.java b/qpid/java/systests/src/main/java/org/apache/qpid/server/logging/ExchangeLoggingTest.java
index b3c080b2e3..ec96f778f6 100644
--- a/qpid/java/systests/src/main/java/org/apache/qpid/server/logging/ExchangeLoggingTest.java
+++ b/qpid/java/systests/src/main/java/org/apache/qpid/server/logging/ExchangeLoggingTest.java
@@ -125,7 +125,7 @@ public class ExchangeLoggingTest extends AbstractTestLogging
     public void testExchangeCreate() throws JMSException, IOException
     {
         //Ignore broker startup messages
-        _monitor.reset();
+        _monitor.markDiscardPoint();
 
         _session.createConsumer(_queue);
         // Ensure we have received the EXH log msg.
@@ -179,7 +179,7 @@ public class ExchangeLoggingTest extends AbstractTestLogging
     public void testExchangeDelete() throws Exception, IOException
     {
         //Ignore broker startup messages
-        _monitor.reset();
+        _monitor.markDiscardPoint();
 
         //create the exchange by creating a consumer
         _session.createConsumer(_queue);
@@ -220,7 +220,7 @@ public class ExchangeLoggingTest extends AbstractTestLogging
     public void testDiscardedMessage() throws Exception
     {
         //Ignore broker startup messages
-        _monitor.reset();
+        _monitor.markDiscardPoint();
 
         if (!isBroker010())
         {
diff --git a/qpid/java/systests/src/main/java/org/apache/qpid/server/logging/QueueLoggingTest.java b/qpid/java/systests/src/main/java/org/apache/qpid/server/logging/QueueLoggingTest.java
index b8a42c0ab3..76ebda0ebd 100644
--- a/qpid/java/systests/src/main/java/org/apache/qpid/server/logging/QueueLoggingTest.java
+++ b/qpid/java/systests/src/main/java/org/apache/qpid/server/logging/QueueLoggingTest.java
@@ -53,7 +53,7 @@ public class QueueLoggingTest extends AbstractTestLogging
     {
         super.setUp();
         //Remove broker startup logging messages
-        _monitor.reset();
+        _monitor.markDiscardPoint();
         
         _connection = getConnection();
         _session = _connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
diff --git a/qpid/java/systests/src/main/java/org/apache/qpid/server/logging/SubscriptionLoggingTest.java b/qpid/java/systests/src/main/java/org/apache/qpid/server/logging/SubscriptionLoggingTest.java
index 6e156f091e..b6efe53580 100644
--- a/qpid/java/systests/src/main/java/org/apache/qpid/server/logging/SubscriptionLoggingTest.java
+++ b/qpid/java/systests/src/main/java/org/apache/qpid/server/logging/SubscriptionLoggingTest.java
@@ -58,7 +58,7 @@ public class SubscriptionLoggingTest extends AbstractTestLogging
     {
         super.setUp();
         //Remove broker startup logging messages
-        _monitor.reset();
+        _monitor.markDiscardPoint();
 
         _connection = getConnection();
 
diff --git a/qpid/java/systests/src/main/java/org/apache/qpid/server/queue/ProducerFlowControlTest.java b/qpid/java/systests/src/main/java/org/apache/qpid/server/queue/ProducerFlowControlTest.java
index f78b327209..a724e6c66e 100644
--- a/qpid/java/systests/src/main/java/org/apache/qpid/server/queue/ProducerFlowControlTest.java
+++ b/qpid/java/systests/src/main/java/org/apache/qpid/server/queue/ProducerFlowControlTest.java
@@ -64,7 +64,7 @@ public class ProducerFlowControlTest extends AbstractTestLogging
         _jmxUtilConnected=false;
         super.setUp();
 
-        _monitor.reset();
+        _monitor.markDiscardPoint();
 
         producerConnection = getConnection();
         producerSession = producerConnection.createSession(false, Session.AUTO_ACKNOWLEDGE);
diff --git a/qpid/java/systests/src/main/java/org/apache/qpid/server/security/acl/AbstractACLTestCase.java b/qpid/java/systests/src/main/java/org/apache/qpid/server/security/acl/AbstractACLTestCase.java
index 8aa5b6d9de..32b0185f88 100644
--- a/qpid/java/systests/src/main/java/org/apache/qpid/server/security/acl/AbstractACLTestCase.java
+++ b/qpid/java/systests/src/main/java/org/apache/qpid/server/security/acl/AbstractACLTestCase.java
@@ -83,14 +83,8 @@ public abstract class AbstractACLTestCase extends QpidBrokerTestCase implements
     @Override
     public void setUp() throws Exception
     {
-        if (QpidHome == null)
-        {
-            fail("QPID_HOME not set");
-        }
-
         // Initialise ACLs.
-        _configFile = new File(QpidHome, "etc" + File.separator + getConfig());
-        
+        _configFile = new File("build" + File.separator + "etc" + File.separator + getConfig());
         // Initialise ACL files
         for (String virtualHost : getHostList())
         {
@@ -156,7 +150,7 @@ public abstract class AbstractACLTestCase extends QpidBrokerTestCase implements
      */
     public void setUpACLFile(String virtualHost) throws IOException, ConfigurationException
     {
-        String path = QpidHome + File.separator + "etc";
+        String path = "build" + File.separator + "etc";
         String className = StringUtils.substringBeforeLast(getClass().getSimpleName().toLowerCase(), "test");
         String testName = StringUtils.substringAfter(getName(), "test").toLowerCase();
         
diff --git a/qpid/java/systests/src/main/java/org/apache/qpid/server/security/acl/ExternalACLTest.java b/qpid/java/systests/src/main/java/org/apache/qpid/server/security/acl/ExternalACLTest.java
index 782ca22965..254e1fe6ac 100644
--- a/qpid/java/systests/src/main/java/org/apache/qpid/server/security/acl/ExternalACLTest.java
+++ b/qpid/java/systests/src/main/java/org/apache/qpid/server/security/acl/ExternalACLTest.java
@@ -91,9 +91,9 @@ public class ExternalACLTest extends AbstractACLTestCase
             //send a message to each queue (also causing an exchange declare)
             MessageProducer sender = ((AMQSession<?, ?>) sess).createProducer(null);
             ((org.apache.qpid.jms.MessageProducer) sender).send(namedQueue, sess.createTextMessage("test"),
-                                                                DeliveryMode.NON_PERSISTENT, 0, 0L, false, false, true);
+                                                                DeliveryMode.NON_PERSISTENT, 0, 0L, false, false);
             ((org.apache.qpid.jms.MessageProducer) sender).send(tempQueue, sess.createTextMessage("test"),
-                                                                DeliveryMode.NON_PERSISTENT, 0, 0L, false, false, true);
+                                                                DeliveryMode.NON_PERSISTENT, 0, 0L, false, false);
 
             //consume the messages from the queues
             consumer.receive(2000);
@@ -309,7 +309,7 @@ public class ExternalACLTest extends AbstractACLTestCase
             // before we finish this test. Message is set !immed !mand as the queue is invalid so want to test ACLs not
             // queue existence.
             ((org.apache.qpid.jms.MessageProducer) sender).send(queue, sess.createTextMessage("test"),
-                                                                DeliveryMode.NON_PERSISTENT, 0, 0L, false, false, true);
+                                                                DeliveryMode.NON_PERSISTENT, 0, 0L, false, false);
 
             conn.close();
         }
@@ -337,7 +337,7 @@ public class ExternalACLTest extends AbstractACLTestCase
             // before we finish this test. Message is set !immed !mand as the queue is invalid so want to test ACLs not
             // queue existence.
             ((org.apache.qpid.jms.MessageProducer) sender).send(queue, session.createTextMessage("test"),
-                                                                DeliveryMode.NON_PERSISTENT, 0, 0L, false, false, true);
+                                                                DeliveryMode.NON_PERSISTENT, 0, 0L, false, false);
 
             // Test the connection with a valid consumer
             // This may fail as the session may be closed before the queue or the consumer created.
@@ -608,7 +608,7 @@ public class ExternalACLTest extends AbstractACLTestCase
             // before we finish this test. Message is set !immed !mand as the queue is invalid so want to test ACLs not
             // queue existence.
             ((org.apache.qpid.jms.MessageProducer) sender).send(queue, session.createTextMessage("test"),
-                                                                DeliveryMode.NON_PERSISTENT, 0, 0L, false, false, true);
+                                                                DeliveryMode.NON_PERSISTENT, 0, 0L, false, false);
 
             // Test the connection with a valid consumer
             // This may not work as the session may be closed before the queue or consumer creation can occur.
diff --git a/qpid/java/systests/src/main/java/org/apache/qpid/server/security/firewall/FirewallConfigTest.java b/qpid/java/systests/src/main/java/org/apache/qpid/server/security/firewall/FirewallConfigTest.java
index 2d99a44532..044a0af335 100644
--- a/qpid/java/systests/src/main/java/org/apache/qpid/server/security/firewall/FirewallConfigTest.java
+++ b/qpid/java/systests/src/main/java/org/apache/qpid/server/security/firewall/FirewallConfigTest.java
@@ -35,16 +35,8 @@ public class FirewallConfigTest extends QpidBrokerTestCase
     @Override
     protected void setUp() throws Exception
     {
-        // do setup
-        final String QPID_HOME = System.getProperty("QPID_HOME");
-
-        if (QPID_HOME == null)
-        {
-            fail("QPID_HOME not set");
-        }
-
         // Setup initial config file.
-        _configFile = new File(QPID_HOME, "etc/config-systests-firewall.xml");
+        _configFile = new File("build/etc/config-systests-firewall.xml");
         
         // Setup temporary config file
         _tmpConfig = File.createTempFile("config-systests-firewall", ".xml");
@@ -86,7 +78,7 @@ public class FirewallConfigTest extends QpidBrokerTestCase
     public void testVhostAllowBrokerDeny() throws Exception
     {
 
-        _configFile = new File(System.getProperty("QPID_HOME"), "etc/config-systests-firewall-2.xml");
+        _configFile = new File("build/etc/config-systests-firewall-2.xml");
         
         super.setUp();
         
@@ -119,7 +111,7 @@ public class FirewallConfigTest extends QpidBrokerTestCase
     
     public void testVhostDenyBrokerAllow() throws Exception
     {
-        _configFile = new File(System.getProperty("QPID_HOME"), "etc/config-systests-firewall-3.xml");
+        _configFile = new File("build/etc/config-systests-firewall-3.xml");
         
         super.setUp();
         
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 b5bb74327e..8c3c247e2b 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
@@ -1116,10 +1116,10 @@ public class AddressBasedDestinationTest extends QpidBrokerTestCase
     public void testDestinationOnSend() throws Exception
     {
     	Session ssn = _connection.createSession(false,Session.CLIENT_ACKNOWLEDGE);
-        MessageConsumer cons = ssn.createConsumer(ssn.createTopic("amq.topic/test"));
+        MessageConsumer cons = ssn.createConsumer(ssn.createTopic("ADDR:amq.topic/test"));
         MessageProducer prod = ssn.createProducer(null);
         
-        Queue queue = ssn.createQueue("amq.topic/test");
+        Queue queue = ssn.createQueue("ADDR:amq.topic/test");
         prod.send(queue,ssn.createTextMessage("A"));
         
         Message msg = cons.receive(1000);
@@ -1147,7 +1147,7 @@ public class AddressBasedDestinationTest extends QpidBrokerTestCase
 		Destination replyToDest = AMQDestination.createDestination(replyTo);
 	    MessageConsumer replyToCons = session.createConsumer(replyToDest);
 	    		    			
-		Destination dest = session.createQueue("amq.direct/test");
+		Destination dest = session.createQueue("ADDR:amq.direct/test");
 					
 		MessageConsumer cons = session.createConsumer(dest);
 		MessageProducer prod = session.createProducer(dest);
diff --git a/qpid/java/systests/src/main/java/org/apache/qpid/test/client/timeouts/SyncWaitTimeoutDelayTest.java b/qpid/java/systests/src/main/java/org/apache/qpid/test/client/timeouts/SyncWaitTimeoutDelayTest.java
index 1a23eee8ab..6189c37306 100644
--- a/qpid/java/systests/src/main/java/org/apache/qpid/test/client/timeouts/SyncWaitTimeoutDelayTest.java
+++ b/qpid/java/systests/src/main/java/org/apache/qpid/test/client/timeouts/SyncWaitTimeoutDelayTest.java
@@ -63,7 +63,7 @@ public class SyncWaitTimeoutDelayTest extends SyncWaitDelayTest
         catch (JMSException e)
         {
             assertTrue("Wrong exception type received.", e.getLinkedException() instanceof AMQTimeoutException);
-            assertTrue("Wrong message received on exception.", e.getMessage().startsWith("Failed to commit"));
+            assertTrue("Wrong message received on exception.", e.getMessage().startsWith("Exception during commit"));
             // As we are using Nano time ensure to multiply up the millis.            
             assertTrue("Timeout was more than 30s default", (System.nanoTime() - start) < (1000000L * 1000 * 30));
         }
diff --git a/qpid/java/systests/src/main/java/org/apache/qpid/test/unit/ack/ClientAcknowledgeTest.java b/qpid/java/systests/src/main/java/org/apache/qpid/test/unit/ack/ClientAcknowledgeTest.java
new file mode 100644
index 0000000000..06be5cf456
--- /dev/null
+++ b/qpid/java/systests/src/main/java/org/apache/qpid/test/unit/ack/ClientAcknowledgeTest.java
@@ -0,0 +1,82 @@
+/*
+ * 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.test.unit.ack;
+
+import javax.jms.Connection;
+import javax.jms.Message;
+import javax.jms.MessageConsumer;
+import javax.jms.MessageProducer;
+import javax.jms.Queue;
+import javax.jms.Session;
+
+import org.apache.qpid.test.utils.QpidBrokerTestCase;
+
+public class ClientAcknowledgeTest extends QpidBrokerTestCase
+{
+    private static final long ONE_DAY_MS = 1000l * 60 * 60 * 24;
+    private Connection _connection;
+    private Queue _queue;
+    private Session _consumerSession;
+    private MessageConsumer _consumer;
+    private MessageProducer _producer;
+
+    @Override
+    protected void setUp() throws Exception
+    {
+        super.setUp();
+        _queue = getTestQueue();
+        _connection = getConnection();
+    }
+
+    /**
+     * Test that message.acknowledge actually acknowledges, regardless of
+     * the flusher thread period, by restarting the broker after calling
+     * acknowledge, and then verifying after restart that the message acked
+     * is no longer present. This test requires a persistent store.
+     */
+    public void testClientAckWithLargeFlusherPeriod() throws Exception
+    {
+        setTestClientSystemProperty("qpid.session.max_ack_delay", Long.toString(ONE_DAY_MS));
+        _consumerSession = _connection.createSession(false, Session.CLIENT_ACKNOWLEDGE);
+        _consumer = _consumerSession.createConsumer(_queue);
+        _connection.start();
+
+        _producer = _consumerSession.createProducer(_queue);
+        _producer.send(createNextMessage(_consumerSession, 1));
+        _producer.send(createNextMessage(_consumerSession, 2));
+
+        Message message = _consumer.receive(1000l);
+        assertNotNull("Message has not been received", message);
+        assertEquals("Unexpected message is received", 1, message.getIntProperty(INDEX));
+        message.acknowledge();
+
+        //restart broker to allow verification of the acks
+        //without explicitly closing connection (which acks)
+        restartBroker();
+
+        // try to receive the message again, which should fail (as it was ackd)
+        _connection = getConnection();
+        _connection.start();
+        _consumerSession = _connection.createSession(false, Session.CLIENT_ACKNOWLEDGE);
+        _consumer = _consumerSession.createConsumer(_queue);
+        message = _consumer.receive(1000l);
+        assertNotNull("Message has not been received", message);
+        assertEquals("Unexpected message is received", 2, message.getIntProperty(INDEX));
+    }
+}
diff --git a/qpid/java/systests/src/main/java/org/apache/qpid/test/unit/ack/RecoverTest.java b/qpid/java/systests/src/main/java/org/apache/qpid/test/unit/ack/RecoverTest.java
index 5e7ba5482d..66ca1d8345 100644
--- a/qpid/java/systests/src/main/java/org/apache/qpid/test/unit/ack/RecoverTest.java
+++ b/qpid/java/systests/src/main/java/org/apache/qpid/test/unit/ack/RecoverTest.java
@@ -46,7 +46,7 @@ public class RecoverTest extends FailoverBaseCase
 {
     static final Logger _logger = LoggerFactory.getLogger(RecoverTest.class);
 
-    private Exception _error;
+    private volatile Exception _error;
     private AtomicInteger count;
 
     protected AMQConnection _connection;
@@ -249,14 +249,13 @@ public class RecoverTest extends FailoverBaseCase
                     {
                         if (!message.getJMSRedelivered())
                         {
-                            setError(
-                                    new Exception("Message not marked as redelivered on what should be second delivery attempt"));
+                            setError(new Exception("Message not marked as redelivered on what should be second delivery attempt"));
                         }
                     }
                     else
                     {
-                        System.err.println(message);
-                        fail("Message delivered too many times!: " + count);
+                        _logger.error(message.toString());
+                        setError(new Exception("Message delivered too many times!: " + count));
                     }
                 }
                 catch (JMSException e)
diff --git a/qpid/java/systests/src/main/java/org/apache/qpid/test/unit/basic/close/CloseTest.java b/qpid/java/systests/src/main/java/org/apache/qpid/test/unit/basic/close/CloseTest.java
index c6b8069300..3c7962d873 100644
--- a/qpid/java/systests/src/main/java/org/apache/qpid/test/unit/basic/close/CloseTest.java
+++ b/qpid/java/systests/src/main/java/org/apache/qpid/test/unit/basic/close/CloseTest.java
@@ -21,14 +21,13 @@
 package org.apache.qpid.test.unit.basic.close;
 import org.apache.qpid.test.utils.QpidBrokerTestCase;
 import org.apache.qpid.client.AMQConnection;
-import org.apache.qpid.client.AMQQueue;
-import org.apache.qpid.url.AMQBindingURL;
 
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 import javax.jms.MessageConsumer;
 import javax.jms.MessageProducer;
+import javax.jms.Queue;
 import javax.jms.Session;
 
 public class CloseTest extends QpidBrokerTestCase
@@ -41,7 +40,7 @@ public class CloseTest extends QpidBrokerTestCase
 
         Session session = connection.createQueueSession(false, Session.AUTO_ACKNOWLEDGE);
 
-        AMQQueue queue = new AMQQueue(new AMQBindingURL("test-queue"));
+        Queue queue = session.createQueue("test-queue");
         MessageConsumer consumer = session.createConsumer(queue);
 
         MessageProducer producer_not_used_but_created_for_testing = session.createProducer(queue);
diff --git a/qpid/java/systests/src/main/java/org/apache/qpid/test/unit/client/AMQConnectionTest.java b/qpid/java/systests/src/main/java/org/apache/qpid/test/unit/client/AMQConnectionTest.java
index e79fe69199..53a7533869 100644
--- a/qpid/java/systests/src/main/java/org/apache/qpid/test/unit/client/AMQConnectionTest.java
+++ b/qpid/java/systests/src/main/java/org/apache/qpid/test/unit/client/AMQConnectionTest.java
@@ -20,16 +20,6 @@
  */
 package org.apache.qpid.test.unit.client;
 
-import java.io.BufferedWriter;
-import java.io.InputStreamReader;
-import java.io.LineNumberReader;
-import java.io.OutputStreamWriter;
-import java.io.PrintWriter;
-import java.util.Map;
-import java.util.concurrent.atomic.AtomicBoolean;
-
-import javax.jms.Connection;
-import javax.jms.ExceptionListener;
 import javax.jms.JMSException;
 import javax.jms.Message;
 import javax.jms.MessageConsumer;
diff --git a/qpid/java/systests/src/main/java/org/apache/qpid/test/unit/client/AMQSSLConnectionTest.java b/qpid/java/systests/src/main/java/org/apache/qpid/test/unit/client/AMQSSLConnectionTest.java
index 53a433c543..5f3daa407a 100644
--- a/qpid/java/systests/src/main/java/org/apache/qpid/test/unit/client/AMQSSLConnectionTest.java
+++ b/qpid/java/systests/src/main/java/org/apache/qpid/test/unit/client/AMQSSLConnectionTest.java
@@ -26,9 +26,9 @@ import org.apache.qpid.client.AMQConnectionURL;
 
 public class AMQSSLConnectionTest extends AMQConnectionTest
 {
-    private static final String KEYSTORE = TEST_RESOURCES_DIR + "/ssl/java_client_keystore.jks";
+    private static final String KEYSTORE = "test-profiles/test_resources/ssl/java_client_keystore.jks";
     private static final String KEYSTORE_PASSWORD = "password";
-    private static final String TRUSTSTORE = TEST_RESOURCES_DIR + "/ssl/java_client_truststore.jks";
+    private static final String TRUSTSTORE = "test-profiles/test_resources/ssl/java_client_truststore.jks";
     private static final String TRUSTSTORE_PASSWORD = "password";
     
     @Override
diff --git a/qpid/java/systests/src/main/java/org/apache/qpid/test/unit/client/message/ObjectMessageTest.java b/qpid/java/systests/src/main/java/org/apache/qpid/test/unit/client/message/ObjectMessageTest.java
index fd28b86762..8ad8fa77d7 100644
--- a/qpid/java/systests/src/main/java/org/apache/qpid/test/unit/client/message/ObjectMessageTest.java
+++ b/qpid/java/systests/src/main/java/org/apache/qpid/test/unit/client/message/ObjectMessageTest.java
@@ -20,14 +20,10 @@
  */
 package org.apache.qpid.test.unit.client.message;
 
-import org.apache.qpid.client.AMQConnection;
-import org.apache.qpid.client.AMQDestination;
-import org.apache.qpid.client.AMQQueue;
-import org.apache.qpid.client.AMQSession;
-import org.apache.qpid.test.utils.QpidBrokerTestCase;
-
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
+import java.io.Serializable;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.HashMap;
 
 import javax.jms.JMSException;
 import javax.jms.Message;
@@ -35,10 +31,13 @@ import javax.jms.MessageListener;
 import javax.jms.MessageProducer;
 import javax.jms.ObjectMessage;
 
-import java.io.Serializable;
-import java.util.ArrayList;
-import java.util.Arrays;
-import java.util.HashMap;
+import org.apache.qpid.client.AMQConnection;
+import org.apache.qpid.client.AMQDestination;
+import org.apache.qpid.client.AMQQueue;
+import org.apache.qpid.client.AMQSession;
+import org.apache.qpid.test.utils.QpidBrokerTestCase;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 public class ObjectMessageTest extends QpidBrokerTestCase implements MessageListener
 {
@@ -67,7 +66,7 @@ public class ObjectMessageTest extends QpidBrokerTestCase implements MessageList
         connection.start();
 
         // create a publisher
-        producer = session.createProducer(destination, false, false, true);
+        producer = session.createProducer(destination, false, false);
         A a1 = new A(1, "A");
         A a2 = new A(2, "a");
         B b = new B(1, "B");
diff --git a/qpid/java/systests/src/main/java/org/apache/qpid/test/unit/publish/DirtyTransactedPublishTest.java b/qpid/java/systests/src/main/java/org/apache/qpid/test/unit/publish/DirtyTransactedPublishTest.java
deleted file mode 100644
index 3ec7937812..0000000000
--- a/qpid/java/systests/src/main/java/org/apache/qpid/test/unit/publish/DirtyTransactedPublishTest.java
+++ /dev/null
@@ -1,403 +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.qpid.test.unit.publish;
-
-import org.apache.qpid.client.AMQConnection;
-import org.apache.qpid.client.AMQDestination;
-import org.apache.qpid.client.AMQSession;
-import org.apache.qpid.jms.ConnectionListener;
-import org.apache.qpid.test.utils.FailoverBaseCase;
-
-import javax.jms.Connection;
-import javax.jms.JMSException;
-import javax.jms.Message;
-import javax.jms.MessageConsumer;
-import javax.jms.MessageListener;
-import javax.jms.MessageProducer;
-import javax.jms.Queue;
-import javax.jms.Session;
-import javax.jms.TransactionRolledBackException;
-import java.util.concurrent.CountDownLatch;
-import java.util.concurrent.TimeUnit;
-import java.util.concurrent.atomic.AtomicReference;
-
-/**
- * QPID-1816 : Whilst testing Acknoledgement after failover this completes testing
- * of the client after failover. When we have a dirty session we should receive
- * an error if we attempt to publish. This test ensures that both in the synchronous
- * and asynchronous message delivery paths we receive the expected exceptions at
- * the expected time.
- */
-public class DirtyTransactedPublishTest extends FailoverBaseCase implements ConnectionListener
-{
-    protected CountDownLatch _failoverCompleted = new CountDownLatch(1);
-
-    protected int NUM_MESSAGES;
-    protected Connection _connection;
-    protected Queue _queue;
-    protected Session _consumerSession;
-    protected MessageConsumer _consumer;
-    protected MessageProducer _producer;
-
-    private static final String MSG = "MSG";
-    private static final String SEND_FROM_ON_MESSAGE_TEXT = "sendFromOnMessage";
-    protected CountDownLatch _receviedAll;
-    protected AtomicReference<Exception> _causeOfFailure = new AtomicReference<Exception>(null);
-
-    @Override
-    protected void setUp() throws Exception
-    {
-        super.setUp();
-        NUM_MESSAGES = 10;
-
-        _queue = getTestQueue();
-
-        //Create Producer put some messages on the queue
-        _connection = getConnection();
-    }
-
-    /**
-     * Initialise the test variables
-     * @param transacted is this a transacted test
-     * @param mode if not trasacted then what ack mode to use
-     * @throws Exception if there is a setup issue.
-     */
-    protected void init(boolean transacted, int mode) throws Exception
-    {
-        _consumerSession = _connection.createSession(transacted, mode);
-        _consumer = _consumerSession.createConsumer(_queue);
-        _producer = _consumerSession.createProducer(_queue);
-
-        // These should all end up being prefetched by session
-        sendMessage(_consumerSession, _queue, 1);
-
-        assertEquals("Wrong number of messages on queue", 1,
-                     ((AMQSession) _consumerSession).getQueueDepth((AMQDestination) _queue));
-    }
-
-    /**
-     * If a transacted session has failed over whilst it has uncommitted sent
-     * data then we need to throw a TransactedRolledbackException on commit()
-     *
-     * The alternative would be to maintain a replay buffer so that the message
-     * could be resent. This is not currently implemented
-     *
-     * @throws Exception if something goes wrong.
-     */
-    public void testDirtySendingSynchronousTransacted() throws Exception
-    {
-        Session producerSession = _connection.createSession(true, Session.SESSION_TRANSACTED);
-
-        // Ensure we get failover notifications
-        ((AMQConnection) _connection).setConnectionListener(this);
-
-        MessageProducer producer = producerSession.createProducer(_queue);
-
-        // Create and send message 0
-        Message msg = producerSession.createMessage();
-        msg.setIntProperty(INDEX, 0);
-        producer.send(msg);
-
-        // DON'T commit message .. fail connection
-
-        failBroker(getFailingPort());
-
-        // Ensure destination exists for sending
-        producerSession.createConsumer(_queue).close();
-
-        // Send the next message
-        msg.setIntProperty(INDEX, 1);
-        try
-        {
-            producer.send(msg);
-            fail("Should fail with Qpid as we provide early warning of the dirty session via a JMSException.");
-        }
-        catch (JMSException jmse)
-        {
-            assertEquals("Early warning of dirty session not correct",
-                         "Failover has occurred and session is dirty so unable to send.", jmse.getMessage());
-        }
-
-        // Ignore that the session is dirty and attempt to commit to validate the
-        // exception is thrown. AND that the above failure notification did NOT
-        // clean up the session.
-
-        try
-        {
-            producerSession.commit();
-            fail("Session is dirty we should get an TransactionRolledBackException");
-        }
-        catch (TransactionRolledBackException trbe)
-        {
-            // Normal path.
-        }
-
-        // Resending of messages should now work ok as the commit was forcilbly rolledback
-        msg.setIntProperty(INDEX, 0);
-        producer.send(msg);
-        msg.setIntProperty(INDEX, 1);
-        producer.send(msg);
-
-        producerSession.commit();
-
-        assertEquals("Wrong number of messages on queue", 2,
-                     ((AMQSession) producerSession).getQueueDepth((AMQDestination) _queue));
-    }
-
-    /**
-     * If a transacted session has failed over whilst it has uncommitted sent
-     * data then we need to throw a TransactedRolledbackException on commit()
-     *
-     * The alternative would be to maintain a replay buffer so that the message
-     * could be resent. This is not currently implemented
-     *
-     * @throws Exception if something goes wrong.
-     */
-    public void testDirtySendingOnMessageTransacted() throws Exception
-    {
-        NUM_MESSAGES = 1;
-        _receviedAll = new CountDownLatch(NUM_MESSAGES);
-        ((AMQConnection) _connection).setConnectionListener(this);
-
-        init(true, Session.SESSION_TRANSACTED);
-
-        _consumer.setMessageListener(new MessageListener()
-        {
-
-            public void onMessage(Message message)
-            {
-                try
-                {
-                    // Create and send message 0
-                    Message msg = _consumerSession.createMessage();
-                    msg.setIntProperty(INDEX, 0);
-                    _producer.send(msg);
-
-                    // DON'T commit message .. fail connection
-
-                    failBroker(getFailingPort());
-
-                    // rep
-                    repopulateBroker();
-
-                    // Destination will exist as this failBroker will populate
-                    // the queue with 1 message
-
-                    // Send the next message
-                    msg.setIntProperty(INDEX, 1);
-                    try
-                    {
-                        _producer.send(msg);
-                        fail("Should fail with Qpid as we provide early warning of the dirty session via a JMSException.");
-                    }
-                    catch (JMSException jmse)
-                    {
-                        assertEquals("Early warning of dirty session not correct",
-                                     "Failover has occurred and session is dirty so unable to send.", jmse.getMessage());
-                    }
-
-                    // Ignore that the session is dirty and attempt to commit to validate the
-                    // exception is thrown. AND that the above failure notification did NOT
-                    // clean up the session.
-
-                    try
-                    {
-                        _consumerSession.commit();
-                        fail("Session is dirty we should get an TransactionRolledBackException");
-                    }
-                    catch (TransactionRolledBackException trbe)
-                    {
-                        // Normal path.
-                    }
-
-                    // Resend messages
-                    msg.setIntProperty(INDEX, 0);
-                    msg.setStringProperty(MSG, SEND_FROM_ON_MESSAGE_TEXT);
-                    _producer.send(msg);
-                    msg.setIntProperty(INDEX, 1);
-                    msg.setStringProperty(MSG, SEND_FROM_ON_MESSAGE_TEXT);
-                    _producer.send(msg);
-
-                    _consumerSession.commit();
-
-                    // Stop this consumer .. can't do _consumer.stop == DEADLOCK
-                    // this doesn't seem to stop dispatcher running
-                    _connection.stop();
-
-                    // Signal that the onMessage send part of test is complete
-                    // main thread can validate that messages are correct
-                    _receviedAll.countDown();
-
-                }
-                catch (Exception e)
-                {
-                    fail(e);
-                }
-
-            }
-
-        });
-
-        _connection.start();
-
-        if (!_receviedAll.await(10000L, TimeUnit.MILLISECONDS))
-        {
-            // Check to see if we ended due to an exception in the onMessage handler
-            Exception cause = _causeOfFailure.get();
-            if (cause != null)
-            {
-                cause.printStackTrace();
-                fail(cause.getMessage());
-            }
-            else
-            {
-                fail("All messages not received:" + _receviedAll.getCount() + "/" + NUM_MESSAGES);
-            }
-        }
-
-        // Check to see if we ended due to an exception in the onMessage handler
-        Exception cause = _causeOfFailure.get();
-        if (cause != null)
-        {
-            cause.printStackTrace();
-            fail(cause.getMessage());
-        }
-
-        _consumer.close();
-        _consumerSession.close();
-
-        _consumerSession = _connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
-
-        _connection.start();
-
-        // Validate that we could send the messages as expected.
-        assertEquals("Wrong number of messages on queue", 3,
-                     ((AMQSession) _consumerSession).getQueueDepth((AMQDestination) _queue));
-
-        MessageConsumer consumer = _consumerSession.createConsumer(_queue);
-
-        //Validate the message sent to setup the failed over broker.
-        Message message = consumer.receive(1000);
-        assertNotNull("Message " + 0 + " not received.", message);
-        assertEquals("Incorrect message received", 0, message.getIntProperty(INDEX));
-
-        // Validate the two messages sent from within the onMessage
-        for (int index = 0; index <= 1; index++)
-        {
-            message = consumer.receive(1000);
-            assertNotNull("Message " + index + " not received.", message);
-            assertEquals("Incorrect message received", index, message.getIntProperty(INDEX));
-            assertEquals("Incorrect message text for message:" + index, SEND_FROM_ON_MESSAGE_TEXT, message.getStringProperty(MSG));
-        }
-
-        assertNull("Extra message received.", consumer.receiveNoWait());
-
-        _consumerSession.close();
-
-        assertEquals("Wrong number of messages on queue", 0,
-                     ((AMQSession) getConnection().createSession(false, Session.AUTO_ACKNOWLEDGE)).getQueueDepth((AMQDestination) _queue));
-    }
-
-    private void repopulateBroker() throws Exception
-    {
-        // Repopulate this new broker so we can test what happends after failover
-
-        //Get the connection to the first (main port) broker.
-        Connection connection = getConnection();
-        // Use a transaction to send messages so we can be sure they arrive.
-        Session session = connection.createSession(true, Session.SESSION_TRANSACTED);
-        // ensure destination is created.
-        session.createConsumer(_queue).close();
-
-        sendMessage(session, _queue, NUM_MESSAGES);
-
-        assertEquals("Wrong number of messages on queue", NUM_MESSAGES,
-                     ((AMQSession) session).getQueueDepth((AMQDestination) _queue));
-
-        connection.close();
-    }
-
-    // AMQConnectionListener Interface.. used so we can validate that we
-    // actually failed over.
-
-    public void bytesSent(long count)
-    {
-    }
-
-    public void bytesReceived(long count)
-    {
-    }
-
-    public boolean preFailover(boolean redirect)
-    {
-        //Allow failover
-        return true;
-    }
-
-    public boolean preResubscribe()
-    {
-        //Allow failover
-        return true;
-    }
-
-    public void failoverComplete()
-    {
-        _failoverCompleted.countDown();
-    }
-
-    /**
-     * Override so we can block until failover has completd
-     *
-     * @param port int the port of the broker to fail.
-     */
-    @Override
-    public void failBroker(int port)
-    {
-        super.failBroker(port);
-
-        try
-        {
-            if (!_failoverCompleted.await(DEFAULT_FAILOVER_TIME, TimeUnit.MILLISECONDS))
-            {
-                fail("Failover did not occur in specified time:" + DEFAULT_FAILOVER_TIME);
-            }
-        }
-        catch (InterruptedException e)
-        {
-            fail("Failover was interuppted");
-        }
-    }
-
-    /**
-     * Pass the given exception back to the waiting thread to fail the test run.
-     *
-     * @param e The exception that is causing the test to fail.
-     */
-    protected void fail(Exception e)
-    {
-        _causeOfFailure.set(e);
-        // End the test.
-        while (_receviedAll.getCount() != 0)
-        {
-            _receviedAll.countDown();
-        }
-    }
-}
diff --git a/qpid/java/systests/src/main/java/org/apache/qpid/test/unit/transacted/TransactionTimeoutConfigurationTest.java b/qpid/java/systests/src/main/java/org/apache/qpid/test/unit/transacted/TransactionTimeoutConfigurationTest.java
deleted file mode 100644
index 46e5d214f5..0000000000
--- a/qpid/java/systests/src/main/java/org/apache/qpid/test/unit/transacted/TransactionTimeoutConfigurationTest.java
+++ /dev/null
@@ -1,82 +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.qpid.test.unit.transacted;
-
-/**
- * This verifies that changing the {@code transactionTimeout} configuration will alter
- * the behaviour of the transaction open and idle logging, and that when the connection
- * will be closed.
- */
-public class TransactionTimeoutConfigurationTest extends TransactionTimeoutTestCase
-{
-    @Override
-    protected void configure() throws Exception
-    {
-        // Setup housekeeping every second
-        setConfigurationProperty("virtualhosts.virtualhost." + VIRTUALHOST + ".housekeeping.checkPeriod", "100");
-        
-        // Set transaction timout properties.
-        setConfigurationProperty("virtualhosts.virtualhost." + VIRTUALHOST + ".transactionTimeout.openWarn", "200");
-        setConfigurationProperty("virtualhosts.virtualhost." + VIRTUALHOST + ".transactionTimeout.openClose", "1000");
-        setConfigurationProperty("virtualhosts.virtualhost." + VIRTUALHOST + ".transactionTimeout.idleWarn", "100");
-        setConfigurationProperty("virtualhosts.virtualhost." + VIRTUALHOST + ".transactionTimeout.idleClose", "500");
-    }
-
-    public void testProducerIdleCommit() throws Exception
-    {
-        try
-        {
-            send(5, 0);
-            
-            sleep(2.0f);
-
-            _psession.commit();
-            fail("should fail");
-        }
-        catch (Exception e)
-        {
-            _exception = e;
-        }
-        
-        monitor(5, 0);
-        
-        check(IDLE);
-    }
-
-    public void testProducerOpenCommit() throws Exception
-    {
-        try
-        {
-            send(5, 0.3f);
-
-            _psession.commit();
-            fail("should fail");
-        }
-        catch (Exception e)
-        {
-            _exception = e;
-        }
-        
-        monitor(6, 3);
-        
-        check(OPEN);
-    }
-}
diff --git a/qpid/java/systests/src/main/java/org/apache/qpid/test/unit/transacted/TransactionTimeoutDisabledTest.java b/qpid/java/systests/src/main/java/org/apache/qpid/test/unit/transacted/TransactionTimeoutDisabledTest.java
index db508143f9..fd8beffbe6 100644
--- a/qpid/java/systests/src/main/java/org/apache/qpid/test/unit/transacted/TransactionTimeoutDisabledTest.java
+++ b/qpid/java/systests/src/main/java/org/apache/qpid/test/unit/transacted/TransactionTimeoutDisabledTest.java
@@ -30,6 +30,8 @@ public class TransactionTimeoutDisabledTest extends TransactionTimeoutTestCase
     {
         // Setup housekeeping every second
         setConfigurationProperty("virtualhosts.virtualhost." + VIRTUALHOST + ".housekeeping.checkPeriod", "100");
+
+        // No transaction timeout configuration.
     }
 
     public void testProducerIdleCommit() throws Exception
@@ -47,7 +49,7 @@ public class TransactionTimeoutDisabledTest extends TransactionTimeoutTestCase
             fail("Should have succeeded");
         }
         
-        assertTrue("Listener should not have received exception", _caught.getCount() == 1);
+        assertEquals("Listener should not have received exception", 0, getNumberOfDeliveredExceptions());
         
         monitor(0, 0);
     }
@@ -65,7 +67,7 @@ public class TransactionTimeoutDisabledTest extends TransactionTimeoutTestCase
             fail("Should have succeeded");
         }
         
-        assertTrue("Listener should not have received exception", _caught.getCount() == 1);
+        assertEquals("Listener should not have received exception", 0, getNumberOfDeliveredExceptions());
         
         monitor(0, 0);
     }
diff --git a/qpid/java/systests/src/main/java/org/apache/qpid/test/unit/transacted/TransactionTimeoutTest.java b/qpid/java/systests/src/main/java/org/apache/qpid/test/unit/transacted/TransactionTimeoutTest.java
index c912d6a323..f554b0089e 100644
--- a/qpid/java/systests/src/main/java/org/apache/qpid/test/unit/transacted/TransactionTimeoutTest.java
+++ b/qpid/java/systests/src/main/java/org/apache/qpid/test/unit/transacted/TransactionTimeoutTest.java
@@ -30,152 +30,125 @@ package org.apache.qpid.test.unit.transacted;
  */
 public class TransactionTimeoutTest extends TransactionTimeoutTestCase
 {
-    public void testProducerIdle() throws Exception
+
+    protected void configure() throws Exception
     {
-        try
+        // Setup housekeeping every second
+        setConfigurationProperty("virtualhosts.virtualhost." + VIRTUALHOST + ".housekeeping.checkPeriod", "100");
+
+        if (getName().contains("ProducerIdle"))
         {
-            sleep(2.0f);
-    
-            _psession.commit();
+            setConfigurationProperty("virtualhosts.virtualhost." + VIRTUALHOST + ".transactionTimeout.openWarn", "0");
+            setConfigurationProperty("virtualhosts.virtualhost." + VIRTUALHOST + ".transactionTimeout.openClose", "0");
+            setConfigurationProperty("virtualhosts.virtualhost." + VIRTUALHOST + ".transactionTimeout.idleWarn", "500");
+            setConfigurationProperty("virtualhosts.virtualhost." + VIRTUALHOST + ".transactionTimeout.idleClose", "1500");
         }
-        catch (Exception e)
+        else if (getName().contains("ProducerOpen"))
         {
-            fail("Should have succeeded");
+            setConfigurationProperty("virtualhosts.virtualhost." + VIRTUALHOST + ".transactionTimeout.openWarn", "1000");
+            setConfigurationProperty("virtualhosts.virtualhost." + VIRTUALHOST + ".transactionTimeout.openClose", "2000");
+            setConfigurationProperty("virtualhosts.virtualhost." + VIRTUALHOST + ".transactionTimeout.idleWarn", "0");
+            setConfigurationProperty("virtualhosts.virtualhost." + VIRTUALHOST + ".transactionTimeout.idleClose", "0");
         }
-        
-        assertTrue("Listener should not have received exception", _caught.getCount() == 1);
-        
-        monitor(0, 0);
-    }
-    
-    public void testProducerIdleCommit() throws Exception
-    {
-        try
+        else
         {
-            send(5, 0);
-            
-            sleep(2.0f);
-    
-            _psession.commit();
-            fail("should fail");
+            setConfigurationProperty("virtualhosts.virtualhost." + VIRTUALHOST + ".transactionTimeout.openWarn", "1000");
+            setConfigurationProperty("virtualhosts.virtualhost." + VIRTUALHOST + ".transactionTimeout.openClose", "2000");
+            setConfigurationProperty("virtualhosts.virtualhost." + VIRTUALHOST + ".transactionTimeout.idleWarn", "500");
+            setConfigurationProperty("virtualhosts.virtualhost." + VIRTUALHOST + ".transactionTimeout.idleClose", "1000");
         }
-        catch (Exception e)
-        {
-            _exception = e;
-        }
-        
-        monitor(5, 0);
-        
-        check(IDLE);
     }
-    
-    public void testProducerOpenCommit() throws Exception
+
+    public void testProducerIdle() throws Exception
     {
-        try
-        {
-            send(6, 0.5f);
-    
-            _psession.commit();
-            fail("should fail");
-        }
-        catch (Exception e)
-        {
-            _exception = e;
-        }
-        
-        monitor(0, 10);
-        
-        check(OPEN);
+        sleep(2.0f);
+
+        _psession.commit();
+
+        assertEquals("Listener should not have received exception", 0, getNumberOfDeliveredExceptions());
+
+        monitor(0, 0);
     }
-    
-    public void testProducerIdleCommitTwice() throws Exception
+
+    public void testProducerIdleCommit() throws Exception
     {
+        send(5, 0);
+        // Idle for more than idleClose to generate idle-warns and cause a close.
+        sleep(2.0f);
+
         try
         {
-            send(5, 0);
-            
-            sleep(1.0f);
-            
             _psession.commit();
-            
-            send(5, 0);
-            
-            sleep(2.0f);
-    
-            _psession.commit();
-            fail("should fail");
+            fail("Exception not thrown");
         }
         catch (Exception e)
         {
             _exception = e;
         }
-        
+
         monitor(10, 0);
-        
+
         check(IDLE);
     }
-    
-    public void testProducerOpenCommitTwice() throws Exception
+
+    public void testProducerIdleCommitTwice() throws Exception
     {
+        send(5, 0);
+        // Idle for less than idleClose to generate idle-warns
+        sleep(1.0f);
+
+        _psession.commit();
+
+        send(5, 0);
+        // Now idle for more than idleClose to generate more idle-warns and cause a close.
+        sleep(2.0f);
+
         try
         {
-            send(5, 0);
-            
-            sleep(1.0f);
-            
             _psession.commit();
-            
-            send(6, 0.5f);
-    
-            _psession.commit();
-            fail("should fail");
+            fail("Exception not thrown");
         }
         catch (Exception e)
         {
             _exception = e;
         }
-        
-        // the presistent store generates more idle messages?
-        monitor(isBrokerStorePersistent() ? 10 : 5, 10);
-        
-        check(OPEN);
+
+        monitor(15, 0);
+
+        check(IDLE);
     }
-    
+
     public void testProducerIdleRollback() throws Exception
     {
+        send(5, 0);
+        // Now idle for more than idleClose to generate more idle-warns and cause a close.
+        sleep(2.0f);
         try
         {
-            send(5, 0);
-            
-            sleep(2.0f);
-    
             _psession.rollback();
-            fail("should fail");
+            fail("Exception not thrown");
         }
         catch (Exception e)
         {
             _exception = e;
         }
-        
-        monitor(5, 0);
-        
+
+        monitor(10, 0);
+
         check(IDLE);
     }
-    
+
     public void testProducerIdleRollbackTwice() throws Exception
     {
+        send(5, 0);
+        // Idle for less than idleClose to generate idle-warns
+        sleep(1.0f);
+        _psession.rollback();
+        send(5, 0);
+        // Now idle for more than idleClose to generate more idle-warns and cause a close.
+        sleep(2.0f);
         try
         {
-            send(5, 0);
-            
-            sleep(1.0f);
-            
-            _psession.rollback();
-            
-            send(5, 0);
-            
-            sleep(2.0f);
-    
             _psession.rollback();
             fail("should fail");
         }
@@ -183,153 +156,153 @@ public class TransactionTimeoutTest extends TransactionTimeoutTestCase
         {
             _exception = e;
         }
-        
-        monitor(10, 0);
+
+        monitor(15, 0);
         
         check(IDLE);
     }
-    
-    public void testConsumerCommitClose() throws Exception
+
+    public void testProducerOpenCommit() throws Exception
     {
         try
         {
-            send(1, 0);
-    
+            // Sleep between sends to cause open warns and then cause a close.
+            send(6, 0.5f);
             _psession.commit();
-    
-            expect(1, 0);
-            
-            _csession.commit();
-            
-            sleep(3.0f);
-    
-            _csession.close();
+            fail("Exception not thrown");
         }
         catch (Exception e)
         {
-            fail("should have succeeded: " + e.getMessage());
+            _exception = e;
         }
-        
-        assertTrue("Listener should not have received exception", _caught.getCount() == 1);
-        
-        monitor(0, 0);
+
+        monitor(0, 10);
+
+        check(OPEN);
     }
     
-    public void testConsumerIdleReceiveCommit() throws Exception
+    public void testProducerOpenCommitTwice() throws Exception
     {
+        send(5, 0);
+        sleep(1.0f);
+        _psession.commit();
+
         try
         {
-            send(1, 0);
-    
+            // Now sleep between sends to cause open warns and then cause a close.
+            send(6, 0.5f);
             _psession.commit();
-    
-            sleep(2.0f);
-            
-            expect(1, 0);
-            
-            sleep(2.0f);
-    
-            _csession.commit();
+            fail("Exception not thrown");
         }
         catch (Exception e)
         {
-            fail("Should have succeeded");
+            _exception = e;
         }
+
+        monitor(0, 10);
         
-        assertTrue("Listener should not have received exception", _caught.getCount() == 1);
-        
+        check(OPEN);
+    }
+
+    public void testConsumerCommitClose() throws Exception
+    {
+        send(1, 0);
+
+        _psession.commit();
+
+        expect(1, 0);
+
+        _csession.commit();
+
+        sleep(3.0f);
+
+        _csession.close();
+
+        assertEquals("Listener should not have received exception", 0, getNumberOfDeliveredExceptions());
+
         monitor(0, 0);
     }
     
+    public void testConsumerIdleReceiveCommit() throws Exception
+    {
+        send(1, 0);
+
+        _psession.commit();
+
+        sleep(2.0f);
+
+        expect(1, 0);
+
+        sleep(2.0f);
+
+        _csession.commit();
+
+        assertEquals("Listener should not have received exception", 0, getNumberOfDeliveredExceptions());
+
+        monitor(0, 0);
+    }
+
     public void testConsumerIdleCommit() throws Exception
     {
-        try
-        {
-            send(1, 0);
-    
-            _psession.commit();
-    
-            expect(1, 0);
-            
-            sleep(2.0f);
-    
-            _csession.commit();
-        }
-        catch (Exception e)
-        {
-            fail("Should have succeeded");
-        }
-        
-        assertTrue("Listener should not have received exception", _caught.getCount() == 1);
-        
+        send(1, 0);
+
+        _psession.commit();
+
+        expect(1, 0);
+
+        sleep(2.0f);
+
+        _csession.commit();
+
+        assertEquals("Listener should not have received exception", 0, getNumberOfDeliveredExceptions());
+
         monitor(0, 0);
     }
     
     public void testConsumerIdleRollback() throws Exception
     {
-        try
-        {
-            send(1, 0);
-    
-            _psession.commit();
-            
-            expect(1, 0);
-            
-            sleep(2.0f);
-    
-            _csession.rollback();
-        }
-        catch (Exception e)
-        {
-            fail("Should have succeeded");
-        }
-        
-        assertTrue("Listener should not have received exception", _caught.getCount() == 1);
-        
+        send(1, 0);
+
+        _psession.commit();
+
+        expect(1, 0);
+
+        sleep(2.0f);
+
+        _csession.rollback();
+
+        assertEquals("Listener should not have received exception", 0, getNumberOfDeliveredExceptions());
+
         monitor(0, 0);
     }
-    
+
     public void testConsumerOpenCommit() throws Exception
     {
-        try
-        {
-            send(1, 0);
-    
-            _psession.commit();
-            
-            sleep(3.0f);
-    
-            _csession.commit();
-        }
-        catch (Exception e)
-        {
-            fail("Should have succeeded");
-        }
-        
-        assertTrue("Listener should not have received exception", _caught.getCount() == 1);
-        
+        send(1, 0);
+
+        _psession.commit();
+
+        sleep(3.0f);
+
+        _csession.commit();
+
+        assertEquals("Listener should not have received exception", 0, getNumberOfDeliveredExceptions());
+
         monitor(0, 0);
     }
     
     public void testConsumerOpenRollback() throws Exception
     {
-        try
-        {
-            send(1, 0);
-    
-            _psession.commit();
-    
-            sleep(3.0f);
-    
-            _csession.rollback();
-        }
-        catch (Exception e)
-        {
-            fail("Should have succeeded");
-        }
-        
-        assertTrue("Listener should not have received exception", _caught.getCount() == 1);
+        send(1, 0);
         
+        _psession.commit();
+
+        sleep(3.0f);
+
+        _csession.rollback();
+
+        assertEquals("Listener should not have received exception", 0, getNumberOfDeliveredExceptions());
+
         monitor(0, 0);
     }
 }
diff --git a/qpid/java/systests/src/main/java/org/apache/qpid/test/unit/transacted/TransactionTimeoutTestCase.java b/qpid/java/systests/src/main/java/org/apache/qpid/test/unit/transacted/TransactionTimeoutTestCase.java
index ef2de5c592..2b90d38049 100644
--- a/qpid/java/systests/src/main/java/org/apache/qpid/test/unit/transacted/TransactionTimeoutTestCase.java
+++ b/qpid/java/systests/src/main/java/org/apache/qpid/test/unit/transacted/TransactionTimeoutTestCase.java
@@ -23,6 +23,7 @@ package org.apache.qpid.test.unit.transacted;
 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.ExceptionListener;
@@ -49,7 +50,7 @@ import org.apache.qpid.util.LogMonitor;
 /**
  * The {@link TestCase} for transaction timeout testing.
  */
-public class TransactionTimeoutTestCase extends QpidBrokerTestCase implements ExceptionListener
+public abstract class TransactionTimeoutTestCase extends QpidBrokerTestCase implements ExceptionListener
 {
     public static final String VIRTUALHOST = "test";
     public static final String TEXT = "0123456789abcdefghiforgettherest";
@@ -64,31 +65,16 @@ public class TransactionTimeoutTestCase extends QpidBrokerTestCase implements Ex
     protected Queue _queue;
     protected MessageConsumer _consumer;
     protected MessageProducer _producer;
-    protected CountDownLatch _caught = new CountDownLatch(1);
+    private CountDownLatch _exceptionLatch = new CountDownLatch(1);
+    protected AtomicInteger _exceptionCount = new AtomicInteger(0);
     protected String _message;
     protected Exception _exception;
     protected AMQConstant _code;
-    
-    protected void configure() throws Exception
-    {
-        // Setup housekeeping every second
-        setConfigurationProperty("virtualhosts.virtualhost." + VIRTUALHOST + ".housekeeping.checkPeriod", "100");
-        
-        /*
-         * Set transaction timout properties. The XML in the virtualhosts configuration is as follows:
-         * 
-         *  <transactionTimeout>
-         *      <openWarn>1000</openWarn>
-         *      <openClose>2000</openClose>
-         *      <idleWarn>500</idleWarn>
-         *      <idleClose>1500</idleClose>
-         *  </transactionTimeout>
-         */
-        setConfigurationProperty("virtualhosts.virtualhost." + VIRTUALHOST + ".transactionTimeout.openWarn", "1000");
-        setConfigurationProperty("virtualhosts.virtualhost." + VIRTUALHOST + ".transactionTimeout.openClose", "2000");
-        setConfigurationProperty("virtualhosts.virtualhost." + VIRTUALHOST + ".transactionTimeout.idleWarn", "500");
-        setConfigurationProperty("virtualhosts.virtualhost." + VIRTUALHOST + ".transactionTimeout.idleClose", "1000");
-    }
+
+    /**
+     * Subclasses must implement this to configure transaction timeout parameters.
+     */
+    protected abstract void configure() throws Exception;
         
     protected void setUp() throws Exception
     {
@@ -233,7 +219,7 @@ public class TransactionTimeoutTestCase extends QpidBrokerTestCase implements Ex
      */
     protected void check(String reason)throws InterruptedException
     {
-        assertTrue("Should have caught exception in listener", _caught.await(1, TimeUnit.SECONDS));
+        assertTrue("Should have caught exception in listener", _exceptionLatch.await(1, TimeUnit.SECONDS));
         assertNotNull("Should have thrown exception to client", _exception);
         assertTrue("Exception message should contain '" + reason + "': " + _message, _message.contains(reason + " transaction timed out"));
         assertNotNull("Exception should have an error code", _code);
@@ -243,11 +229,18 @@ public class TransactionTimeoutTestCase extends QpidBrokerTestCase implements Ex
     /** @see javax.jms.ExceptionListener#onException(javax.jms.JMSException) */
     public void onException(JMSException jmse)
     {
-        _caught.countDown();
+        _exceptionLatch.countDown();
+        _exceptionCount.incrementAndGet();
+
         _message = jmse.getLinkedException().getMessage();
         if (jmse.getLinkedException() instanceof AMQException)
         {
             _code = ((AMQException) jmse.getLinkedException()).getErrorCode();
         }
     }
+
+    protected int getNumberOfDeliveredExceptions()
+    {
+        return _exceptionCount.get();
+    }
 }
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 40df024c67..bb44aea659 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
@@ -52,7 +52,6 @@ import javax.naming.NamingException;
 import org.apache.commons.configuration.ConfigurationException;
 import org.apache.commons.configuration.XMLConfiguration;
 import org.apache.commons.lang.StringUtils;
-import org.apache.log4j.Level;
 import org.apache.log4j.Logger;
 import org.apache.qpid.AMQException;
 import org.apache.qpid.client.AMQConnectionFactory;
@@ -115,7 +114,6 @@ public class QpidBrokerTestCase extends QpidTestCase
     private static final String BROKER_LANGUAGE = "broker.language";
     private static final String BROKER_TYPE = "broker.type";
     private static final String BROKER_COMMAND = "broker.command";
-    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";
@@ -143,10 +141,9 @@ public class QpidBrokerTestCase extends QpidTestCase
     protected String _brokerLanguage = System.getProperty(BROKER_LANGUAGE, JAVA);
     protected BrokerType _brokerType = BrokerType.valueOf(System.getProperty(BROKER_TYPE, "").toUpperCase());
     protected String _brokerCommand = System.getProperty(BROKER_COMMAND);
-    private String _brokerClean = System.getProperty(BROKER_CLEAN, null);
     private Boolean _brokerCleanBetweenTests = Boolean.getBoolean(BROKER_CLEAN_BETWEEN_TESTS);
     private final AmqpProtocolVersion _brokerVersion = AmqpProtocolVersion.valueOf(System.getProperty(BROKER_VERSION, ""));
-    protected String _output = System.getProperty(TEST_OUTPUT);
+    protected String _output = System.getProperty(TEST_OUTPUT, System.getProperty("java.io.tmpdir"));
     protected Boolean _brokerPersistent = Boolean.getBoolean(BROKER_PERSITENT);
     private String _brokerProtocolExcludes = System.getProperty(BROKER_PROTOCOL_EXCLUDES);
 
@@ -217,8 +214,13 @@ public class QpidBrokerTestCase extends QpidTestCase
         if (redirected)
         {
             _outputFile = new File(String.format("%s/TEST-%s.out", _output, qname));
-            out = new PrintStream(_outputFile);
+            out = new PrintStream(new FileOutputStream(_outputFile), true);
             err = new PrintStream(String.format("%s/TEST-%s.err", _output, qname));
+
+            // This is relying on behaviour specific to log4j 1.2.12.   If we were to upgrade to 1.2.13 or
+            // beyond we must change either code (or config) to ensure that ConsoleAppender#setFollow
+            // is set to true otherwise log4j logging will not respect the following reassignment.
+
             System.setOut(out);
             System.setErr(err);
 
@@ -259,14 +261,9 @@ public class QpidBrokerTestCase extends QpidTestCase
 
             if(_brokerCleanBetweenTests)
             {
-            	try
-            	{
-            		cleanBroker();
-            	}
-            	catch (Exception e)
-            	{
-            		_logger.error("exception cleaning up broker", e);
-            	}
+                final String qpidWork = System.getProperty("QPID_WORK");
+                cleanBrokerWork(qpidWork);
+                createBrokerWork(qpidWork);
             }
 
             _logger.info("==========  stop " + getTestName() + " ==========");
@@ -298,11 +295,11 @@ public class QpidBrokerTestCase extends QpidTestCase
         String existingQpidWorkPath = System.getProperty(BROKER_EXISTING_QPID_WORK);
         if(existingQpidWorkPath != null && !existingQpidWorkPath.equals(""))
         {
-            cleanBroker();
 
+            String qpidWork = getQpidWork(_brokerType, getPort());
             File existing = new File(existingQpidWorkPath);
-            File qpidWork = new File(getQpidWork(_brokerType, getPort()));
-            FileUtils.copyRecursive(existing, qpidWork);
+            cleanBrokerWork(qpidWork);
+            FileUtils.copyRecursive(existing, new File(qpidWork));
         }
 
         startBroker();
@@ -494,25 +491,22 @@ public class QpidBrokerTestCase extends QpidTestCase
         }
         else if (!_brokerType.equals(BrokerType.EXTERNAL))
         {
+            // Add the port to QPID_WORK to ensure unique working dirs for multi broker tests
+            final String qpidWork = getQpidWork(_brokerType, port);
             String cmd = getBrokerCommand(port);
             _logger.info("starting external broker: " + cmd);
             ProcessBuilder pb = new ProcessBuilder(cmd.split("\\s+"));
             pb.redirectErrorStream(true);
-
             Map<String, String> env = pb.environment();
-
             String qpidHome = System.getProperty(QPID_HOME);
             env.put(QPID_HOME, qpidHome);
-
             //Augment Path with bin directory in QPID_HOME.
             env.put("PATH", env.get("PATH").concat(File.pathSeparator + qpidHome + "/bin"));
 
             //Add the test name to the broker run.
             // DON'T change PNAME, qpid.stop needs this value.
             env.put("QPID_PNAME", "-DPNAME=QPBRKR -DTNAME=\"" + getTestName() + "\"");
-            // Add the port to QPID_WORK to ensure unique working dirs for multi broker tests
-            env.put("QPID_WORK", getQpidWork(_brokerType, port));
-
+            env.put("QPID_WORK", qpidWork);
 
             // Use the environment variable to set amqj.logging.level for the broker
             // The value used is a 'server' value in the test configuration to
@@ -563,6 +557,10 @@ public class QpidBrokerTestCase extends QpidTestCase
                     env.put("QPID_OPTS", QPID_OPTS);
                 }
             }
+
+            // cpp broker requires that the work directory is created
+            createBrokerWork(qpidWork);
+
             Process process = pb.start();;
 
             Piper p = new Piper(process.getInputStream(),
@@ -577,7 +575,7 @@ public class QpidBrokerTestCase extends QpidTestCase
                 _logger.info("broker failed to become ready (" + p.ready + "):" + p.getStopLine());
                 //Ensure broker has stopped
                 process.destroy();
-                cleanBroker();
+                cleanBrokerWork(qpidWork);
                 throw new RuntimeException("broker failed to become ready:"
                                            + p.getStopLine());
             }
@@ -587,7 +585,7 @@ public class QpidBrokerTestCase extends QpidTestCase
                 //test that the broker is still running and hasn't exited unexpectedly
                 int exit = process.exitValue();
                 _logger.info("broker aborted: " + exit);
-                cleanBroker();
+                cleanBrokerWork(qpidWork);
                 throw new RuntimeException("broker aborted: " + exit);
             }
             catch (IllegalThreadStateException e)
@@ -655,21 +653,28 @@ public class QpidBrokerTestCase extends QpidTestCase
 
     public String getTestConfigFile()
     {
-        String path = _output == null ? System.getProperty("java.io.tmpdir") : _output;
-        return path + "/" + getTestQueueName() + "-config.xml";
+        return _output + "/" + getTestQueueName() + "-config.xml";
     }
 
     public String getTestVirtualhostsFile()
     {
-        String path = _output == null ? System.getProperty("java.io.tmpdir") : _output;
-        return path + "/" + getTestQueueName() + "-virtualhosts.xml";
+        return _output + "/" + getTestQueueName() + "-virtualhosts.xml";
+    }
+
+    private String relativeToQpidHome(String file)
+    {
+        return file.replace(System.getProperty(QPID_HOME,"QPID_HOME") + "/","");
     }
 
     protected void saveTestConfiguration() throws ConfigurationException
     {
-        // Specifiy the test config file
+        // Specify the test config file
         String testConfig = getTestConfigFile();
-        setSystemProperty("test.config", testConfig);
+        String relative = relativeToQpidHome(testConfig);
+
+        setSystemProperty("test.config", relative);
+        _logger.info("Set test.config property to: " + relative);
+        _logger.info("Saving test virtualhosts file at: " + testConfig);
 
         // Create the file if configuration does not exist
         if (_testConfiguration.isEmpty())
@@ -681,9 +686,13 @@ public class QpidBrokerTestCase extends QpidTestCase
 
     protected void saveTestVirtualhosts() throws ConfigurationException
     {
-        // Specifiy the test virtualhosts file
+        // Specify the test virtualhosts file
         String testVirtualhosts = getTestVirtualhostsFile();
-        setSystemProperty("test.virtualhosts", testVirtualhosts);
+        String relative = relativeToQpidHome(testVirtualhosts);
+
+        setSystemProperty("test.virtualhosts", relative);
+        _logger.info("Set test.virtualhosts property to: " + relative);
+        _logger.info("Saving test virtualhosts file at: " + testVirtualhosts);
 
         // Create the file if configuration does not exist
         if (_testVirtualhosts.isEmpty())
@@ -693,30 +702,33 @@ public class QpidBrokerTestCase extends QpidTestCase
         _testVirtualhosts.save(testVirtualhosts);
     }
 
-    public void cleanBroker()
+    protected void cleanBrokerWork(final String qpidWork)
     {
-        if (_brokerClean != null)
+        if (qpidWork != null)
         {
-            _logger.info("clean: " + _brokerClean);
+            _logger.info("Cleaning broker work dir: " + qpidWork);
 
-            try
+            File file = new File(qpidWork);
+            if (file.exists())
             {
-                ProcessBuilder pb = new ProcessBuilder(_brokerClean.split("\\s+"));
-                pb.redirectErrorStream(true);
-                Process clean = pb.start();
-                new Piper(clean.getInputStream(),_brokerOutputStream).start();
-
-                clean.waitFor();
-
-                _logger.info("clean exited: " + clean.exitValue());
-            }
-            catch (IOException e)
-            {
-                throw new RuntimeException(e);
+                final boolean success = FileUtils.delete(file, true);
+                if(!success)
+                {
+                    throw new RuntimeException("Failed to recursively delete beneath : " + file);
+                }
             }
-            catch (InterruptedException e)
+        }
+    }
+
+    protected void createBrokerWork(final String qpidWork)
+    {
+        if (qpidWork != null)
+        {
+            final File dir = new File(qpidWork);
+            dir.mkdirs();
+            if (!dir.isDirectory())
             {
-                throw new RuntimeException(e);
+                throw new RuntimeException("Failed to created Qpid work directory : " + qpidWork);
             }
         }
     }
@@ -730,7 +742,7 @@ public class QpidBrokerTestCase extends QpidTestCase
     {
         port = getPort(port);
 
-        _logger.info("stopping broker: " + getBrokerCommand(port));
+        _logger.info("stopping broker on port : " + port);
         BrokerHolder broker = _brokers.remove(port);
         broker.shutdown();
     }
@@ -906,7 +918,7 @@ public class QpidBrokerTestCase extends QpidTestCase
     }
 
     /**
-     * Add an environtmen variable for the external broker environment
+     * Add an environment variable for the external broker environment
      *
      * @param property the property to set
      * @param value    the value to set it to
@@ -990,9 +1002,9 @@ public class QpidBrokerTestCase extends QpidTestCase
      * Get the default connection factory for the currently used broker
      * Default factory is "local"
      *
-     * @return A conection factory
+     * @return A connection factory
      *
-     * @throws Exception if there is an error getting the tactory
+     * @throws Exception if there is an error getting the factory
      */
     public AMQConnectionFactory getConnectionFactory() throws NamingException
     {
@@ -1016,7 +1028,7 @@ public class QpidBrokerTestCase extends QpidTestCase
      *
      * @param factoryName The factory name
      *
-     * @return A conection factory
+     * @return A connection factory
      *
      * @throws Exception if there is an error getting the tactory
      */
@@ -1054,7 +1066,7 @@ public class QpidBrokerTestCase extends QpidTestCase
     {
         _logger.info("get connection");
         Connection con = getConnectionFactory().createConnection(username, password);
-        //add the connection in the lis of connections
+        //add the connection in the list of connections
         _connections.add(con);
         return con;
     }
@@ -1063,7 +1075,7 @@ public class QpidBrokerTestCase extends QpidTestCase
     {
         _logger.info("get Connection");
         Connection con = getConnectionFactory().createConnection(username, password, id);
-        //add the connection in the lis of connections
+        //add the connection in the list of connections
         _connections.add(con);
         return con;
     }
@@ -1154,7 +1166,7 @@ public class QpidBrokerTestCase extends QpidTestCase
     /**
      * Send messages to the given destination.
      *
-     * If session is transacted then messages will be commited before returning
+     * If session is transacted then messages will be committed before returning
      *
      * @param session the session to use for sending
      * @param destination where to send them to
@@ -1162,7 +1174,7 @@ public class QpidBrokerTestCase extends QpidTestCase
      *
      * @param batchSize the batchSize in which to commit, 0 means no batching,
      * but a single commit at the end
-     * @return the sent messgse
+     * @return the sent message
      *
      * @throws Exception
      */
@@ -1175,7 +1187,7 @@ public class QpidBrokerTestCase extends QpidTestCase
     /**
      * Send messages to the given destination.
      *
-     * If session is transacted then messages will be commited before returning
+     * If session is transacted then messages will be committed before returning
      *
      * @param session the session to use for sending
      * @param destination where to send them to
@@ -1184,7 +1196,7 @@ public class QpidBrokerTestCase extends QpidTestCase
      * @param offset offset allows the INDEX value of the message to be adjusted.
      * @param batchSize the batchSize in which to commit, 0 means no batching,
      * but a single commit at the end
-     * @return the sent messgse
+     * @return the sent message
      *
      * @throws Exception
      */
diff --git a/qpid/java/systests/src/main/java/org/apache/qpid/util/LogMonitor.java b/qpid/java/systests/src/main/java/org/apache/qpid/util/LogMonitor.java
index a5e2b80f64..c09e63308c 100644
--- a/qpid/java/systests/src/main/java/org/apache/qpid/util/LogMonitor.java
+++ b/qpid/java/systests/src/main/java/org/apache/qpid/util/LogMonitor.java
@@ -27,11 +27,9 @@ import org.apache.log4j.SimpleLayout;
 import java.io.BufferedReader;
 import java.io.File;
 import java.io.FileNotFoundException;
-import java.io.FileOutputStream;
 import java.io.FileReader;
 import java.io.IOException;
-import java.io.OutputStreamWriter;
-import java.util.ArrayList;
+import java.io.LineNumberReader;
 import java.util.List;
 import java.util.LinkedList;
 
@@ -45,10 +43,12 @@ import java.util.LinkedList;
 public class LogMonitor
 {
     // The file that the log statements will be written to.
-    private File _logfile;
+    private final File _logfile;
 
     // The appender we added to the get messages
-    private FileAppender _appender;
+    private final FileAppender _appender;
+
+    private int _linesToSkip = 0;
 
     /**
      * Create a new LogMonitor that creates a new Log4j Appender and monitors
@@ -78,6 +78,7 @@ public class LogMonitor
         if (file != null && file.exists())
         {
             _logfile = file;
+            _appender = null;
         }
         else
         {
@@ -99,13 +100,13 @@ public class LogMonitor
      * @param wait    the time in ms to wait for the message to occur
      * @return true if the message was found
      *
-     * @throws java.io.FileNotFoundException if the Log file can nolonger be found
+     * @throws java.io.FileNotFoundException if the Log file can no longer be found
      * @throws IOException                   thrown when reading the log file
      */
     public List<String> waitAndFindMatches(String message, long wait)
             throws FileNotFoundException, IOException
     {
-        if (waitForMessage(message, wait, true))
+        if (waitForMessage(message, wait))
         {
             return findMatches(message);
         }
@@ -116,7 +117,9 @@ public class LogMonitor
     }
 
     /**
-     * Checks the log for instances of the search string.
+     * Checks the log for instances of the search string. If the caller
+     * has previously called {@link #markDiscardPoint()}, lines up until the discard
+     * point are not considered.
      *
      * The pattern parameter can take any valid argument used in String.contains()
      *
@@ -130,66 +133,99 @@ public class LogMonitor
      */
     public List<String> findMatches(String pattern) throws IOException
     {
-        return FileUtils.searchFile(_logfile, pattern);
+
+        List<String> results = new LinkedList<String>();
+
+        LineNumberReader reader = new LineNumberReader(new FileReader(_logfile));
+        try
+        {
+            while (reader.ready())
+            {
+                String line = reader.readLine();
+                if (reader.getLineNumber()  > _linesToSkip && line.contains(pattern))
+                {
+                    results.add(line);
+                }
+            }
+        }
+        finally
+        {
+            reader.close();
+        }
+
+        return results;
     }
 
     /**
-     * Checks the log file for a given message to appear.
+     * Checks the log file for a given message to appear.  If the caller
+     * has previously called {@link #markDiscardPoint()}, lines up until the discard
+     * point are not considered.
      *
      * @param message the message to wait for in the log
      * @param wait    the time in ms to wait for the message to occur
-     *
-     * @param printFileOnFailure should we print the contents that have been
-     * read if we fail ot find the message.
      * @return true if the message was found
      *
-     * @throws java.io.FileNotFoundException if the Log file can nolonger be found
+     * @throws java.io.FileNotFoundException if the Log file can no longer be found
      * @throws IOException                   thrown when reading the log file
      */
-    public boolean waitForMessage(String message, long wait, boolean printFileOnFailure)
+    public boolean waitForMessage(String message, long wait)
             throws FileNotFoundException, IOException
     {
         // Loop through alerts until we're done or wait ms seconds have passed,
         // just in case the logfile takes a while to flush.
-        BufferedReader reader = new BufferedReader(new FileReader(_logfile));
-        boolean found = false;
-        long endtime = System.currentTimeMillis() + wait;
-        ArrayList<String> contents = new ArrayList<String>();
-        while (!found && System.currentTimeMillis() < endtime)
+        LineNumberReader reader = null;
+        try
         {
-            while (reader.ready())
+            reader = new LineNumberReader(new FileReader(_logfile));
+
+            boolean found = false;
+            long endtime = System.currentTimeMillis() + wait;
+            while (!found && System.currentTimeMillis() < endtime)
             {
-                String line = reader.readLine();
-                contents.add(line);
-                if (line.contains(message))
+                boolean ready = true;
+                while (ready = reader.ready())
                 {
-                    found = true;
+                    String line = reader.readLine();
+
+                    if (reader.getLineNumber() > _linesToSkip)
+                    {
+                        if (line.contains(message))
+                        {
+                            found = true;
+                            break;
+                        }
+                    }
+                }
+                if (!ready)
+                {
+                    try
+                    {
+                        Thread.sleep(50);
+                    }
+                    catch (InterruptedException ie)
+                    {
+                        Thread.currentThread().interrupt();
+                    }
                 }
             }
+            return found;
+
         }
-        if (!found && printFileOnFailure)
+        finally
         {
-            for (String line : contents)
+            if (reader != null)
             {
-                System.out.println(line);
+                reader.close();
             }
         }
-        return found;
     }
     
-
-    public boolean waitForMessage(String message, long alertLogWaitPeriod) throws FileNotFoundException, IOException
-    {
-       return waitForMessage(message, alertLogWaitPeriod, true);
-    }
-
-
     /**
      * Read the log file in to memory as a String
      *
      * @return the current contents of the log file
      *
-     * @throws java.io.FileNotFoundException if the Log file can nolonger be found
+     * @throws java.io.FileNotFoundException if the Log file can no longer be found
      * @throws IOException                   thrown when reading the log file
      */
     public String readFile() throws FileNotFoundException, IOException
@@ -208,14 +244,37 @@ public class LogMonitor
     }
 
     /**
-     * Clears the log file and writes: 'Log Monitor Reset' at the start of the file
+     * Marks the discard point in the log file.
      *
-     * @throws java.io.FileNotFoundException if the Log file can nolonger be found
+     * @throws java.io.FileNotFoundException if the Log file can no longer be found
      * @throws IOException                   thrown if there is a problem with the log file
      */
-    public void reset() throws FileNotFoundException, IOException
+    public void markDiscardPoint() throws FileNotFoundException, IOException
+    {
+        _linesToSkip = countLinesInFile();
+    }
+
+    private int countLinesInFile() throws IOException
     {
-        new FileOutputStream(_logfile).getChannel().truncate(0);
+        int lineCount = 0;
+        BufferedReader br = null;
+        try
+        {
+            br = new BufferedReader(new FileReader(_logfile));
+            while(br.readLine() != null)
+            {
+                lineCount++;
+            }
+
+            return lineCount;
+        }
+        finally
+        {
+            if (br != null)
+            {
+                br.close();
+            }
+        }
     }
 
     /**
diff --git a/qpid/java/systests/src/main/java/org/apache/qpid/util/LogMonitorTest.java b/qpid/java/systests/src/main/java/org/apache/qpid/util/LogMonitorTest.java
index a99abe4b94..89f707fbef 100644
--- a/qpid/java/systests/src/main/java/org/apache/qpid/util/LogMonitorTest.java
+++ b/qpid/java/systests/src/main/java/org/apache/qpid/util/LogMonitorTest.java
@@ -155,7 +155,7 @@ public class LogMonitorTest extends TestCase
 
         String notLogged = "This text was not logged";
 
-        validateLogDoesNotContainsMessage(_monitor, notLogged);
+        validateLogDoesNotContainMessage(_monitor, notLogged);
     }
 
     public void testWaitForMessage_Timeout() throws IOException
@@ -168,28 +168,27 @@ public class LogMonitorTest extends TestCase
 
         // Verify that we can time out waiting for a message
         assertFalse("Message was logged ",
-                    _monitor.waitForMessage(message, TIME_OUT / 2, false));
+                    _monitor.waitForMessage(message, TIME_OUT / 2));
 
         // Verify that the message did eventually get logged.
         assertTrue("Message was never logged.",
                     _monitor.waitForMessage(message, TIME_OUT));
     }
 
-    public void testReset() throws IOException
+    public void testDiscardPoint() throws IOException
     {
-        String message = getName() + ": Test Message";
-
-        Logger.getRootLogger().warn(message);
-
-        validateLogContainsMessage(_monitor, message);
+        String firstMessage = getName() + ": Test Message1";
+        Logger.getRootLogger().warn(firstMessage);
 
-        String LOG_RESET_TEXT = "Log Monitor Reset";
+        validateLogContainsMessage(_monitor, firstMessage);
 
-        validateLogDoesNotContainsMessage(_monitor, LOG_RESET_TEXT);
+        _monitor.markDiscardPoint();
 
-        _monitor.reset();
+        validateLogDoesNotContainMessage(_monitor, firstMessage);
 
-        assertEquals("", _monitor.readFile());
+        String secondMessage = getName() + ": Test Message2";
+        Logger.getRootLogger().warn(secondMessage);
+        validateLogContainsMessage(_monitor, secondMessage);
     }
 
     public void testRead() throws IOException
@@ -214,7 +213,7 @@ public class LogMonitorTest extends TestCase
      *
      * @throws IOException if a problems occurs
      */
-    protected void validateLogDoesNotContainsMessage(LogMonitor log, String message)
+    protected void validateLogDoesNotContainMessage(LogMonitor log, String message)
             throws IOException
     {
         List<String> results = log.findMatches(message);