QPID-3936: initial checkin of new testing framework, initially to be used for performance testing but later to be expanded for use with other testing scenarios.

Applied patch from Philip Harvey <phil@philharveyonline.com>, Oleksandr Rudyy <orudyy@gmail.com>, Andrew MacBean <andymacbean@gmail.com>, Keith Wall <kwall@apache.org>, and myself.

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

63 files changed, 6845 insertions, 0 deletions

diff --git a/qpid/java/perftests/src/test/java/org/apache/qpid/disttest/ArgumentParserTest.java b/qpid/java/perftests/src/test/java/org/apache/qpid/disttest/ArgumentParserTest.java
new file mode 100644
index 0000000000..3be82627fe
--- /dev/null
+++ b/qpid/java/perftests/src/test/java/org/apache/qpid/disttest/ArgumentParserTest.java
@@ -0,0 +1,96 @@
+/*
+ * 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.disttest;
+
+import java.util.HashMap;
+import java.util.Map;
+
+import org.apache.qpid.test.utils.QpidTestCase;
+
+public class ArgumentParserTest extends QpidTestCase
+{
+    private static final String TEST_CONFIG_FILENAME = "ControllerRunnerTest-test-config-filename.json";
+    private static final String JNDI_CONFIG_FILENAME = "ControllerRunnerTest-jndi-config-filename.properties";
+    private static final String DISTRIBUTED_MODE = "true";
+
+    public static final String TEST_CONFIG_PROP = "test-config";
+    public static final String JNDI_CONFIG_PROP = "jndi-config";
+    public static final String DISTRIBUTED_PROP = "distributed";
+
+    public static final String TEST_CONFIG_DEFAULT = "perftests-config.json";
+    public static final String JNDI_CONFIG_DEFAULT = "perftests-jndi.properties";
+    public static final String DISTRIBUTED_DEFAULT = "false";
+
+    private Map<String,String> _options = new HashMap<String, String>();
+
+    private ArgumentParser _parser;
+
+    @Override
+    protected void setUp() throws Exception
+    {
+        super.setUp();
+        _parser = new ArgumentParser();
+
+        _options.clear();
+        _options.put(TEST_CONFIG_PROP, TEST_CONFIG_DEFAULT);
+        _options.put(JNDI_CONFIG_PROP, JNDI_CONFIG_DEFAULT);
+        _options.put(DISTRIBUTED_PROP, DISTRIBUTED_DEFAULT);
+    }
+
+    public void testInvalidArguments()
+    {
+        String[] args = new String[]{"nonExistentConfigProperty" + "=" + TEST_CONFIG_FILENAME};
+
+        try
+        {
+            _parser.parseArgumentsIntoConfig(_options, args);
+            fail("expected exception to be thrown due to provision of a non existent config property");
+        }
+        catch(IllegalArgumentException e)
+        {
+            //expected
+        }
+    }
+
+    public void testDefaultConfigValues()
+    {
+        String[] args = new String[0];
+
+        _parser.parseArgumentsIntoConfig(_options, args);
+
+        assertEquals("unexpected config value", TEST_CONFIG_DEFAULT, _options.get(TEST_CONFIG_PROP));
+        assertEquals("unexpected config value", JNDI_CONFIG_DEFAULT, _options.get(JNDI_CONFIG_PROP));
+        assertEquals("unexpected config value", DISTRIBUTED_DEFAULT, _options.get(DISTRIBUTED_PROP));
+    }
+
+    public void testConfigurationParsingOverridesDefault() throws Exception
+    {
+        String[] args = new String[]{TEST_CONFIG_PROP + "=" + TEST_CONFIG_FILENAME,
+                JNDI_CONFIG_PROP + "=" + JNDI_CONFIG_FILENAME,
+                DISTRIBUTED_PROP + "=" + DISTRIBUTED_MODE};
+
+        _parser.parseArgumentsIntoConfig(_options, args);
+
+        assertEquals("unexpected config value", TEST_CONFIG_FILENAME, _options.get(TEST_CONFIG_PROP));
+        assertEquals("unexpected config value", JNDI_CONFIG_FILENAME, _options.get(JNDI_CONFIG_PROP));
+        assertEquals("unexpected config value", DISTRIBUTED_MODE, _options.get(DISTRIBUTED_PROP));
+        assertFalse("override value was the same as the default", DISTRIBUTED_MODE.equalsIgnoreCase(_options.get(DISTRIBUTED_DEFAULT)));
+    }
+}
diff --git a/qpid/java/perftests/src/test/java/org/apache/qpid/disttest/ConfigFileHelper.java b/qpid/java/perftests/src/test/java/org/apache/qpid/disttest/ConfigFileHelper.java
new file mode 100644
index 0000000000..12ba3b56ad
--- /dev/null
+++ b/qpid/java/perftests/src/test/java/org/apache/qpid/disttest/ConfigFileHelper.java
@@ -0,0 +1,48 @@
+/*
+ * 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.disttest;
+
+import java.io.InputStream;
+import java.io.InputStreamReader;
+import java.io.Reader;
+
+import org.apache.qpid.disttest.controller.config.Config;
+import org.apache.qpid.disttest.controller.config.ConfigReader;
+
+public class ConfigFileHelper
+{
+    public static Reader getConfigFileReader(Class<?> testClass, String resourceName)
+    {
+        InputStream inputStream = testClass.getResourceAsStream(resourceName);
+        if(inputStream == null)
+        {
+            throw new RuntimeException("Can't find resource " + resourceName + " using classloader of class " + testClass);
+        }
+        Reader reader = new InputStreamReader(inputStream);
+        return reader;
+    }
+
+    public static Config getConfigFromResource(Class<?> testClass, String resourceName)
+    {
+        ConfigReader configReader = new ConfigReader();
+        Config config = configReader.readConfig(getConfigFileReader(testClass, resourceName));
+        return config;
+    }
+}
diff --git a/qpid/java/perftests/src/test/java/org/apache/qpid/disttest/VisitorTest.java b/qpid/java/perftests/src/test/java/org/apache/qpid/disttest/VisitorTest.java
new file mode 100644
index 0000000000..320e7d8c9d
--- /dev/null
+++ b/qpid/java/perftests/src/test/java/org/apache/qpid/disttest/VisitorTest.java
@@ -0,0 +1,92 @@
+/*
+ * 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.disttest;
+
+import junit.framework.TestCase;
+
+import org.apache.qpid.disttest.message.Command;
+
+
+public class VisitorTest extends TestCase
+{
+
+    public void testStringVisited() throws Exception
+    {
+        Object argument = new String();
+
+        TestVisitor visitor = new TestVisitor();
+        visitor.visit(argument);
+
+        assertSame(argument, visitor._string);
+    }
+
+    public void testCommandVisited() throws Exception
+    {
+        Object argument = new TestCommand();
+
+        TestVisitor visitor = new TestVisitor();
+        visitor.visit(argument);
+
+        assertSame(argument, visitor._testCommand);
+    }
+
+    public void testNoVisitIntegerImplementatiom() throws Exception
+    {
+        Integer argument = Integer.valueOf(1);
+
+        TestVisitor visitor = new TestVisitor();
+
+        try
+        {
+            visitor.visit(argument);
+            fail("Exception not thrown");
+        }
+        catch (DistributedTestException e)
+        {
+            assertNotNull(e.getCause());
+            assertEquals(NoSuchMethodException.class, e.getCause().getClass());
+        }
+    }
+
+    class TestVisitor extends Visitor
+    {
+        String _string = null;
+        TestCommand _testCommand = null;
+
+        public void visit(String string)
+        {
+            _string = string;
+        }
+
+        public void visit(TestCommand command)
+        {
+            _testCommand = command;
+        }
+    }
+
+    class TestCommand extends Command
+    {
+
+        public TestCommand()
+        {
+            super(null);
+        }
+    }
+}
diff --git a/qpid/java/perftests/src/test/java/org/apache/qpid/disttest/client/ClientCommandVisitorTest.java b/qpid/java/perftests/src/test/java/org/apache/qpid/disttest/client/ClientCommandVisitorTest.java
new file mode 100644
index 0000000000..4a82f6719f
--- /dev/null
+++ b/qpid/java/perftests/src/test/java/org/apache/qpid/disttest/client/ClientCommandVisitorTest.java
@@ -0,0 +1,108 @@
+/*
+ * 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.disttest.client;
+
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.verify;
+import junit.framework.TestCase;
+
+import org.apache.qpid.disttest.jms.ClientJmsDelegate;
+import org.apache.qpid.disttest.message.CreateConnectionCommand;
+import org.apache.qpid.disttest.message.CreateConsumerCommand;
+import org.apache.qpid.disttest.message.CreateMessageProviderCommand;
+import org.apache.qpid.disttest.message.CreateProducerCommand;
+import org.apache.qpid.disttest.message.CreateSessionCommand;
+import org.apache.qpid.disttest.message.StartTestCommand;
+import org.apache.qpid.disttest.message.StopClientCommand;
+import org.apache.qpid.disttest.message.TearDownTestCommand;
+
+public class ClientCommandVisitorTest extends TestCase
+{
+    private Client _client;
+    private ClientCommandVisitor _visitor;
+    private ClientJmsDelegate _delegate;
+
+
+    @Override
+    protected void setUp() throws Exception
+    {
+        super.setUp();
+        _client = mock(Client.class);
+        _delegate = mock(ClientJmsDelegate.class);
+        _visitor = new ClientCommandVisitor(_client, _delegate);
+    }
+
+    public void testStopClient()
+    {
+        StopClientCommand command = new StopClientCommand();
+        _visitor.visit(command);
+        verify(_client).stop();
+    }
+
+    public void testCreateConnection() throws Exception
+    {
+        final CreateConnectionCommand command = new CreateConnectionCommand();
+        _visitor.visit(command);
+        verify(_delegate).createConnection(command);
+    }
+
+    public void testCreateSession() throws Exception
+    {
+        final CreateSessionCommand command = new CreateSessionCommand();
+        _visitor.visit(command);
+        verify(_delegate).createSession(command);
+    }
+
+    public void testCreateProducer() throws Exception
+    {
+        final CreateProducerCommand command = new CreateProducerCommand();
+        _visitor.visit(command);
+        verify(_delegate).createProducer(command);
+    }
+
+    public void testCreateConsumer() throws Exception
+    {
+        final CreateConsumerCommand command = new CreateConsumerCommand();
+        _visitor.visit(command);
+        verify(_delegate).createConsumer(command);
+    }
+
+    public void testStartTest() throws Exception
+    {
+        final StartTestCommand command = new StartTestCommand();
+        _visitor.visit(command);
+        verify(_client).startTest();
+    }
+
+    public void testStopTest() throws Exception
+    {
+        final TearDownTestCommand stopCommand = new TearDownTestCommand();
+        _visitor.visit(stopCommand);
+        verify(_client).tearDownTest();
+    }
+
+    public void testCreateMessageProvider() throws Exception
+    {
+        final CreateMessageProviderCommand command = new CreateMessageProviderCommand();
+        command.setProviderName("test");
+        _visitor.visit(command);
+        verify(_delegate).createMessageProvider(command);
+    }
+}
diff --git a/qpid/java/perftests/src/test/java/org/apache/qpid/disttest/client/ClientTest.java b/qpid/java/perftests/src/test/java/org/apache/qpid/disttest/client/ClientTest.java
new file mode 100644
index 0000000000..198baa6ef4
--- /dev/null
+++ b/qpid/java/perftests/src/test/java/org/apache/qpid/disttest/client/ClientTest.java
@@ -0,0 +1,159 @@
+/*
+ * 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.disttest.client;
+
+import static org.mockito.Matchers.isA;
+import static org.mockito.Mockito.inOrder;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+import java.util.Collections;
+import java.util.Timer;
+import java.util.TimerTask;
+
+import junit.framework.TestCase;
+
+import org.apache.qpid.disttest.jms.ClientJmsDelegate;
+import org.apache.qpid.disttest.message.Command;
+import org.apache.qpid.disttest.message.ParticipantResult;
+import org.apache.qpid.disttest.message.Response;
+import org.apache.qpid.disttest.message.StopClientCommand;
+import org.mockito.InOrder;
+import org.mockito.Mockito;
+
+public class ClientTest extends TestCase
+{
+    private Client _client;
+    private ClientJmsDelegate _delegate;
+    private ClientCommandVisitor _visitor;
+    private ParticipantExecutor _participant;
+    private ParticipantExecutorRegistry _participantRegistry;
+
+    @Override
+    protected void setUp() throws Exception
+    {
+        super.setUp();
+        _delegate = mock(ClientJmsDelegate.class);
+        _visitor = mock(ClientCommandVisitor.class);
+        _client = new Client(_delegate);
+        _client.setClientCommandVisitor(_visitor);
+        _participant = mock(ParticipantExecutor.class);
+        when(_participant.getParticipantName()).thenReturn("testParticipantMock");
+
+        _participantRegistry = mock(ParticipantExecutorRegistry.class);
+        when(_participantRegistry.executors()).thenReturn(Collections.singletonList(_participant));
+        _client.setParticipantRegistry(_participantRegistry);
+    }
+
+    public void testInitialState() throws Exception
+    {
+        assertEquals("Expected client to be in CREATED state", ClientState.CREATED, _client.getState());
+    }
+
+    public void testStart() throws Exception
+    {
+        _client.start();
+        final InOrder inOrder = inOrder(_delegate);
+        inOrder.verify(_delegate).setInstructionListener(_client);
+        inOrder.verify(_delegate).sendRegistrationMessage();
+        assertEquals("Expected client to be in STARTED state", ClientState.READY, _client.getState());
+    }
+
+    public void testStopClient() throws Exception
+    {
+        _client.stop();
+
+        assertEquals("Expected client to be in STOPPED state", ClientState.STOPPED, _client.getState());
+    }
+
+    public void testProcessInstructionVisitsCommandAndResponds() throws Exception
+    {
+        // has to be declared to be of supertype Command otherwise Mockito verify()
+        // refers to wrong method
+        final Command command = new StopClientCommand();
+        _client.processInstruction(command);
+
+        verify(_visitor).visit(command);
+        verify(_delegate).sendResponseMessage(isA(Response.class));
+    }
+
+    public void testWaitUntilStopped() throws Exception
+    {
+        stopClientLater(500);
+        _client.waitUntilStopped(1000);
+        verify(_delegate).destroy();
+    }
+
+    public void testStartTest() throws Exception
+    {
+        _client.start();
+        _client.addParticipantExecutor(_participant);
+
+        verify(_participantRegistry).add(_participant);
+
+        _client.startTest();
+
+        InOrder inOrder = Mockito.inOrder(_delegate, _participant);
+        inOrder.verify(_delegate).startConnections();
+        inOrder.verify(_participant).start(_client);
+    }
+
+    public void testTearDownTest() throws Exception
+    {
+        // before we can tear down the test the client needs to be in the "running test" state, which requires a participant
+        _client.start();
+        _client.addParticipantExecutor(_participant);
+        _client.startTest();
+
+        _client.tearDownTest();
+
+        verify(_delegate).closeTestConnections();
+
+        verify(_participantRegistry).clear();
+    }
+
+    public void testResults() throws Exception
+    {
+        ParticipantResult testResult = mock(ParticipantResult.class);
+        _client.sendResults(testResult);
+        verify(_delegate).sendResponseMessage(testResult);
+    }
+
+    private void stopClientLater(long delay)
+    {
+        doLater(new TimerTask()
+        {
+            @Override
+            public void run()
+            {
+                _client.stop();
+            }
+
+        }, delay);
+    }
+
+    private void doLater(TimerTask task, long delayInMillis)
+    {
+        Timer timer = new Timer();
+        timer.schedule(task, delayInMillis);
+    }
+
+}
diff --git a/qpid/java/perftests/src/test/java/org/apache/qpid/disttest/client/ConsumerParticipantTest.java b/qpid/java/perftests/src/test/java/org/apache/qpid/disttest/client/ConsumerParticipantTest.java
new file mode 100644
index 0000000000..805ebd3be4
--- /dev/null
+++ b/qpid/java/perftests/src/test/java/org/apache/qpid/disttest/client/ConsumerParticipantTest.java
@@ -0,0 +1,173 @@
+/*
+ * 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.disttest.client;
+
+import static org.apache.qpid.disttest.client.ParticipantTestHelper.assertExpectedResults;
+import static org.mockito.Matchers.anyLong;
+import static org.mockito.Matchers.anyString;
+import static org.mockito.Mockito.atLeastOnce;
+import static org.mockito.Mockito.inOrder;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+import javax.jms.Message;
+
+import junit.framework.TestCase;
+
+import org.apache.qpid.disttest.DistributedTestException;
+import org.apache.qpid.disttest.jms.ClientJmsDelegate;
+import org.apache.qpid.disttest.message.CreateConsumerCommand;
+import org.apache.qpid.disttest.message.ParticipantResult;
+import org.mockito.InOrder;
+
+public class ConsumerParticipantTest extends TestCase
+{
+    private static final String SESSION_NAME1 = "SESSION1";
+    private static final String PARTICIPANT_NAME1 = "PARTICIPANT_NAME1";
+    private static final long RECEIVE_TIMEOUT = 100;
+    private static final String CLIENT_NAME = "CLIENT_NAME";
+    private static final int PAYLOAD_SIZE_PER_MESSAGE = 1024;
+
+    private final Message _mockMessage = mock(Message.class);
+    private final CreateConsumerCommand _command = new CreateConsumerCommand();
+    private ClientJmsDelegate _delegate;
+    private ConsumerParticipant _consumerParticipant;
+    private InOrder _inOrder;
+
+    /** used to check start/end time of results */
+    private long _testStartTime;
+
+    @Override
+    protected void setUp() throws Exception
+    {
+        super.setUp();
+        _delegate = mock(ClientJmsDelegate.class);
+        _inOrder = inOrder(_delegate);
+
+        _command.setSessionName(SESSION_NAME1);
+        _command.setParticipantName(PARTICIPANT_NAME1);
+        _command.setSynchronous(true);
+        _command.setReceiveTimeout(RECEIVE_TIMEOUT);
+
+        _consumerParticipant = new ConsumerParticipant(_delegate, _command);
+
+        when(_delegate.consumeMessage(PARTICIPANT_NAME1, RECEIVE_TIMEOUT)).thenReturn(_mockMessage);
+        when(_delegate.calculatePayloadSizeFrom(_mockMessage)).thenReturn(PAYLOAD_SIZE_PER_MESSAGE);
+
+        _testStartTime = System.currentTimeMillis();
+    }
+
+    public void testNoMessagesToReceive() throws Exception
+    {
+        _command.setNumberOfMessages(0);
+        _command.setMaximumDuration(0);
+
+        try
+        {
+            _consumerParticipant.doIt(CLIENT_NAME);
+            fail("Exception not thrown");
+        }
+        catch(DistributedTestException e)
+        {
+            // PASS
+            assertEquals("number of messages and duration cannot both be zero", e.getMessage());
+
+        }
+
+        verify(_delegate, never()).consumeMessage(anyString(), anyLong());
+    }
+
+    public void testReceiveOneMessageSynch() throws Exception
+    {
+        int numberOfMessages = 1;
+        long totalPayloadSize = PAYLOAD_SIZE_PER_MESSAGE * numberOfMessages;
+        _command.setNumberOfMessages(numberOfMessages);
+
+        ParticipantResult result = _consumerParticipant.doIt(CLIENT_NAME);
+
+        assertExpectedResults(result, PARTICIPANT_NAME1, CLIENT_NAME, _testStartTime, numberOfMessages, PAYLOAD_SIZE_PER_MESSAGE, totalPayloadSize, null);
+
+        _inOrder.verify(_delegate).consumeMessage(PARTICIPANT_NAME1, RECEIVE_TIMEOUT);
+        _inOrder.verify(_delegate).calculatePayloadSizeFrom(_mockMessage);
+        _inOrder.verify(_delegate).commitOrAcknowledgeMessage(_mockMessage, SESSION_NAME1);
+    }
+
+    public void testReceiveMessagesForDurationSynch() throws Exception
+    {
+        long duration = 100;
+        _command.setMaximumDuration(duration);
+
+        ParticipantResult result = _consumerParticipant.doIt(CLIENT_NAME);
+
+        assertExpectedResults(result, PARTICIPANT_NAME1, CLIENT_NAME, _testStartTime, null, PAYLOAD_SIZE_PER_MESSAGE, null, duration);
+
+        verify(_delegate, atLeastOnce()).consumeMessage(PARTICIPANT_NAME1, RECEIVE_TIMEOUT);
+        verify(_delegate, atLeastOnce()).calculatePayloadSizeFrom(_mockMessage);
+        verify(_delegate, atLeastOnce()).commitOrAcknowledgeMessage(_mockMessage, SESSION_NAME1);
+    }
+
+    public void testReceiveMessagesBatchedSynch() throws Exception
+    {
+        int numberOfMessages = 10;
+        long totalPayloadSize = PAYLOAD_SIZE_PER_MESSAGE * numberOfMessages;
+        _command.setNumberOfMessages(numberOfMessages);
+        _command.setBatchSize(3);
+
+        ParticipantResult result = _consumerParticipant.doIt(CLIENT_NAME);
+
+        assertExpectedResults(result, PARTICIPANT_NAME1, CLIENT_NAME, _testStartTime, numberOfMessages, PAYLOAD_SIZE_PER_MESSAGE, totalPayloadSize, null);
+
+        verify(_delegate, times(numberOfMessages)).consumeMessage(PARTICIPANT_NAME1, RECEIVE_TIMEOUT);
+        verify(_delegate, times(numberOfMessages)).calculatePayloadSizeFrom(_mockMessage);
+        verify(_delegate, times(4)).commitOrAcknowledgeMessage(_mockMessage, SESSION_NAME1);
+    }
+
+    public void testReceiveMessagesWithVaryingPayloadSize() throws Exception
+    {
+        int numberOfMessages = 3;
+
+        int firstPayloadSize = PAYLOAD_SIZE_PER_MESSAGE;
+        int secondPayloadSize = PAYLOAD_SIZE_PER_MESSAGE * 2;
+        int thirdPayloadSize = PAYLOAD_SIZE_PER_MESSAGE * 4;
+
+        _command.setNumberOfMessages(numberOfMessages);
+
+        when(_delegate.calculatePayloadSizeFrom(_mockMessage)).thenReturn(firstPayloadSize, secondPayloadSize, thirdPayloadSize);
+
+        ParticipantResult result = _consumerParticipant.doIt(CLIENT_NAME);
+
+        final int expectedPayloadResultPayloadSize = 0;
+        final long totalPayloadSize = firstPayloadSize + secondPayloadSize + thirdPayloadSize;
+        assertExpectedResults(result, PARTICIPANT_NAME1, CLIENT_NAME, _testStartTime, numberOfMessages, expectedPayloadResultPayloadSize, totalPayloadSize, null);
+
+        verify(_delegate, times(numberOfMessages)).consumeMessage(PARTICIPANT_NAME1, RECEIVE_TIMEOUT);
+        verify(_delegate, times(numberOfMessages)).calculatePayloadSizeFrom(_mockMessage);
+        verify(_delegate, times(numberOfMessages)).commitOrAcknowledgeMessage(_mockMessage, SESSION_NAME1);
+    }
+
+    public void testReleaseResources()
+    {
+        _consumerParticipant.releaseResources();
+        verify(_delegate).closeTestConsumer(PARTICIPANT_NAME1);
+    }
+
+}
diff --git a/qpid/java/perftests/src/test/java/org/apache/qpid/disttest/client/MessageProviderTest.java b/qpid/java/perftests/src/test/java/org/apache/qpid/disttest/client/MessageProviderTest.java
new file mode 100644
index 0000000000..ffc3733eb7
--- /dev/null
+++ b/qpid/java/perftests/src/test/java/org/apache/qpid/disttest/client/MessageProviderTest.java
@@ -0,0 +1,107 @@
+/*
+ * 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.disttest.client;
+
+import static org.mockito.Matchers.isA;
+import static org.mockito.Mockito.atLeastOnce;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import javax.jms.Message;
+import javax.jms.Session;
+import javax.jms.TextMessage;
+
+import junit.framework.TestCase;
+
+import org.apache.qpid.disttest.client.property.ListPropertyValue;
+import org.apache.qpid.disttest.client.property.PropertyValue;
+import org.apache.qpid.disttest.client.property.SimplePropertyValue;
+import org.apache.qpid.disttest.message.CreateProducerCommand;
+
+public class MessageProviderTest extends TestCase
+{
+    private Session _session;
+    private TextMessage _message;
+
+    @Override
+    protected void setUp() throws Exception
+    {
+        super.setUp();
+        _session = mock(Session.class);
+        _message = mock(TextMessage.class);
+        when(_session.createTextMessage(isA(String.class))).thenReturn(_message);
+        when(_session.createTextMessage()).thenReturn(_message);
+    }
+
+    public void testGetMessagePayload() throws Exception
+    {
+        MessageProvider messageProvider = new MessageProvider(null)
+        {
+            public String getMessagePayload(CreateProducerCommand command)
+            {
+                return super.getMessagePayload(command);
+            }
+        };
+        CreateProducerCommand command = new CreateProducerCommand();
+        command.setMessageSize(100);
+        String payloadValue = messageProvider.getMessagePayload(command);
+        assertNotNull("Mesage payload should not be null", payloadValue);
+        assertEquals("Unexpected payload size", 100, payloadValue.length());
+    }
+
+    public void testNextMessage() throws Exception
+    {
+        MessageProvider messageProvider = new MessageProvider(null);
+        CreateProducerCommand command = new CreateProducerCommand();
+        command.setMessageSize(100);
+        Message message = messageProvider.nextMessage(_session, command);
+        assertNotNull("Mesage should be returned", message);
+        verify(_message, atLeastOnce()).setText(isA(String.class));
+    }
+
+    public void testNextMessageWithProperties() throws Exception
+    {
+        Map<String, PropertyValue> properties = new HashMap<String, PropertyValue>();
+        properties.put("test1", new SimplePropertyValue("testValue1"));
+        properties.put("test2", new SimplePropertyValue(new Integer(1)));
+        properties.put("priority", new SimplePropertyValue(new Integer(2)));
+        List<PropertyValue> listItems = new ArrayList<PropertyValue>();
+        listItems.add(new SimplePropertyValue(new Double(2.0)));
+        ListPropertyValue list = new ListPropertyValue();
+        list.setItems(listItems);
+        properties.put("test3", list);
+
+        MessageProvider messageProvider = new MessageProvider(properties);
+        CreateProducerCommand command = new CreateProducerCommand();
+        command.setMessageSize(100);
+        Message message = messageProvider.nextMessage(_session, command);
+        assertNotNull("Mesage should be returned", message);
+        verify(_message, atLeastOnce()).setText(isA(String.class));
+        verify(_message, atLeastOnce()).setJMSPriority(2);
+        verify(_message, atLeastOnce()).setStringProperty("test1", "testValue1");
+        verify(_message, atLeastOnce()).setIntProperty("test2", 1);
+        verify(_message, atLeastOnce()).setDoubleProperty("test3", 2.0);
+    }
+}
diff --git a/qpid/java/perftests/src/test/java/org/apache/qpid/disttest/client/ParticipantExecutorTest.java b/qpid/java/perftests/src/test/java/org/apache/qpid/disttest/client/ParticipantExecutorTest.java
new file mode 100644
index 0000000000..f30e4664ff
--- /dev/null
+++ b/qpid/java/perftests/src/test/java/org/apache/qpid/disttest/client/ParticipantExecutorTest.java
@@ -0,0 +1,183 @@
+/*
+ * 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.disttest.client;
+
+import static org.mockito.Matchers.argThat;
+import static org.mockito.Mockito.inOrder;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.Executor;
+import java.util.concurrent.TimeUnit;
+
+import junit.framework.TestCase;
+
+import org.apache.qpid.disttest.DistributedTestException;
+import org.apache.qpid.disttest.message.ParticipantResult;
+import org.mockito.ArgumentMatcher;
+import org.mockito.InOrder;
+
+public class ParticipantExecutorTest extends TestCase
+{
+    private static final ResultHasError HAS_ERROR = new ResultHasError();
+    private static final String CLIENT_NAME = "CLIENT_NAME";
+    private static final String PARTICIPANT_NAME = "PARTICIPANT_NAME";
+    private ParticipantExecutor _participantExecutor = null;
+    private Client _client = null;
+    private Participant _participant = null;
+    private ParticipantResult _mockResult;
+
+    @Override
+    protected void setUp() throws Exception
+    {
+        super.setUp();
+
+        _client = mock(Client.class);
+        when(_client.getClientName()).thenReturn(CLIENT_NAME);
+        _participant = mock(Participant.class);
+
+        _participantExecutor = new ParticipantExecutor(_participant);
+        _participantExecutor.setExecutor(new SynchronousExecutor());
+
+        _mockResult = mock(ParticipantResult.class);
+    }
+
+    public void testStart() throws Exception
+    {
+        when(_participant.doIt(CLIENT_NAME)).thenReturn(_mockResult);
+
+        _participantExecutor.start(_client);
+
+        InOrder inOrder = inOrder(_participant, _client);
+
+        inOrder.verify(_participant).doIt(CLIENT_NAME);
+        inOrder.verify(_client).sendResults(_mockResult);
+        inOrder.verify(_participant).releaseResources();
+    }
+
+    public void testParticipantThrowsException() throws Exception
+    {
+        when(_participant.doIt(CLIENT_NAME)).thenThrow(DistributedTestException.class);
+
+        _participantExecutor.start(_client);
+
+        InOrder inOrder = inOrder(_participant, _client);
+
+        inOrder.verify(_participant).doIt(CLIENT_NAME);
+        inOrder.verify(_client).sendResults(argThat(HAS_ERROR));
+        inOrder.verify(_participant).releaseResources();
+    }
+
+    public void testThreadNameAndDaemonness() throws Exception
+    {
+
+        ThreadPropertyReportingParticipant participant = new ThreadPropertyReportingParticipant(PARTICIPANT_NAME);
+        _participantExecutor = new ParticipantExecutor(participant);
+
+        _participantExecutor.start(_client);
+        participant.awaitExecution();
+
+        assertTrue("Participant should be run in a thread named after it", participant.threadWasCalled().endsWith(PARTICIPANT_NAME));
+        assertTrue("Executor should use daemon threads to avoid them preventing JVM termination", participant.wasDaemon());
+    }
+
+    private static final class ThreadPropertyReportingParticipant implements Participant
+    {
+        private final String _participantName;
+        private final CountDownLatch _participantExecuted = new CountDownLatch(1);
+        private String _threadName;
+        private boolean _daemon;
+
+        public ThreadPropertyReportingParticipant(String participantName)
+        {
+            _participantName = participantName;
+        }
+
+        public String threadWasCalled()
+        {
+            return _threadName;
+        }
+
+        public boolean wasDaemon()
+        {
+            return _daemon;
+        }
+
+        @Override
+        public void releaseResources()
+        {
+        }
+
+        @Override
+        public String getName()
+        {
+            return _participantName;
+        }
+
+        @Override
+        public ParticipantResult doIt(String registeredClientName) throws Exception
+        {
+            Thread currentThread = Thread.currentThread();
+            _threadName = currentThread.getName();
+            _daemon = currentThread.isDaemon();
+
+            _participantExecuted.countDown();
+
+            return null; // unused
+        }
+
+        public void awaitExecution()
+        {
+            boolean success = false;
+            try
+            {
+                success = _participantExecuted.await(5, TimeUnit.SECONDS);
+            }
+            catch (InterruptedException e)
+            {
+                Thread.currentThread().interrupt();
+            }
+
+            assertTrue("Participant not executed", success);
+        }
+    }
+
+    /** avoids our unit test needing to use multiple threads */
+    private final class SynchronousExecutor implements Executor
+    {
+        @Override
+        public void execute(Runnable command)
+        {
+            command.run();
+        }
+    }
+
+    private static class ResultHasError extends ArgumentMatcher<ParticipantResult>
+    {
+        @Override
+        public boolean matches(Object argument)
+        {
+            ParticipantResult result = (ParticipantResult) argument;
+            return result.hasError();
+        }
+    }
+
+}
diff --git a/qpid/java/perftests/src/test/java/org/apache/qpid/disttest/client/ParticipantRegistryTest.java b/qpid/java/perftests/src/test/java/org/apache/qpid/disttest/client/ParticipantRegistryTest.java
new file mode 100644
index 0000000000..bd0d5a39c8
--- /dev/null
+++ b/qpid/java/perftests/src/test/java/org/apache/qpid/disttest/client/ParticipantRegistryTest.java
@@ -0,0 +1,55 @@
+/*
+ * 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.disttest.client;
+
+import static org.mockito.Mockito.mock;
+import junit.framework.TestCase;
+
+public class ParticipantRegistryTest extends TestCase
+{
+    private ParticipantExecutorRegistry _participantRegistry = new ParticipantExecutorRegistry();
+
+    private ParticipantExecutor _testParticipant1 = mock(ParticipantExecutor.class);
+    private ParticipantExecutor _testParticipant2 = mock(ParticipantExecutor.class);
+
+    public void testAdd()
+    {
+        assertTrue(_participantRegistry.executors().isEmpty());
+
+        _participantRegistry.add(_testParticipant1);
+
+        assertTrue(_participantRegistry.executors().contains(_testParticipant1));
+
+        _participantRegistry.add(_testParticipant2);
+
+        assertTrue(_participantRegistry.executors().contains(_testParticipant2));
+    }
+
+    public void testClear()
+    {
+        _participantRegistry.add(_testParticipant1);
+
+        assertEquals(1, _participantRegistry.executors().size());
+
+        _participantRegistry.clear();
+
+        assertTrue(_participantRegistry.executors().isEmpty());
+    }
+}
diff --git a/qpid/java/perftests/src/test/java/org/apache/qpid/disttest/client/ParticipantResultFactoryTest.java b/qpid/java/perftests/src/test/java/org/apache/qpid/disttest/client/ParticipantResultFactoryTest.java
new file mode 100644
index 0000000000..73836f68e5
--- /dev/null
+++ b/qpid/java/perftests/src/test/java/org/apache/qpid/disttest/client/ParticipantResultFactoryTest.java
@@ -0,0 +1,168 @@
+/*
+ * 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.disttest.client;
+
+import java.util.Date;
+
+import javax.jms.DeliveryMode;
+
+import junit.framework.TestCase;
+
+import org.apache.qpid.disttest.message.ConsumerParticipantResult;
+import org.apache.qpid.disttest.message.CreateConsumerCommand;
+import org.apache.qpid.disttest.message.CreateParticpantCommand;
+import org.apache.qpid.disttest.message.CreateProducerCommand;
+import org.apache.qpid.disttest.message.ParticipantResult;
+import org.apache.qpid.disttest.message.ProducerParticipantResult;
+
+public class ParticipantResultFactoryTest extends TestCase
+{
+    private static final String PARTICIPANT_NAME = "participantName";
+    private static final String REGISTERED_CLIENT_NAME = "registeredClientName";
+
+    private static final int BATCH_SIZE = 10;
+    private static final long MAXIMUM_DURATION = 500;
+    private static final int NUMBER_OF_MESSAGES_PROCESSED = 100;
+    private static final long TIME_TAKEN = 100;
+    private static final long TOTAL_PAYLOAD_PROCESSED = 200;
+    private static final int PAYLOAD_SIZE = 300;
+
+    private static final Date START = new Date(0);
+    private static final Date END = new Date(START.getTime() + TIME_TAKEN);
+
+    private ParticipantResultFactory _participantResultFactory;
+
+    @Override
+    protected void setUp() throws Exception
+    {
+        super.setUp();
+
+        _participantResultFactory = new ParticipantResultFactory();
+    }
+
+    public void testCreateForProducer()
+    {
+        CreateProducerCommand command = new CreateProducerCommand();
+        setCommonCommandFields(command);
+
+        long producerStartDelay = 30;
+        command.setStartDelay(producerStartDelay);
+
+        int deliveryMode = DeliveryMode.PERSISTENT;
+        command.setDeliveryMode(deliveryMode);
+
+        int priority = 5;
+        command.setPriority(priority);
+
+        long producerInterval = 50;
+        command.setInterval(producerInterval);
+
+        long timeToLive = 60;
+        command.setTimeToLive(timeToLive);
+
+        ProducerParticipantResult result = _participantResultFactory.createForProducer(PARTICIPANT_NAME,
+                                                                                       REGISTERED_CLIENT_NAME,
+                                                                                       command,
+                                                                                       NUMBER_OF_MESSAGES_PROCESSED,
+                                                                                       PAYLOAD_SIZE,
+                                                                                       TOTAL_PAYLOAD_PROCESSED,
+                                                                                       START,
+                                                                                       END);
+
+        assertCommonResultProperties(result);
+
+        assertEquals(deliveryMode, result.getDeliveryMode());
+        assertEquals(priority, result.getPriority());
+        assertEquals(producerInterval, result.getInterval());
+        assertEquals(producerStartDelay, result.getStartDelay());
+        assertEquals(timeToLive, result.getTimeToLive());
+    }
+
+    public void testCreateForConsumer()
+    {
+        CreateConsumerCommand command = new CreateConsumerCommand();
+        setCommonCommandFields(command);
+
+        boolean topic = true;
+        command.setTopic(topic);
+
+        boolean durable = true;
+        command.setDurableSubscription(durable);
+
+        boolean browsingSubscription = false;
+        command.setBrowsingSubscription(browsingSubscription);
+
+        String selector = "selector";
+        boolean isSelector = true;
+        command.setSelector(selector);
+
+        boolean noLocal = false;
+        command.setNoLocal(noLocal);
+
+        boolean synchronousConsumer = true;
+        command.setSynchronous(synchronousConsumer);
+
+        ConsumerParticipantResult result = _participantResultFactory.createForConsumer(PARTICIPANT_NAME,
+                                                                                       REGISTERED_CLIENT_NAME,
+                                                                                       command,
+                                                                                       NUMBER_OF_MESSAGES_PROCESSED,
+                                                                                       PAYLOAD_SIZE,
+                                                                                       TOTAL_PAYLOAD_PROCESSED,
+                                                                                       START,
+                                                                                       END);
+
+        assertCommonResultProperties(result);
+
+        assertEquals(topic,                  result.isTopic());
+        assertEquals(durable,                result.isDurableSubscription());
+        assertEquals(browsingSubscription,   result.isBrowsingSubscription());
+        assertEquals(isSelector,             result.isSelector());
+        assertEquals(noLocal,                result.isNoLocal());
+        assertEquals(synchronousConsumer,    result.isSynchronousConsumer());
+    }
+
+    public void testCreateForError()
+    {
+        String errorMessage = "error";
+        ParticipantResult result = _participantResultFactory.createForError(PARTICIPANT_NAME, REGISTERED_CLIENT_NAME, errorMessage);
+        assertEquals(PARTICIPANT_NAME, result.getParticipantName());
+        assertEquals(REGISTERED_CLIENT_NAME, result.getRegisteredClientName());
+    }
+
+
+    private void setCommonCommandFields(CreateParticpantCommand command)
+    {
+        command.setBatchSize(BATCH_SIZE);
+        command.setMaximumDuration(MAXIMUM_DURATION);
+    }
+
+
+    private void assertCommonResultProperties(ParticipantResult result)
+    {
+        assertEquals(PARTICIPANT_NAME, result.getParticipantName());
+        assertEquals(REGISTERED_CLIENT_NAME, result.getRegisteredClientName());
+        assertEquals(BATCH_SIZE, result.getBatchSize());
+        assertEquals(MAXIMUM_DURATION, result.getMaximumDuration());
+        assertEquals(TIME_TAKEN, result.getTimeTaken());
+        assertEquals(NUMBER_OF_MESSAGES_PROCESSED, result.getNumberOfMessagesProcessed());
+        assertEquals(TOTAL_PAYLOAD_PROCESSED, result.getTotalPayloadProcessed());
+        assertEquals(PAYLOAD_SIZE, result.getPayloadSize());
+    }
+
+}
diff --git a/qpid/java/perftests/src/test/java/org/apache/qpid/disttest/client/ParticipantTestHelper.java b/qpid/java/perftests/src/test/java/org/apache/qpid/disttest/client/ParticipantTestHelper.java
new file mode 100644
index 0000000000..b30d5c1c7c
--- /dev/null
+++ b/qpid/java/perftests/src/test/java/org/apache/qpid/disttest/client/ParticipantTestHelper.java
@@ -0,0 +1,61 @@
+/*
+ * 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.disttest.client;
+
+import static junit.framework.Assert.assertEquals;
+import static junit.framework.Assert.assertFalse;
+import static junit.framework.Assert.assertTrue;
+
+import org.apache.qpid.disttest.message.ParticipantResult;
+
+public class ParticipantTestHelper
+{
+
+    public static void assertAtLeast(String message, final long minimumExpected, final long actual)
+    {
+        assertTrue(message + " " + actual, actual >= minimumExpected);
+    }
+
+    public static void assertExpectedResults(ParticipantResult result, String participantName, String registeredClientName, long expectedTestStartTime, Integer expectedNumberOfMessages, Integer expectedPayloadSize, Long expectedTotalPayloadProcessed, Long expectedMinimumExpectedDuration)
+    {
+        assertFalse(result.hasError());
+
+        assertEquals("unexpected participant name", participantName, result.getParticipantName());
+        assertEquals("unexpected client name", registeredClientName, result.getRegisteredClientName());
+
+        assertAtLeast("start time of result is too low", expectedTestStartTime, result.getStartInMillis());
+        assertAtLeast("end time of result should be after start time", result.getStartInMillis(), result.getEndInMillis());
+        if(expectedNumberOfMessages != null)
+        {
+            assertEquals("unexpected number of messages", expectedNumberOfMessages.intValue(), result.getNumberOfMessagesProcessed());
+        }
+        if (expectedPayloadSize != null)
+        {
+            assertEquals("unexpected payload size", expectedPayloadSize.intValue(), result.getPayloadSize());
+        }
+        if (expectedTotalPayloadProcessed != null)
+        {
+            assertEquals("unexpected total payload processed", expectedTotalPayloadProcessed.longValue(), result.getTotalPayloadProcessed());
+        }
+        if(expectedMinimumExpectedDuration != null)
+        {
+            assertAtLeast("participant did not take a sufficient length of time.", expectedMinimumExpectedDuration, result.getTimeTaken());
+        }
+    }
+}
diff --git a/qpid/java/perftests/src/test/java/org/apache/qpid/disttest/client/ProducerParticipantTest.java b/qpid/java/perftests/src/test/java/org/apache/qpid/disttest/client/ProducerParticipantTest.java
new file mode 100644
index 0000000000..3852948201
--- /dev/null
+++ b/qpid/java/perftests/src/test/java/org/apache/qpid/disttest/client/ProducerParticipantTest.java
@@ -0,0 +1,211 @@
+/*
+ * 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.disttest.client;
+
+import static org.apache.qpid.disttest.client.ParticipantTestHelper.assertExpectedResults;
+import static org.mockito.Matchers.isA;
+import static org.mockito.Mockito.atLeastOnce;
+import static org.mockito.Mockito.inOrder;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+import javax.jms.DeliveryMode;
+import javax.jms.Message;
+
+import junit.framework.TestCase;
+
+import org.apache.qpid.disttest.DistributedTestException;
+import org.apache.qpid.disttest.jms.ClientJmsDelegate;
+import org.apache.qpid.disttest.message.CreateProducerCommand;
+import org.apache.qpid.disttest.message.ParticipantResult;
+import org.mockito.InOrder;
+
+public class ProducerParticipantTest extends TestCase
+{
+    private ProducerParticipant _producer;
+
+    private static final String SESSION_NAME1 = "SESSION1";
+    private static final String PARTICIPANT_NAME1 = "PARTICIPANT_NAME1";
+
+    private static final String CLIENT_NAME = "CLIENT_NAME";
+    private static final int PAYLOAD_SIZE_PER_MESSAGE = 1024;
+
+
+    private final Message _mockMessage = mock(Message.class);
+    private final CreateProducerCommand _command = new CreateProducerCommand();
+    private ClientJmsDelegate _delegate;
+    private InOrder _inOrder;
+
+    /** used to check start/end time of results */
+    private long _testStartTime;
+
+    @Override
+    protected void setUp() throws Exception
+    {
+        super.setUp();
+        _delegate = mock(ClientJmsDelegate.class);
+        _inOrder = inOrder(_delegate);
+
+        _command.setSessionName(SESSION_NAME1);
+        _command.setParticipantName(PARTICIPANT_NAME1);
+
+        when(_delegate.sendNextMessage(isA(CreateProducerCommand.class))).thenReturn(_mockMessage);
+        when(_delegate.calculatePayloadSizeFrom(_mockMessage)).thenReturn(PAYLOAD_SIZE_PER_MESSAGE);
+
+        _producer = new ProducerParticipant(_delegate, _command);
+
+        _testStartTime = System.currentTimeMillis();
+    }
+
+    public void testStartDelay() throws Exception
+    {
+        final long delay = 100;
+        int numberOfMessages = 1;
+        long totalPayloadSize = PAYLOAD_SIZE_PER_MESSAGE * numberOfMessages;
+
+        _command.setStartDelay(delay);
+        _command.setNumberOfMessages(numberOfMessages);
+
+        ParticipantResult result = _producer.doIt(CLIENT_NAME);
+
+        long expectedPublishedStartTime = _testStartTime + delay;
+        assertExpectedResults(result, PARTICIPANT_NAME1, CLIENT_NAME, expectedPublishedStartTime, numberOfMessages, PAYLOAD_SIZE_PER_MESSAGE, totalPayloadSize, null);
+    }
+
+
+    public void testNoMessagesToSend() throws Exception
+    {
+        _command.setNumberOfMessages(0);
+        _command.setMaximumDuration(0);
+
+        try
+        {
+            _producer.doIt(CLIENT_NAME);
+            fail("Exception not thrown");
+        }
+        catch (DistributedTestException e)
+        {
+            // PASS
+            assertEquals("number of messages and duration cannot both be zero", e.getMessage());
+        }
+    }
+
+    public void testOneMessageToSend()  throws Exception
+    {
+        int batchSize = 1;
+        int numberOfMessages = 1;
+        long totalPayloadSize = PAYLOAD_SIZE_PER_MESSAGE * numberOfMessages;
+        int deliveryMode = DeliveryMode.PERSISTENT;
+
+        _command.setNumberOfMessages(numberOfMessages);
+        _command.setBatchSize(batchSize);
+        _command.setDeliveryMode(deliveryMode);
+
+        ParticipantResult result = (ParticipantResult) _producer.doIt(CLIENT_NAME);
+        assertExpectedResults(result, PARTICIPANT_NAME1, CLIENT_NAME, _testStartTime, numberOfMessages, PAYLOAD_SIZE_PER_MESSAGE, totalPayloadSize, null);
+
+        _inOrder.verify(_delegate).sendNextMessage(isA(CreateProducerCommand.class));
+        _inOrder.verify(_delegate).calculatePayloadSizeFrom(_mockMessage);
+        _inOrder.verify(_delegate).commitOrAcknowledgeMessage(_mockMessage, SESSION_NAME1);
+
+    }
+
+    public void testSendMessagesForDuration() throws Exception
+    {
+        final long duration = 100;
+        _command.setMaximumDuration(duration);
+
+        ParticipantResult result = _producer.doIt(CLIENT_NAME);
+        assertExpectedResults(result, PARTICIPANT_NAME1, CLIENT_NAME, _testStartTime, null, PAYLOAD_SIZE_PER_MESSAGE, null, duration);
+
+        verify(_delegate, atLeastOnce()).sendNextMessage(isA(CreateProducerCommand.class));
+        verify(_delegate, atLeastOnce()).calculatePayloadSizeFrom(_mockMessage);
+        verify(_delegate, atLeastOnce()).commitOrAcknowledgeMessage(_mockMessage, SESSION_NAME1);
+    }
+
+    public void testSendMessageBatches() throws Exception
+    {
+        final int numberOfMessages = 10;
+        final int expectedNumberOfCommits = 4; // one for each batch of 3 messages, plus one more at the end of the test for the tenth msg.
+        long totalPayloadSize = PAYLOAD_SIZE_PER_MESSAGE * numberOfMessages;
+
+        _command.setNumberOfMessages(numberOfMessages);
+        _command.setBatchSize(3);
+
+        ParticipantResult result = _producer.doIt(CLIENT_NAME);
+        assertExpectedResults(result, PARTICIPANT_NAME1, CLIENT_NAME, _testStartTime, numberOfMessages, PAYLOAD_SIZE_PER_MESSAGE, totalPayloadSize, null);
+
+        verify(_delegate, times(numberOfMessages)).sendNextMessage(isA(CreateProducerCommand.class));
+        verify(_delegate, times(numberOfMessages)).calculatePayloadSizeFrom(_mockMessage);
+        verify(_delegate, times(expectedNumberOfCommits)).commitOrAcknowledgeMessage(_mockMessage, SESSION_NAME1);
+    }
+
+    public void testSendMessageWithPublishInterval() throws Exception
+    {
+        final int batchSize = 3;
+        final long publishInterval = 100;
+        int numberOfMessages = 10;
+        long totalPayloadSize = PAYLOAD_SIZE_PER_MESSAGE * numberOfMessages;
+
+        final long expectedTimeToRunTest = batchSize * publishInterval;
+
+        _command.setNumberOfMessages(numberOfMessages);
+        _command.setBatchSize(batchSize);
+        _command.setInterval(publishInterval);
+
+        ParticipantResult result = _producer.doIt(CLIENT_NAME);
+        assertExpectedResults(result, PARTICIPANT_NAME1, CLIENT_NAME, _testStartTime, numberOfMessages, null, totalPayloadSize, expectedTimeToRunTest);
+
+        verify(_delegate, times(numberOfMessages)).sendNextMessage(isA(CreateProducerCommand.class));
+        verify(_delegate, times(numberOfMessages)).calculatePayloadSizeFrom(_mockMessage);
+        verify(_delegate, times(4)).commitOrAcknowledgeMessage(_mockMessage, SESSION_NAME1);
+    }
+
+    public void testSendMessageWithVaryingPayloadSize() throws Exception
+    {
+        int numberOfMessages = 3;
+
+        int firstPayloadSize = PAYLOAD_SIZE_PER_MESSAGE;
+        int secondPayloadSize = PAYLOAD_SIZE_PER_MESSAGE * 2;
+        int thirdPayloadSize = PAYLOAD_SIZE_PER_MESSAGE * 4;
+
+        final long totalPayloadSize = firstPayloadSize + secondPayloadSize + thirdPayloadSize;
+
+        when(_delegate.calculatePayloadSizeFrom(_mockMessage)).thenReturn(firstPayloadSize, secondPayloadSize, thirdPayloadSize);
+
+        _command.setNumberOfMessages(numberOfMessages);
+
+        ParticipantResult result = _producer.doIt(CLIENT_NAME);
+
+        final int expectedPayloadResultPayloadSize = 0;
+        assertExpectedResults(result, PARTICIPANT_NAME1, CLIENT_NAME, _testStartTime, numberOfMessages, expectedPayloadResultPayloadSize, totalPayloadSize, null);
+
+        verify(_delegate, times(numberOfMessages)).sendNextMessage(isA(CreateProducerCommand.class));
+        verify(_delegate, times(numberOfMessages)).calculatePayloadSizeFrom(_mockMessage);
+        verify(_delegate, times(numberOfMessages)).commitOrAcknowledgeMessage(_mockMessage, SESSION_NAME1);
+    }
+
+    public void testReleaseResources()
+    {
+        _producer.releaseResources();
+        verify(_delegate).closeTestProducer(PARTICIPANT_NAME1);
+    }
+}
diff --git a/qpid/java/perftests/src/test/java/org/apache/qpid/disttest/client/property/ListPropertyValueTest.java b/qpid/java/perftests/src/test/java/org/apache/qpid/disttest/client/property/ListPropertyValueTest.java
new file mode 100644
index 0000000000..75a634ba54
--- /dev/null
+++ b/qpid/java/perftests/src/test/java/org/apache/qpid/disttest/client/property/ListPropertyValueTest.java
@@ -0,0 +1,88 @@
+/*
+ * 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.disttest.client.property;
+
+import java.util.ArrayList;
+import java.util.List;
+
+import junit.framework.TestCase;
+
+import org.apache.qpid.disttest.client.property.ListPropertyValue;
+import org.apache.qpid.disttest.client.property.PropertyValue;
+import org.apache.qpid.disttest.client.property.SimplePropertyValue;
+
+public class ListPropertyValueTest extends TestCase
+{
+    private ListPropertyValue _generator;
+    private List<PropertyValue> _items;
+
+    public void setUp() throws Exception
+    {
+        super.setUp();
+        _generator = new ListPropertyValue();
+        _items = new ArrayList<PropertyValue>();
+        _items.add(new SimplePropertyValue(new Integer(1)));
+        _items.add(new SimplePropertyValue(new Double(2.1)));
+        _items.add(new SimplePropertyValue(new Boolean(true)));
+        ListPropertyValue innerList = new ListPropertyValue();
+        List<PropertyValue> innerListItems = new ArrayList<PropertyValue>();
+        innerListItems.add(new SimplePropertyValue("test"));
+        innerListItems.add(new SimplePropertyValue(new Integer(2)));
+        innerList.setItems(innerListItems);
+        _items.add(innerList);
+        _generator.setItems(_items);
+    }
+
+    public void testGetItems()
+    {
+        List<? extends Object> items = _generator.getItems();
+        assertEquals("Unexpected list items", _items, items);
+    }
+
+    public void testGetValue()
+    {
+        for (int i = 0; i < 2; i++)
+        {
+            assertEquals("Unexpected first item", new Integer(1), _generator.getValue());
+            assertEquals("Unexpected second item", new Double(2.1), _generator.getValue());
+            assertEquals("Unexpected third item", new Boolean(true), _generator.getValue());
+            if (i == 0)
+            {
+                assertEquals("Unexpected forth item", "test", _generator.getValue());
+            }
+            else
+            {
+                assertEquals("Unexpected forth item", new Integer(2), _generator.getValue());
+            }
+        }
+    }
+
+    public void testNonCyclicGetValue()
+    {
+        _generator.setCyclic(false);
+        assertFalse("Generator should not be cyclic", _generator.isCyclic());
+        assertEquals("Unexpected first item", new Integer(1), _generator.getValue());
+        assertEquals("Unexpected second item", new Double(2.1), _generator.getValue());
+        assertEquals("Unexpected third item", new Boolean(true), _generator.getValue());
+        assertEquals("Unexpected forth item", "test", _generator.getValue());
+        assertEquals("Unexpected fifth item", new Integer(2), _generator.getValue());
+        assertEquals("Unexpected sixs item", "test", _generator.getValue());
+        assertEquals("Unexpected sevens item", new Integer(2), _generator.getValue());
+    }
+}
diff --git a/qpid/java/perftests/src/test/java/org/apache/qpid/disttest/client/property/PropertyValueFactoryTest.java b/qpid/java/perftests/src/test/java/org/apache/qpid/disttest/client/property/PropertyValueFactoryTest.java
new file mode 100644
index 0000000000..2d560163c2
--- /dev/null
+++ b/qpid/java/perftests/src/test/java/org/apache/qpid/disttest/client/property/PropertyValueFactoryTest.java
@@ -0,0 +1,73 @@
+/*
+ * 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.disttest.client.property;
+
+import junit.framework.TestCase;
+
+public class PropertyValueFactoryTest extends TestCase
+{
+    private PropertyValueFactory _factory;
+
+    public void setUp() throws Exception
+    {
+        super.setUp();
+        _factory = new PropertyValueFactory();
+    }
+
+    public void testCreateListPropertyValue()
+    {
+        PropertyValue propertyValue = _factory.createPropertyValue("list");
+        assertNotNull("List generator is not created", propertyValue);
+        assertTrue("Unexpected type of list generator", propertyValue instanceof ListPropertyValue);
+    }
+
+    public void testCreateRangePropertyValue()
+    {
+        PropertyValue propertyValue = _factory.createPropertyValue("range");
+        assertNotNull("Range generator is not created", propertyValue);
+        assertTrue("Unexpected type of range generator", propertyValue instanceof RangePropertyValue);
+    }
+
+    public void testCreateRandomPropertyValue()
+    {
+        PropertyValue propertyValue = _factory.createPropertyValue("random");
+        assertNotNull("Random generator is not created", propertyValue);
+        assertTrue("Unexpected type of range generator", propertyValue instanceof RandomPropertyValue);
+    }
+
+    public void testCreateSimplePropertyValue()
+    {
+        PropertyValue propertyValue = _factory.createPropertyValue("simple");
+        assertNotNull("Simple property value is not created", propertyValue);
+        assertTrue("Unexpected type of property value", propertyValue instanceof SimplePropertyValue);
+    }
+
+    public void testCreateNonExistingPropertyValue()
+    {
+        try
+        {
+            _factory.createPropertyValue("nonExisting");
+            fail("Non existing property value should not be created");
+        }
+        catch (Exception e)
+        {
+            // pass
+        }
+    }
+}
diff --git a/qpid/java/perftests/src/test/java/org/apache/qpid/disttest/client/property/RandomPropertyValueTest.java b/qpid/java/perftests/src/test/java/org/apache/qpid/disttest/client/property/RandomPropertyValueTest.java
new file mode 100644
index 0000000000..bd5de3e370
--- /dev/null
+++ b/qpid/java/perftests/src/test/java/org/apache/qpid/disttest/client/property/RandomPropertyValueTest.java
@@ -0,0 +1,76 @@
+/*
+ * 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.disttest.client.property;
+
+import org.apache.qpid.disttest.client.property.RandomPropertyValue;
+
+import junit.framework.TestCase;
+
+public class RandomPropertyValueTest extends TestCase
+{
+    private RandomPropertyValue _generator;
+
+    public void setUp() throws Exception
+    {
+        super.setUp();
+        _generator = new RandomPropertyValue();
+        _generator.setUpper(20.0);
+        _generator.setLower(10.0);
+        _generator.setType("double");
+    }
+
+    public void testGetters()
+    {
+        assertEquals("Unexpected upper boundary", new Double(20.0), _generator.getUpper());
+        assertEquals("Unexpected lower boundary", new Double(10.0), _generator.getLower());
+        assertEquals("Unexpected type", "double", _generator.getType());
+    }
+
+    public void testGetValue()
+    {
+        Object value = _generator.getValue();
+        assertTrue("Unexpected type", value instanceof Double);
+        assertTrue("Unexpected value", ((Double) value).doubleValue() >= 10.0
+                && ((Double) value).doubleValue() <= 20.0);
+    }
+
+    public void testGetValueInt()
+    {
+        _generator.setType("int");
+        Object value = _generator.getValue();
+        assertTrue("Unexpected type", value instanceof Integer);
+        assertTrue("Unexpected value", ((Integer) value).intValue() >= 10 && ((Integer) value).intValue() <= 20);
+    }
+
+    public void testGetValueLong()
+    {
+        _generator.setType("long");
+        Object value = _generator.getValue();
+        assertTrue("Unexpected type", value instanceof Long);
+        assertTrue("Unexpected value", ((Long) value).longValue() >= 10 && ((Long) value).longValue() <= 20);
+    }
+
+    public void testGetValueFloat()
+    {
+        _generator.setType("float");
+        Object value = _generator.getValue();
+        assertTrue("Unexpected type", value instanceof Float);
+        assertTrue("Unexpected value", ((Float) value).floatValue() >= 10.0 && ((Float) value).floatValue() <= 20.0);
+    }
+}
diff --git a/qpid/java/perftests/src/test/java/org/apache/qpid/disttest/client/property/RangePropertyValueTest.java b/qpid/java/perftests/src/test/java/org/apache/qpid/disttest/client/property/RangePropertyValueTest.java
new file mode 100644
index 0000000000..91791c9d55
--- /dev/null
+++ b/qpid/java/perftests/src/test/java/org/apache/qpid/disttest/client/property/RangePropertyValueTest.java
@@ -0,0 +1,153 @@
+/*
+ * 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.disttest.client.property;
+
+import org.apache.qpid.disttest.client.property.RangePropertyValue;
+
+import junit.framework.TestCase;
+
+public class RangePropertyValueTest extends TestCase
+{
+    private RangePropertyValue _generator;
+
+    public void setUp() throws Exception
+    {
+        super.setUp();
+        _generator = new RangePropertyValue();
+        _generator.setUpper(10.0);
+        _generator.setLower(0.0);
+        _generator.setStep(2.0);
+        _generator.setType("double");
+    }
+
+    public void testGetters()
+    {
+        assertEquals("Unexpected upper boundary", new Double(10.0), _generator.getUpper());
+        assertEquals("Unexpected lower boundary", new Double(0.0), _generator.getLower());
+        assertEquals("Unexpected step", new Double(2.0), _generator.getStep());
+        assertEquals("Unexpected type", "double", _generator.getType());
+        assertTrue("Unexpected cyclic", _generator.isCyclic());
+    }
+
+    public void testGetValue()
+    {
+        double[] expected = { 0.0, 2.0, 4.0, 6.0, 8.0, 10.0 };
+        for (int j = 0; j < 2; j++)
+        {
+            for (int i = 0; i < expected.length; i++)
+            {
+                Object value = _generator.getValue();
+                assertTrue("Should be Double", value instanceof Double);
+                assertEquals("Unexpected value ", expected[i], value);
+            }
+        }
+    }
+
+    public void testGetValueNonCyclic()
+    {
+        _generator.setCyclic(false);
+        double[] expected = { 0.0, 2.0, 4.0, 6.0, 8.0, 10.0 };
+        for (int i = 0; i < expected.length; i++)
+        {
+            Object value = _generator.getValue();
+            assertTrue("Should be Double", value instanceof Double);
+            assertEquals("Unexpected value ", expected[i], value);
+        }
+        for (int i = 0; i < expected.length; i++)
+        {
+            Object value = _generator.getValue();
+            assertEquals("Unexpected value ", expected[expected.length - 1], value);
+        }
+    }
+
+    public void testGetValueInt()
+    {
+        _generator.setType("int");
+        int[] expected = { 0, 2, 4, 6, 8, 10 };
+        for (int j = 0; j < 2; j++)
+        {
+            for (int i = 0; i < expected.length; i++)
+            {
+                Object value = _generator.getValue();
+                assertTrue("Should be Double", value instanceof Integer);
+                assertEquals("Unexpected value ", expected[i], value);
+            }
+        }
+    }
+
+    public void testGetValueByte()
+    {
+        _generator.setType("byte");
+        byte[] expected = { 0, 2, 4, 6, 8, 10 };
+        for (int j = 0; j < 2; j++)
+        {
+            for (int i = 0; i < expected.length; i++)
+            {
+                Object value = _generator.getValue();
+                assertTrue("Should be Double", value instanceof Byte);
+                assertEquals("Unexpected value ", expected[i], value);
+            }
+        }
+    }
+
+    public void testGetValueLong()
+    {
+        _generator.setType("long");
+        long[] expected = { 0, 2, 4, 6, 8, 10 };
+        for (int j = 0; j < 2; j++)
+        {
+            for (int i = 0; i < expected.length; i++)
+            {
+                Object value = _generator.getValue();
+                assertTrue("Should be Double", value instanceof Long);
+                assertEquals("Unexpected value ", expected[i], value);
+            }
+        }
+    }
+
+    public void testGetValueShort()
+    {
+        _generator.setType("short");
+        short[] expected = { 0, 2, 4, 6, 8, 10 };
+        for (int j = 0; j < 2; j++)
+        {
+            for (int i = 0; i < expected.length; i++)
+            {
+                Object value = _generator.getValue();
+                assertTrue("Should be Double", value instanceof Short);
+                assertEquals("Unexpected value ", expected[i], value);
+            }
+        }
+    }
+
+    public void testGetValueFloat()
+    {
+        _generator.setType("float");
+        float[] expected = { 0.0f, 2.0f, 4.0f, 6.0f, 8.0f, 10.0f };
+        for (int j = 0; j < 2; j++)
+        {
+            for (int i = 0; i < expected.length; i++)
+            {
+                Object value = _generator.getValue();
+                assertTrue("Should be Double", value instanceof Float);
+                assertEquals("Unexpected value ", expected[i], value);
+            }
+        }
+    }
+}
diff --git a/qpid/java/perftests/src/test/java/org/apache/qpid/disttest/client/property/SimplePropertyValueTest.java b/qpid/java/perftests/src/test/java/org/apache/qpid/disttest/client/property/SimplePropertyValueTest.java
new file mode 100644
index 0000000000..a347d866c7
--- /dev/null
+++ b/qpid/java/perftests/src/test/java/org/apache/qpid/disttest/client/property/SimplePropertyValueTest.java
@@ -0,0 +1,30 @@
+/*
+ * 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.disttest.client.property;
+
+import junit.framework.TestCase;
+
+public class SimplePropertyValueTest extends TestCase
+{
+    public void testGetValue()
+    {
+        SimplePropertyValue value = new SimplePropertyValue(new Integer(1));
+        assertEquals("Unexpected value", new Integer(1), value.getValue());
+    }
+}
diff --git a/qpid/java/perftests/src/test/java/org/apache/qpid/disttest/controller/ClientRegistryTest.java b/qpid/java/perftests/src/test/java/org/apache/qpid/disttest/controller/ClientRegistryTest.java
new file mode 100644
index 0000000000..2c161012a9
--- /dev/null
+++ b/qpid/java/perftests/src/test/java/org/apache/qpid/disttest/controller/ClientRegistryTest.java
@@ -0,0 +1,57 @@
+/*
+ * 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.disttest.controller;
+
+import junit.framework.TestCase;
+
+import org.apache.qpid.disttest.DistributedTestException;
+
+public class ClientRegistryTest extends TestCase
+{
+    private static final String CLIENT1_REGISTERED_NAME = "CLIENT1_REGISTERED_NAME";
+
+    private ClientRegistry _clientRegistry = new ClientRegistry();
+
+    public void testRegisterClient()
+    {
+        assertEquals(0, _clientRegistry.getClients().size());
+
+        _clientRegistry.registerClient(CLIENT1_REGISTERED_NAME);
+        assertEquals(1, _clientRegistry.getClients().size());
+
+    }
+
+    public void testRejectsDuplicateClientNames()
+    {
+        _clientRegistry.registerClient(CLIENT1_REGISTERED_NAME);
+        try
+        {
+            _clientRegistry.registerClient(CLIENT1_REGISTERED_NAME);
+            fail("Should have thrown an exception");
+        }
+        catch (final DistributedTestException e)
+        {
+            // pass
+        }
+    }
+
+
+
+
+}
diff --git a/qpid/java/perftests/src/test/java/org/apache/qpid/disttest/controller/ControllerTest.java b/qpid/java/perftests/src/test/java/org/apache/qpid/disttest/controller/ControllerTest.java
new file mode 100644
index 0000000000..c119656afd
--- /dev/null
+++ b/qpid/java/perftests/src/test/java/org/apache/qpid/disttest/controller/ControllerTest.java
@@ -0,0 +1,253 @@
+/*
+ * 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.disttest.controller;
+
+import static org.mockito.Matchers.anyString;
+import static org.mockito.Matchers.eq;
+import static org.mockito.Matchers.isA;
+import static org.mockito.Mockito.doAnswer;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+import java.util.Timer;
+import java.util.TimerTask;
+
+import junit.framework.TestCase;
+
+import org.apache.qpid.disttest.DistributedTestException;
+import org.apache.qpid.disttest.controller.config.Config;
+import org.apache.qpid.disttest.controller.config.TestInstance;
+import org.apache.qpid.disttest.jms.ControllerJmsDelegate;
+import org.apache.qpid.disttest.message.Command;
+import org.apache.qpid.disttest.message.RegisterClientCommand;
+import org.apache.qpid.disttest.message.Response;
+import org.apache.qpid.disttest.message.StopClientCommand;
+import org.apache.qpid.disttest.results.aggregation.ITestResult;
+import org.mockito.invocation.InvocationOnMock;
+import org.mockito.stubbing.Answer;
+
+public class ControllerTest extends TestCase
+{
+    private static final String CLIENT1_REGISTERED_NAME = "client-uid1";
+    private static final String CLIENT2_REGISTERED_NAME = "client-uid2";
+
+    private static final int DELAY = 100;
+    private static final long COMMAND_RESPONSE_TIMEOUT = 1000;
+    private static final long REGISTRATION_TIMEOUT = 1000;
+
+    private Controller _controller;
+    private ControllerJmsDelegate _respondingJmsDelegate;
+    private TestRunner _testRunner;
+    private ClientRegistry _clientRegistry;
+
+    @Override
+    protected void setUp() throws Exception
+    {
+        super.setUp();
+        _respondingJmsDelegate = mock(ControllerJmsDelegate.class);
+        _controller = new Controller(_respondingJmsDelegate, REGISTRATION_TIMEOUT, COMMAND_RESPONSE_TIMEOUT);
+        _testRunner = mock(TestRunner.class);
+        _clientRegistry = mock(ClientRegistry.class);
+
+        Config configWithOneClient = createMockConfig(1);
+        _controller.setConfig(configWithOneClient);
+        _controller.setClientRegistry(_clientRegistry);
+        _controller.setTestRunnerFactory(createTestFactoryReturningMock());
+
+        doAnswer(new Answer<Void>()
+        {
+            @Override
+            public Void answer(InvocationOnMock invocation) throws Throwable
+            {
+                final String clientName = (String)invocation.getArguments()[0];
+                final Command command = (Command)invocation.getArguments()[1];
+                _controller.processStopClientResponse(new Response(clientName, command.getType()));
+                return null;
+            }
+        }).when(_respondingJmsDelegate).sendCommandToClient(anyString(), isA(Command.class));
+    }
+
+
+    public void testControllerRejectsEmptyConfiguration()
+    {
+        Config configWithZeroClients = createMockConfig(0);
+
+        try
+        {
+            _controller.setConfig(configWithZeroClients);
+            fail("Exception not thrown");
+        }
+        catch (DistributedTestException e)
+        {
+            // PASS
+        }
+    }
+
+    public void testControllerRequiresOneClientRegistration()
+    {
+        Config configWithOneClient = createMockConfig(1);
+
+        _controller.setConfig(configWithOneClient);
+        registerClientAndAwait(CLIENT1_REGISTERED_NAME);
+    }
+
+    public void testControllerReceivesTwoExpectedClientRegistrations()
+    {
+        Config configWithTwoClients = createMockConfig(2);
+        _controller.setConfig(configWithTwoClients);
+
+        registerClientLater(_controller, CLIENT1_REGISTERED_NAME);
+        registerClientLater(_controller, CLIENT2_REGISTERED_NAME);
+        _controller.awaitClientRegistrations();
+    }
+
+    public void testControllerDoesntReceiveAnyRegistrations()
+    {
+        try
+        {
+            _controller.awaitClientRegistrations();
+            fail("Exception not thrown");
+        }
+        catch (DistributedTestException e)
+        {
+            // PASS
+        }
+    }
+
+    public void testControllerDoesntReceiveTwoExpectedClientRegistrations()
+    {
+        Config configWithTwoClients = createMockConfig(2);
+        _controller.setConfig(configWithTwoClients);
+
+        registerClientLater(_controller, CLIENT1_REGISTERED_NAME); // only receives one out of two expected registrations
+        try
+        {
+            _controller.awaitClientRegistrations();
+            fail("Exception not thrown");
+        }
+        catch (DistributedTestException e)
+        {
+            // PASS
+        }
+    }
+
+    public void testRegisterClient()
+    {
+        RegisterClientCommand command = new RegisterClientCommand(CLIENT1_REGISTERED_NAME, "dummy");
+        _controller.registerClient(command);
+
+        verify(_clientRegistry).registerClient(CLIENT1_REGISTERED_NAME);
+        verify(_respondingJmsDelegate).registerClient(command);
+
+    }
+
+    public void testControllerSendsClientStopCommandToClient()
+    {
+        when(_clientRegistry.getClients()).thenReturn(Collections.singleton(CLIENT1_REGISTERED_NAME));
+
+        _controller.stopAllRegisteredClients();
+
+        verify(_respondingJmsDelegate).sendCommandToClient(eq(CLIENT1_REGISTERED_NAME), isA(StopClientCommand.class));
+    }
+
+    public void testRunAllTests()
+    {
+        Config config = createSimpleConfig();
+        _controller.setConfig(config);
+
+        TestResult testResult = new TestResult("test1");
+
+        when(_testRunner.run()).thenReturn(testResult);
+
+        ResultsForAllTests results = _controller.runAllTests();
+
+        List<ITestResult> testResults = results.getTestResults();
+        assertEquals(1, testResults.size());
+        assertSame(testResult, testResults.get(0));
+
+        verify(_testRunner).run();
+    }
+
+    private Config createSimpleConfig()
+    {
+        Config config = mock(Config.class);
+        TestInstance testInstance = mock(TestInstance.class);
+
+        List<TestInstance> testInstances = Arrays.asList(testInstance);
+
+        when(config.getTests()).thenReturn(testInstances);
+        when(config.getTotalNumberOfClients()).thenReturn(1); // necessary otherwise controller rejects "invalid" config
+
+        return config;
+    }
+
+    private void doLater(TimerTask task, long delayInMillis)
+    {
+        Timer timer = new Timer();
+        timer.schedule(task, delayInMillis);
+    }
+
+    private Config createMockConfig(int numberOfClients)
+    {
+        Config config = mock(Config.class);
+        when(config.getTotalNumberOfClients()).thenReturn(numberOfClients);
+        return config;
+    }
+
+    private void registerClientAndAwait(String... clientNames)
+    {
+        for (String clientName : clientNames)
+        {
+            registerClientLater(_controller, clientName);
+        }
+        _controller.awaitClientRegistrations();
+    }
+
+    private void registerClientLater(final Controller controller, final String clientName)
+    {
+        doLater(new TimerTask()
+        {
+            @Override
+            public void run()
+            {
+                controller.registerClient(new RegisterClientCommand(clientName, "dummy"));
+            }
+        }, DELAY);
+    }
+
+    private TestRunnerFactory createTestFactoryReturningMock()
+    {
+        TestRunnerFactory testRunnerFactory = mock(TestRunnerFactory.class);
+
+        when(testRunnerFactory.createTestRunner(
+                isA(ParticipatingClients.class),
+                isA(TestInstance.class),
+                isA(ControllerJmsDelegate.class),
+                isA(Long.class),
+                isA(Long.class)))
+                .thenReturn(_testRunner);
+
+        return testRunnerFactory;
+    }
+}
diff --git a/qpid/java/perftests/src/test/java/org/apache/qpid/disttest/controller/ParticipatingClientsTest.java b/qpid/java/perftests/src/test/java/org/apache/qpid/disttest/controller/ParticipatingClientsTest.java
new file mode 100644
index 0000000000..284db38f44
--- /dev/null
+++ b/qpid/java/perftests/src/test/java/org/apache/qpid/disttest/controller/ParticipatingClientsTest.java
@@ -0,0 +1,144 @@
+/*
+ * 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.disttest.controller;
+
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.List;
+
+import junit.framework.TestCase;
+
+public class ParticipatingClientsTest extends TestCase
+{
+    private static final String CLIENT1_CONFIGURED_NAME = "CLIENT1_CONFIGURED_NAME";
+    private static final String CLIENT2_CONFIGURED_NAME = "CLIENT2_CONFIGURED_NAME";
+
+    private static final String CLIENT1_REGISTERED_NAME = "CLIENT1_REGISTERED_NAME";
+    private static final String CLIENT2_REGISTERED_NAME = "CLIENT2_REGISTERED_NAME";
+    private static final String CLIENT3_REGISTERED_NAME = "CLIENT3_REGISTERED_NAME";
+    private ClientRegistry _clientRegistry;
+    private List<String> _configuredClientNamesForTest;
+
+    @Override
+    protected void setUp() throws Exception
+    {
+        super.setUp();
+        _clientRegistry = mock(ClientRegistry.class);
+    }
+
+    public void testTooFewRegisteredClientsForTest()
+    {
+        _configuredClientNamesForTest = Arrays.asList(CLIENT1_CONFIGURED_NAME, CLIENT2_CONFIGURED_NAME);
+        when(_clientRegistry.getClients()).thenReturn(Arrays.asList(CLIENT1_REGISTERED_NAME));
+
+        try
+        {
+            new ParticipatingClients(_clientRegistry, _configuredClientNamesForTest);
+            fail("Exception not thrown");
+        }
+        catch (IllegalArgumentException e)
+        {
+            // PASS
+        }
+
+    }
+
+
+    public void testSelectOneClientFromPoolOfOne()
+    {
+        _configuredClientNamesForTest = Arrays.asList(CLIENT1_CONFIGURED_NAME);
+        when(_clientRegistry.getClients()).thenReturn(Arrays.asList(CLIENT1_REGISTERED_NAME));
+
+        ParticipatingClients clients = new ParticipatingClients(_clientRegistry, _configuredClientNamesForTest);
+        assertBothWays(clients, CLIENT1_REGISTERED_NAME, CLIENT1_CONFIGURED_NAME);
+    }
+
+    public void testSelectTwoClientFromPoolOfMany()
+    {
+        _configuredClientNamesForTest = Arrays.asList(CLIENT1_CONFIGURED_NAME, CLIENT2_CONFIGURED_NAME);
+        when(_clientRegistry.getClients()).thenReturn(Arrays.asList(CLIENT1_REGISTERED_NAME, CLIENT2_REGISTERED_NAME, CLIENT3_REGISTERED_NAME));
+
+        ParticipatingClients clients = new ParticipatingClients(_clientRegistry, _configuredClientNamesForTest);
+
+        assertBothWays(clients, CLIENT1_REGISTERED_NAME, CLIENT1_CONFIGURED_NAME);
+        assertBothWays(clients, CLIENT2_REGISTERED_NAME, CLIENT2_CONFIGURED_NAME);
+    }
+
+    public void testGetUnrecognisedConfiguredName()
+    {
+        _configuredClientNamesForTest = Arrays.asList(CLIENT1_CONFIGURED_NAME);
+        when(_clientRegistry.getClients()).thenReturn(Arrays.asList(CLIENT1_REGISTERED_NAME));
+
+        ParticipatingClients clients = new ParticipatingClients(_clientRegistry, _configuredClientNamesForTest);
+
+        testUnrecognisedClientConfiguredName(clients, "unknown");
+        testUnrecognisedClientRegisteredName(clients, "unknown");
+    }
+
+    public void testGetRegisteredClientNames()
+    {
+        _configuredClientNamesForTest = Arrays.asList(CLIENT1_CONFIGURED_NAME);
+        List<String> registeredNames = Arrays.asList(CLIENT1_REGISTERED_NAME);
+        when(_clientRegistry.getClients()).thenReturn(registeredNames);
+
+        ParticipatingClients clients = new ParticipatingClients(_clientRegistry, _configuredClientNamesForTest);
+
+        Collection<String> registeredParticipatingNames = clients.getRegisteredNames();
+        assertEquals(1, registeredParticipatingNames.size());
+        assertTrue(registeredParticipatingNames.contains(CLIENT1_REGISTERED_NAME));
+    }
+
+    private void testUnrecognisedClientConfiguredName(ParticipatingClients clients, String unrecognisedClientConfiguredName)
+    {
+        try
+        {
+            clients.getRegisteredNameFromConfiguredName(unrecognisedClientConfiguredName);
+            fail("Exception not thrown");
+        }
+        catch (IllegalArgumentException e)
+        {
+            // PASS
+        }
+    }
+
+    private void testUnrecognisedClientRegisteredName(ParticipatingClients clients, String unrecognisedClientRegisteredName)
+    {
+        try
+        {
+            clients.getConfiguredNameFromRegisteredName(unrecognisedClientRegisteredName);
+            fail("Exception not thrown");
+        }
+        catch (IllegalArgumentException e)
+        {
+            // PASS
+        }
+    }
+
+    private void assertBothWays(ParticipatingClients clients, String registeredName, String configuredName)
+    {
+        assertEquals(registeredName, clients.getRegisteredNameFromConfiguredName(configuredName));
+        assertEquals(configuredName, clients.getConfiguredNameFromRegisteredName(registeredName));
+    }
+
+
+
+}
diff --git a/qpid/java/perftests/src/test/java/org/apache/qpid/disttest/controller/TestRunnerTest.java b/qpid/java/perftests/src/test/java/org/apache/qpid/disttest/controller/TestRunnerTest.java
new file mode 100644
index 0000000000..ffd41049be
--- /dev/null
+++ b/qpid/java/perftests/src/test/java/org/apache/qpid/disttest/controller/TestRunnerTest.java
@@ -0,0 +1,252 @@
+/*
+ * 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.disttest.controller;
+
+import static org.mockito.Matchers.anyString;
+import static org.mockito.Matchers.eq;
+import static org.mockito.Matchers.isA;
+import static org.mockito.Mockito.doAnswer;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+import java.util.Timer;
+import java.util.TimerTask;
+
+import junit.framework.TestCase;
+
+import org.apache.qpid.disttest.DistributedTestException;
+import org.apache.qpid.disttest.controller.config.QueueConfig;
+import org.apache.qpid.disttest.controller.config.TestInstance;
+import org.apache.qpid.disttest.jms.ControllerJmsDelegate;
+import org.apache.qpid.disttest.message.Command;
+import org.apache.qpid.disttest.message.CreateConnectionCommand;
+import org.apache.qpid.disttest.message.NoOpCommand;
+import org.apache.qpid.disttest.message.ParticipantResult;
+import org.apache.qpid.disttest.message.Response;
+import org.apache.qpid.disttest.message.StartTestCommand;
+import org.apache.qpid.disttest.message.TearDownTestCommand;
+import org.mockito.invocation.InvocationOnMock;
+import org.mockito.stubbing.Answer;
+
+public class TestRunnerTest extends TestCase
+{
+    private static final String TEST_NAME = "TEST_NAME";
+    private static final int ITERATION_NUMBER = 1;
+
+    private static final String CLIENT1_REGISTERED_NAME = "client-uid1";
+    private static final String CLIENT1_CONFIGURED_NAME = "client1";
+
+    private static final long COMMAND_RESPONSE_TIMEOUT = 1000;
+    private static final long TEST_RESULT_TIMEOUT = 2000;
+    private static final long DELAY = 100;
+
+    private TestRunner _testRunner;
+    private TestInstance _testInstance;
+    private ControllerJmsDelegate _respondingJmsDelegate;
+    private ParticipatingClients _participatingClients;
+
+    @Override
+    protected void setUp() throws Exception
+    {
+        super.setUp();
+        _respondingJmsDelegate = mock(ControllerJmsDelegate.class);
+
+        _participatingClients = mock(ParticipatingClients.class);
+        when(_participatingClients.getRegisteredNameFromConfiguredName(CLIENT1_CONFIGURED_NAME)).thenReturn(CLIENT1_REGISTERED_NAME);
+        when(_participatingClients.getConfiguredNameFromRegisteredName(CLIENT1_REGISTERED_NAME)).thenReturn(CLIENT1_CONFIGURED_NAME);
+        when(_participatingClients.getRegisteredNames()).thenReturn(Collections.singleton(CLIENT1_REGISTERED_NAME));
+
+        doAnswer(new Answer<Void>()
+        {
+            @Override
+            public Void answer(InvocationOnMock invocation) throws Throwable
+            {
+                final String clientName = (String)invocation.getArguments()[0];
+                final Command command = (Command)invocation.getArguments()[1];
+                _testRunner.processCommandResponse(new Response(clientName, command.getType()));
+                return null;
+            }
+        }).when(_respondingJmsDelegate).sendCommandToClient(anyString(), isA(Command.class));
+    }
+
+    public void testSendConnectionCommandToClient()
+    {
+        _testInstance = createTestInstanceWithConnection();
+
+        _testRunner = new TestRunner(_participatingClients, _testInstance , _respondingJmsDelegate, COMMAND_RESPONSE_TIMEOUT, TEST_RESULT_TIMEOUT);
+        _testRunner.sendTestSetupCommands();
+
+        verify(_respondingJmsDelegate).sendCommandToClient(eq(CLIENT1_REGISTERED_NAME), isA(CreateConnectionCommand.class));
+    }
+
+    public void testSendCommandToAllParticipatingClients()
+    {
+        _testRunner = new TestRunner(_participatingClients, mock(TestInstance.class), _respondingJmsDelegate, COMMAND_RESPONSE_TIMEOUT, TEST_RESULT_TIMEOUT);
+
+        StartTestCommand startTestCommand = new StartTestCommand();
+        _testRunner.sendCommandToParticipatingClients(startTestCommand);
+
+        verify(_respondingJmsDelegate).sendCommandToClient(CLIENT1_REGISTERED_NAME, startTestCommand);
+    }
+
+    public void testWaitsForCommandResponses()
+    {
+        _testInstance = createTestInstanceWithConnection();
+        _testRunner = new TestRunner(_participatingClients, _testInstance , _respondingJmsDelegate, COMMAND_RESPONSE_TIMEOUT, TEST_RESULT_TIMEOUT);
+
+        _testRunner.sendTestSetupCommands();
+
+        _testRunner.awaitCommandResponses();
+    }
+
+    public void testClientFailsToSendCommandResponseWithinTimeout()
+    {
+        ControllerJmsDelegate jmsDelegate = mock(ControllerJmsDelegate.class);
+
+        _testInstance = createTestInstanceWithConnection();
+        _testRunner = new TestRunner(_participatingClients, _testInstance , jmsDelegate, COMMAND_RESPONSE_TIMEOUT, TEST_RESULT_TIMEOUT);
+
+        _testRunner.sendTestSetupCommands();
+        // we don't call sendCommandResponseLater so controller should time out
+
+        try
+        {
+            _testRunner.awaitCommandResponses();
+            fail("Exception not thrown");
+        }
+        catch (DistributedTestException e)
+        {
+            // PASS
+        }
+    }
+
+    public void testCreateAndDeleteQueues()
+    {
+        _testInstance =  mock(TestInstance.class);
+        List<QueueConfig> queues = mock(List.class);
+        when(_testInstance.getQueues()).thenReturn(queues);
+
+        _testRunner = new TestRunner(_participatingClients, _testInstance, _respondingJmsDelegate, COMMAND_RESPONSE_TIMEOUT, TEST_RESULT_TIMEOUT);
+
+        _testRunner.createQueues();
+        verify(_respondingJmsDelegate).createQueues(queues);
+
+        _testRunner.deleteQueues();
+        verify(_respondingJmsDelegate).deleteQueues(queues);
+    }
+
+    public void testRun()
+    {
+        _testInstance = createTestInstanceWithOneParticipant();
+        _testRunner = new TestRunner(_participatingClients, _testInstance , _respondingJmsDelegate, COMMAND_RESPONSE_TIMEOUT, TEST_RESULT_TIMEOUT);
+
+        ParticipantResult incomingParticipantResult = new ParticipantResult();
+        incomingParticipantResult.setRegisteredClientName(CLIENT1_REGISTERED_NAME);
+        sendTestResultsLater(_testRunner, incomingParticipantResult);
+
+        TestResult results = _testRunner.run();
+
+        verify(_respondingJmsDelegate).addCommandListener(isA(TestRunner.TestCommandResponseListener.class));
+        verify(_respondingJmsDelegate).addCommandListener(isA(TestRunner.ParticipantResultListener.class));
+
+        verify(_respondingJmsDelegate).createQueues(isA(List.class));
+
+        verify(_respondingJmsDelegate).sendCommandToClient(eq(CLIENT1_REGISTERED_NAME), isA(StartTestCommand.class));
+        verify(_respondingJmsDelegate).sendCommandToClient(eq(CLIENT1_REGISTERED_NAME), isA(NoOpCommand.class));
+        verify(_respondingJmsDelegate).sendCommandToClient(eq(CLIENT1_REGISTERED_NAME), isA(TearDownTestCommand.class));
+
+        verify(_respondingJmsDelegate).deleteQueues(isA(List.class));
+
+        verify(_respondingJmsDelegate).removeCommandListener(isA(TestRunner.ParticipantResultListener.class));
+        verify(_respondingJmsDelegate).removeCommandListener(isA(TestRunner.TestCommandResponseListener.class));
+
+        List<ParticipantResult> participantResults = results.getParticipantResults();
+        assertEquals(1, participantResults.size());
+        ParticipantResult resultingParticipantResult = participantResults.get(0);
+
+        assertResultHasCorrectTestDetails(resultingParticipantResult);
+    }
+
+    private void assertResultHasCorrectTestDetails(ParticipantResult resultingParticipantResult)
+    {
+        assertEquals("Test runner should have set configured name when it received participant results",
+                CLIENT1_CONFIGURED_NAME, resultingParticipantResult.getConfiguredClientName());
+        assertEquals("Test runner should have set test name when it received participant results",
+                TEST_NAME, resultingParticipantResult.getTestName());
+        assertEquals("Test runner should have set test iteration number when it received participant results",
+                ITERATION_NUMBER, resultingParticipantResult.getIterationNumber());
+    }
+
+
+    private TestInstance createTestInstanceWithOneParticipant()
+    {
+        TestInstance testInstance = mock(TestInstance.class);
+
+        List<CommandForClient> commands = Arrays.asList(
+                new CommandForClient(CLIENT1_CONFIGURED_NAME, new NoOpCommand()));
+
+        when(testInstance.createCommands()).thenReturn(commands);
+
+        when(testInstance.getTotalNumberOfParticipants()).thenReturn(1);
+
+        when(testInstance.getName()).thenReturn(TEST_NAME);
+
+        List<QueueConfig> queues = mock(List.class);
+        when(testInstance.getQueues()).thenReturn(queues);
+
+        when(testInstance.getIterationNumber()).thenReturn(ITERATION_NUMBER);
+
+        return testInstance;
+    }
+
+    private TestInstance createTestInstanceWithConnection()
+    {
+        TestInstance testInstance = mock(TestInstance.class);
+
+        List<CommandForClient> commands = Arrays.asList(
+                new CommandForClient(CLIENT1_CONFIGURED_NAME, new CreateConnectionCommand("conn1", "factory")));
+
+        when(testInstance.createCommands()).thenReturn(commands);
+
+        return testInstance;
+    }
+
+    private void sendTestResultsLater(final TestRunner runner, final ParticipantResult result)
+    {
+        doLater(new TimerTask()
+        {
+            @Override
+            public void run()
+            {
+                runner.processParticipantResult(result);
+            }
+        }, DELAY);
+    }
+
+    private void doLater(TimerTask task, long delayInMillis)
+    {
+        Timer timer = new Timer();
+        timer.schedule(task, delayInMillis);
+    }
+
+}
diff --git a/qpid/java/perftests/src/test/java/org/apache/qpid/disttest/controller/config/ClientConfigTest.java b/qpid/java/perftests/src/test/java/org/apache/qpid/disttest/controller/config/ClientConfigTest.java
new file mode 100644
index 0000000000..d4af439dea
--- /dev/null
+++ b/qpid/java/perftests/src/test/java/org/apache/qpid/disttest/controller/config/ClientConfigTest.java
@@ -0,0 +1,111 @@
+/*
+ * 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.disttest.controller.config;
+
+import static org.apache.qpid.disttest.controller.config.ConfigTestUtils.assertCommandForClient;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import junit.framework.TestCase;
+
+import org.apache.qpid.disttest.client.property.PropertyValue;
+import org.apache.qpid.disttest.client.property.SimplePropertyValue;
+import org.apache.qpid.disttest.controller.CommandForClient;
+import org.apache.qpid.disttest.message.Command;
+import org.apache.qpid.disttest.message.CreateMessageProviderCommand;
+import org.apache.qpid.disttest.message.NoOpCommand;
+
+public class ClientConfigTest extends TestCase
+{
+    private static final String CLIENT1 = "client1";
+
+    public void testClientConfigHasZeroArgConstructorForGson()
+    {
+        ClientConfig c = new ClientConfig();
+        assertNotNull(c);
+    }
+
+    public void testCreateCommands()
+    {
+        ClientConfig clientConfig = createClientConfigWithConnectionConfigReturningChildCommands();
+
+        List<CommandForClient> commands = clientConfig.createCommands();
+        assertEquals(2, commands.size());
+
+        assertCommandForClient(commands, 0, CLIENT1, NoOpCommand.class);
+        assertCommandForClient(commands, 1, CLIENT1, NoOpCommand.class);
+    }
+
+    public void testCreateCommandsForMessageProvider()
+    {
+        ClientConfig clientConfig = createClientConfigWithMessageProviderConfigReturningCommands();
+
+        List<CommandForClient> commands = clientConfig.createCommands();
+        assertEquals(1, commands.size());
+
+        assertCommandForClient(commands, 0, CLIENT1, CreateMessageProviderCommand.class);
+    }
+
+    public void testGetTotalNumberOfParticipants()
+    {
+        ClientConfig clientConfig = createClientConfigWithTwoParticipants();
+        assertEquals(2, clientConfig.getTotalNumberOfParticipants());
+    }
+
+    private ClientConfig createClientConfigWithConnectionConfigReturningChildCommands()
+    {
+        ConnectionConfig connectionConfig = mock(ConnectionConfig.class);
+
+        List<Command> commands = Arrays.asList((Command)new NoOpCommand(), (Command)new NoOpCommand());
+        when(connectionConfig.createCommands()).thenReturn(commands);
+
+        return new ClientConfig(CLIENT1, connectionConfig);
+    }
+
+    private ClientConfig createClientConfigWithMessageProviderConfigReturningCommands()
+    {
+        Map<String, PropertyValue> messageProperties = new HashMap<String, PropertyValue>();
+        messageProperties.put("test", new SimplePropertyValue("testValue"));
+        MessageProviderConfig config = new MessageProviderConfig("test", messageProperties);
+
+        List<MessageProviderConfig> providerConfigs = new ArrayList<MessageProviderConfig>();
+        providerConfigs.add(config);
+
+        return new ClientConfig(CLIENT1, new ArrayList<ConnectionConfig>(), providerConfigs);
+    }
+
+    private ClientConfig createClientConfigWithTwoParticipants()
+    {
+        ConnectionConfig connectionConfig1 = mock(ConnectionConfig.class);
+        ConnectionConfig connectionConfig2 = mock(ConnectionConfig.class);
+
+        when(connectionConfig1.getTotalNumberOfParticipants()).thenReturn(1);
+        when(connectionConfig2.getTotalNumberOfParticipants()).thenReturn(1);
+
+        ClientConfig clientConfig = new ClientConfig(CLIENT1, connectionConfig1, connectionConfig2);
+        return clientConfig;
+    }
+}
diff --git a/qpid/java/perftests/src/test/java/org/apache/qpid/disttest/controller/config/ConfigReaderTest.java b/qpid/java/perftests/src/test/java/org/apache/qpid/disttest/controller/config/ConfigReaderTest.java
new file mode 100644
index 0000000000..af9ec28db0
--- /dev/null
+++ b/qpid/java/perftests/src/test/java/org/apache/qpid/disttest/controller/config/ConfigReaderTest.java
@@ -0,0 +1,113 @@
+/*
+ * 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.disttest.controller.config;
+
+import java.io.Reader;
+import java.util.List;
+import java.util.Map;
+
+import junit.framework.TestCase;
+
+import org.apache.qpid.disttest.ConfigFileHelper;
+import org.apache.qpid.disttest.client.MessageProvider;
+import org.apache.qpid.disttest.client.property.PropertyValue;
+import org.apache.qpid.disttest.controller.CommandForClient;
+
+public class ConfigReaderTest extends TestCase
+{
+    private Config _config;
+
+    @Override
+    protected void setUp()
+    {
+        ConfigReader configReader = new ConfigReader();
+        Reader reader = ConfigFileHelper.getConfigFileReader(getClass(), "sampleConfig.json");
+        _config = configReader.readConfig(reader);
+    }
+
+    public void testReadTest()
+    {
+        List<TestConfig> tests = _config.getTestConfigs();
+        assertEquals("Unexpected number of tests", 2, tests.size());
+        TestConfig test1Config = tests.get(0);
+        assertNotNull("Test 1 configuration is expected", test1Config);
+        assertEquals("Unexpected test name", "Test 1", test1Config.getName());
+
+        TestConfig test2Config = tests.get(1);
+        assertNotNull("Test 2 configuration is expected", test2Config);
+    }
+
+    public void testReadsTestWithQueues()
+    {
+        TestConfig test1Config =  _config.getTestConfigs().get(0);
+        List<QueueConfig> queues = test1Config.getQueues();
+        assertEquals("Unexpected number of queues", 2, queues.size());
+        QueueConfig queue1Config = queues.get(0);
+        assertNotNull("Expected queue 1 config", queue1Config);
+        assertEquals("Unexpected queue name", "Json-Queue-Name", queue1Config.getName());
+        assertTrue("Unexpected attributes", queue1Config.getAttributes().isEmpty());
+        assertFalse("Unexpected durable", queue1Config.isDurable());
+
+        QueueConfig queue2Config = queues.get(1);
+        assertNotNull("Expected queue 2 config", queue2Config);
+        assertEquals("Unexpected queue name", "Json Queue Name 2", queue2Config.getName());
+        assertTrue("Unexpected durable", queue2Config.isDurable());
+        Map<String, Object> attributes =  queue2Config.getAttributes();
+        assertNotNull("Expected attributes", attributes);
+        assertFalse("Attributes are not loaded", attributes.isEmpty());
+        assertEquals("Unexpected number of attributes", 1, attributes.size());
+        assertEquals("Unexpected attribute 'x-qpid-priorities' value", 10,
+                ((Number)attributes.get("x-qpid-priorities")).intValue());
+    }
+
+    public void testReadsTestWithIterations()
+    {
+        TestConfig testConfig = _config.getTestConfigs().get(0);
+        List<IterationValue> iterationValues = testConfig.getIterationValues();
+        assertEquals("Unexpected number of iterations", 2, iterationValues.size());
+
+        IterationValue iteration1 = iterationValues.get(0);
+
+        String messageSizeProperty = "_messageSize";
+
+        assertEquals("Unexpected value for property " + messageSizeProperty,
+                "100",
+                iteration1.getIterationPropertyValuesWithUnderscores().get(messageSizeProperty));
+    }
+
+    public void testReadsMessageProviders()
+    {
+        TestConfig testConfig = _config.getTestConfigs().get(0);
+        ClientConfig cleintConfig = testConfig.getClients().get(0);
+        List<MessageProviderConfig> configs = cleintConfig.getMessageProviders();
+        assertNotNull("Message provider configs should not be null", configs);
+        assertEquals("Unexpected number of message providers", 1, configs.size());
+        MessageProviderConfig messageProvider = configs.get(0);
+        assertNotNull("Message provider config should not be null", messageProvider);
+        assertEquals("Unexpected provider name", "testProvider1", messageProvider.getName());
+        Map<String, PropertyValue> properties = messageProvider.getMessageProperties();
+        assertNotNull("Message properties should not be null", properties);
+        assertEquals("Unexpected number of message properties", 3, properties.size());
+        assertNotNull("test property is not found", properties.get("test"));
+        assertNotNull("priority property is not found", properties.get("priority"));
+        assertNotNull("id property is not found", properties.get("id"));
+     }
+
+}
diff --git a/qpid/java/perftests/src/test/java/org/apache/qpid/disttest/controller/config/ConfigTest.java b/qpid/java/perftests/src/test/java/org/apache/qpid/disttest/controller/config/ConfigTest.java
new file mode 100644
index 0000000000..88750b9737
--- /dev/null
+++ b/qpid/java/perftests/src/test/java/org/apache/qpid/disttest/controller/config/ConfigTest.java
@@ -0,0 +1,56 @@
+/*
+ * 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.disttest.controller.config;
+
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+import java.util.Arrays;
+import java.util.List;
+
+import junit.framework.TestCase;
+
+public class ConfigTest extends TestCase
+{
+    public void testGetTestsForTestWithIteratingMessageSizes()
+    {
+        Config config = createConfigWithIteratingMessageSizes();
+        List<TestInstance> testConfigs = config.getTests();
+
+        assertEquals("should have a test config for each message size", 2, testConfigs.size());
+
+        TestInstance instance0 = testConfigs.get(0);
+        assertEquals(0, instance0.getIterationNumber());
+
+        TestInstance instance1 = testConfigs.get(1);
+        assertEquals(1, instance1.getIterationNumber());
+    }
+
+    private Config createConfigWithIteratingMessageSizes()
+    {
+        TestConfig testConfig = mock(TestConfig.class);
+
+        when(testConfig.getIterationValues()).thenReturn(Arrays.asList(new IterationValue(),new IterationValue()));
+
+        Config config = new Config(testConfig);
+
+        return config;
+    }
+
+}
diff --git a/qpid/java/perftests/src/test/java/org/apache/qpid/disttest/controller/config/ConfigTestUtils.java b/qpid/java/perftests/src/test/java/org/apache/qpid/disttest/controller/config/ConfigTestUtils.java
new file mode 100644
index 0000000000..ce5f92724f
--- /dev/null
+++ b/qpid/java/perftests/src/test/java/org/apache/qpid/disttest/controller/config/ConfigTestUtils.java
@@ -0,0 +1,56 @@
+/*
+ * 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.disttest.controller.config;
+
+import static junit.framework.Assert.assertEquals;
+import static junit.framework.Assert.assertTrue;
+
+import java.util.List;
+
+import org.apache.qpid.disttest.controller.CommandForClient;
+import org.apache.qpid.disttest.message.Command;
+
+public class ConfigTestUtils
+{
+    public static <C extends Command> void assertCommandForClient(final List<CommandForClient> commandsForClients, final int index, final String expectedRegisteredClientName, final Class<C> expectedCommandClass)
+    {
+        final CommandForClient commandForClient = commandsForClients.get(index);
+        assertEquals(expectedRegisteredClientName, commandForClient.getClientName());
+        final Command command = commandForClient.getCommand();
+        assertTrue("Command " + index + " is of class " + command.getClass() + " but expecting " + expectedCommandClass,
+                expectedCommandClass.isAssignableFrom(command.getClass()));
+    }
+
+    public static <C extends Command> void assertCommandEquals(final List<Command> commands, final int index, final Class<C> expectedCommandClass)
+    {
+        @SuppressWarnings("unchecked")
+        C command = (C) getCommand(commands, index); //explicit cast added to get round oracle compiler bug (id 6302954)
+        assertTrue("Command " + index + " is of class " + command.getClass() + " but expecting " + expectedCommandClass,
+                expectedCommandClass.isAssignableFrom(command.getClass()));
+    }
+
+    public static <C extends Command>  C getCommand(final List<Command> commands, final int index)
+    {
+        @SuppressWarnings("unchecked")
+        C command = (C) commands.get(index);
+        return command;
+    }
+
+}
diff --git a/qpid/java/perftests/src/test/java/org/apache/qpid/disttest/controller/config/ConnectionConfigTest.java b/qpid/java/perftests/src/test/java/org/apache/qpid/disttest/controller/config/ConnectionConfigTest.java
new file mode 100644
index 0000000000..7c839ed462
--- /dev/null
+++ b/qpid/java/perftests/src/test/java/org/apache/qpid/disttest/controller/config/ConnectionConfigTest.java
@@ -0,0 +1,94 @@
+/*
+ * 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.disttest.controller.config;
+
+import static org.apache.qpid.disttest.controller.config.ConfigTestUtils.assertCommandEquals;
+import static org.apache.qpid.disttest.controller.config.ConfigTestUtils.getCommand;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+import java.util.Arrays;
+import java.util.List;
+
+import junit.framework.TestCase;
+
+import org.apache.qpid.disttest.message.Command;
+import org.apache.qpid.disttest.message.CreateConnectionCommand;
+import org.apache.qpid.disttest.message.NoOpCommand;
+
+public class ConnectionConfigTest extends TestCase
+{
+    private static final String CONNECTION_FACTORY_NAME = "ConnectionFactoryName";
+    private static final String CONNECTION_NAME = "ConnectionName";
+
+    public void testConnectionConfigHasZeroArgConstructorForGson()
+    {
+        ConnectionConfig c = new ConnectionConfig();
+        assertNotNull(c);
+    }
+
+    public void testCreateCommandsForConnectionAndChildren()
+    {
+        ConnectionConfig connectionConfig = createConnectionConfigWithChildCommands();
+
+        List<Command> commands = connectionConfig.createCommands();
+        assertEquals(3, commands.size());
+
+        assertCommandEquals(commands, 0, CreateConnectionCommand.class);
+        assertCommandEquals(commands, 1, NoOpCommand.class);
+        assertCommandEquals(commands, 2, NoOpCommand.class);
+
+        CreateConnectionCommand createConnectionCommand = getCommand(commands, 0);
+        assertEquals(CONNECTION_NAME, createConnectionCommand.getConnectionName());
+        assertEquals(CONNECTION_FACTORY_NAME, createConnectionCommand.getConnectionFactoryName());
+    }
+
+    public void testGetTotalNumberOfParticipants()
+    {
+        ConnectionConfig connectionConfig = createConnectionConfigWithTwoParticipants();
+        assertEquals(2, connectionConfig.getTotalNumberOfParticipants());
+    }
+
+    private ConnectionConfig createConnectionConfigWithTwoParticipants()
+    {
+        SessionConfig sessionConfig1 = mock(SessionConfig.class);
+        SessionConfig sessionConfig2 = mock(SessionConfig.class);
+
+        when(sessionConfig1.getTotalNumberOfParticipants()).thenReturn(1);
+        when(sessionConfig2.getTotalNumberOfParticipants()).thenReturn(1);
+
+        ConnectionConfig connectionConfig = new ConnectionConfig(CONNECTION_NAME, CONNECTION_FACTORY_NAME, sessionConfig1, sessionConfig2);
+
+        return connectionConfig;
+    }
+
+    private ConnectionConfig createConnectionConfigWithChildCommands()
+    {
+        SessionConfig sessionConfig = mock(SessionConfig.class);
+
+        NoOpCommand cmd1 = mock(NoOpCommand.class);
+        NoOpCommand cmd2 = mock(NoOpCommand.class);
+        List<Command> commands = Arrays.asList((Command)cmd1, (Command)cmd2);
+        when(sessionConfig.createCommands(CONNECTION_NAME)).thenReturn(commands);
+
+        return new ConnectionConfig(CONNECTION_NAME, CONNECTION_FACTORY_NAME, sessionConfig);
+    }
+
+}
diff --git a/qpid/java/perftests/src/test/java/org/apache/qpid/disttest/controller/config/ConsumerConfigTest.java b/qpid/java/perftests/src/test/java/org/apache/qpid/disttest/controller/config/ConsumerConfigTest.java
new file mode 100644
index 0000000000..c011ff4711
--- /dev/null
+++ b/qpid/java/perftests/src/test/java/org/apache/qpid/disttest/controller/config/ConsumerConfigTest.java
@@ -0,0 +1,79 @@
+/*
+ * 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.disttest.controller.config;
+
+import junit.framework.TestCase;
+
+import org.apache.qpid.disttest.message.CreateConsumerCommand;
+
+public class ConsumerConfigTest extends TestCase
+{
+    public void testConsumerHasZeroArgConstructorForGson()
+    {
+        ConsumerConfig c = new ConsumerConfig();
+        assertNotNull(c);
+    }
+
+    public void testCreateConsumerCommand()
+    {
+        boolean isTopic = true;
+        boolean isDurableSubscription = true;
+        boolean isBrowsingSubscription = true;
+        boolean noLocal = true;
+        long numberOfMessages = 100;
+        String consumerName = "consumerName";
+        String sessionName = "sessionName";
+        String destinationName = "destinationName";
+        String selector = "selector";
+        int batchSize = 10;;
+        long maximumDuration = 50;
+        boolean isSynchronousNonDefault = false;
+
+        ConsumerConfig consumerConfig = new ConsumerConfig(
+            consumerName,
+            destinationName,
+            numberOfMessages,
+            batchSize,
+            maximumDuration,
+            isTopic,
+            isDurableSubscription,
+            isBrowsingSubscription,
+            selector,
+            noLocal,
+            isSynchronousNonDefault);
+
+        CreateConsumerCommand createConsumerCommand = consumerConfig.createCommand(sessionName);
+
+        assertEquals(sessionName, createConsumerCommand.getSessionName());
+        assertEquals(consumerName, createConsumerCommand.getParticipantName());
+        assertEquals(destinationName, createConsumerCommand.getDestinationName());
+        assertEquals(numberOfMessages, createConsumerCommand.getNumberOfMessages());
+        assertEquals(batchSize, createConsumerCommand.getBatchSize());
+        assertEquals(maximumDuration, createConsumerCommand.getMaximumDuration());
+
+        assertEquals(isTopic, createConsumerCommand.isTopic());
+        assertEquals(isDurableSubscription, createConsumerCommand.isDurableSubscription());
+        assertEquals(isBrowsingSubscription, createConsumerCommand.isBrowsingSubscription());
+        assertEquals(selector, createConsumerCommand.getSelector());
+        assertEquals(noLocal, createConsumerCommand.isNoLocal());
+        assertEquals(isSynchronousNonDefault, createConsumerCommand.isSynchronous());
+    }
+
+}
diff --git a/qpid/java/perftests/src/test/java/org/apache/qpid/disttest/controller/config/IterationValueTest.java b/qpid/java/perftests/src/test/java/org/apache/qpid/disttest/controller/config/IterationValueTest.java
new file mode 100644
index 0000000000..7998eae37e
--- /dev/null
+++ b/qpid/java/perftests/src/test/java/org/apache/qpid/disttest/controller/config/IterationValueTest.java
@@ -0,0 +1,78 @@
+/*
+ * 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.disttest.controller.config;
+
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.verifyZeroInteractions;
+
+import java.util.Collections;
+import java.util.Map;
+
+import junit.framework.TestCase;
+
+import org.apache.qpid.disttest.message.CreateConsumerCommand;
+import org.apache.qpid.disttest.message.CreateProducerCommand;
+
+public class IterationValueTest extends TestCase
+{
+    private static final int MESSAGE_SIZE = 10;
+
+    private CreateProducerCommand _createProducerCommand;
+    private CreateConsumerCommand _createConsumerCommand;
+    private Map<String, String> _iterationValueMap;
+
+    @Override
+    protected void setUp() throws Exception
+    {
+        super.setUp();
+        _createProducerCommand = mock(CreateProducerCommand.class);
+        _createConsumerCommand = mock(CreateConsumerCommand.class);
+
+        _iterationValueMap = Collections.singletonMap("_messageSize", String.valueOf(MESSAGE_SIZE));
+    }
+
+    public void testApplyPopulatedIterationValueToCommandWithMatchingProperties() throws Exception
+    {
+        IterationValue iterationValue = new IterationValue(_iterationValueMap);
+
+        iterationValue.applyToCommand(_createProducerCommand);
+
+        verify(_createProducerCommand).setMessageSize(MESSAGE_SIZE);
+    }
+
+    public void testApplyPopulatedIterationValueToCommandWithoutMatchingProperties() throws Exception
+    {
+        IterationValue iterationValue = new IterationValue(_iterationValueMap);
+
+        iterationValue.applyToCommand(_createConsumerCommand);
+
+        verifyZeroInteractions(_createConsumerCommand);
+    }
+
+    public void testApplyUnpopulatedIterationValueToCommand() throws Exception
+    {
+        IterationValue iterationValue = new IterationValue();
+
+        iterationValue.applyToCommand(_createProducerCommand);
+
+        verifyZeroInteractions(_createProducerCommand);
+    }
+
+}
diff --git a/qpid/java/perftests/src/test/java/org/apache/qpid/disttest/controller/config/MessageProviderConfigTest.java b/qpid/java/perftests/src/test/java/org/apache/qpid/disttest/controller/config/MessageProviderConfigTest.java
new file mode 100644
index 0000000000..a3b367a4b4
--- /dev/null
+++ b/qpid/java/perftests/src/test/java/org/apache/qpid/disttest/controller/config/MessageProviderConfigTest.java
@@ -0,0 +1,51 @@
+/*
+ * 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.disttest.controller.config;
+
+import java.util.HashMap;
+import java.util.Map;
+
+import junit.framework.TestCase;
+
+import org.apache.qpid.disttest.client.property.PropertyValue;
+import org.apache.qpid.disttest.client.property.SimplePropertyValue;
+import org.apache.qpid.disttest.message.CreateMessageProviderCommand;
+
+public class MessageProviderConfigTest extends TestCase
+{
+    public void testCreateCommandsForMessageProvider()
+    {
+        Map<String, PropertyValue> messageProperties = new HashMap<String, PropertyValue>();
+        messageProperties.put("test", new SimplePropertyValue("testValue"));
+        MessageProviderConfig config = new MessageProviderConfig("test", messageProperties);
+        CreateMessageProviderCommand command = config.createCommand();
+        assertNotNull("Command should not be null", command);
+        assertNotNull("Unexpected name", command.getProviderName());
+        assertEquals("Unexpected properties", messageProperties, command.getMessageProperties());
+    }
+
+    public void testMessageProviderConfig()
+    {
+        Map<String, PropertyValue> messageProperties = new HashMap<String, PropertyValue>();
+        messageProperties.put("test", new SimplePropertyValue("testValue"));
+        MessageProviderConfig config = new MessageProviderConfig("test", messageProperties);
+        assertEquals("Unexpected name", "test", config.getName());
+        assertEquals("Unexpected properties", messageProperties, config.getMessageProperties());
+    }
+}
diff --git a/qpid/java/perftests/src/test/java/org/apache/qpid/disttest/controller/config/ProducerConfigTest.java b/qpid/java/perftests/src/test/java/org/apache/qpid/disttest/controller/config/ProducerConfigTest.java
new file mode 100644
index 0000000000..b9e591f113
--- /dev/null
+++ b/qpid/java/perftests/src/test/java/org/apache/qpid/disttest/controller/config/ProducerConfigTest.java
@@ -0,0 +1,89 @@
+/*
+ * 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.disttest.controller.config;
+
+import javax.jms.DeliveryMode;
+import javax.jms.Message;
+
+import junit.framework.TestCase;
+
+import org.apache.qpid.disttest.message.CreateProducerCommand;
+
+public class ProducerConfigTest extends TestCase
+{
+    public void testProducerHasZeroArgConstructorForGson()
+    {
+        ProducerConfig p = new ProducerConfig();
+        assertNotNull(p);
+    }
+
+    public void testConfigProvidesJmsDefaults()
+    {
+        CreateProducerCommand p = new ProducerConfig().createCommand("session1");
+        assertEquals(Message.DEFAULT_DELIVERY_MODE, p.getDeliveryMode());
+        assertEquals(Message.DEFAULT_PRIORITY, p.getPriority());
+        assertEquals(Message.DEFAULT_TIME_TO_LIVE, p.getTimeToLive());
+    }
+
+    public void testCreateProducerCommand()
+    {
+        String destination = "url:/destination";
+        int messageSize = 1000;
+        int numberOfMessages = 10;
+        int priority = 4;
+        long timeToLive = 10000;
+        int batchSize = 5;
+        long interval = 60;
+        long maximumDuration = 70;
+        long startDelay = 80;
+        String providerName = "testProvider1";
+
+        ProducerConfig producerConfig = new ProducerConfig(
+                "producer1",
+                destination,
+                numberOfMessages,
+                batchSize,
+                maximumDuration,
+                DeliveryMode.NON_PERSISTENT,
+                messageSize,
+                priority,
+                timeToLive,
+                interval,
+                startDelay,
+                providerName);
+
+        CreateProducerCommand command = producerConfig.createCommand("session1");
+
+        assertEquals("session1", command.getSessionName());
+        assertEquals("producer1", command.getParticipantName());
+        assertEquals(destination, command.getDestinationName());
+        assertEquals(numberOfMessages, command.getNumberOfMessages());
+        assertEquals(batchSize, command.getBatchSize());
+        assertEquals(maximumDuration, command.getMaximumDuration());
+
+        assertEquals(DeliveryMode.NON_PERSISTENT, command.getDeliveryMode());
+        assertEquals(messageSize, command.getMessageSize());
+        assertEquals(priority, command.getPriority());
+        assertEquals(timeToLive, command.getTimeToLive());
+        assertEquals(interval, command.getInterval());
+        assertEquals(startDelay, command.getStartDelay());
+        assertEquals(providerName, command.getMessageProviderName());
+    }
+}
diff --git a/qpid/java/perftests/src/test/java/org/apache/qpid/disttest/controller/config/SessionConfigTest.java b/qpid/java/perftests/src/test/java/org/apache/qpid/disttest/controller/config/SessionConfigTest.java
new file mode 100644
index 0000000000..8775e4064d
--- /dev/null
+++ b/qpid/java/perftests/src/test/java/org/apache/qpid/disttest/controller/config/SessionConfigTest.java
@@ -0,0 +1,93 @@
+/*
+ * 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.disttest.controller.config;
+
+import static org.apache.qpid.disttest.controller.config.ConfigTestUtils.assertCommandEquals;
+import static org.apache.qpid.disttest.controller.config.ConfigTestUtils.getCommand;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+import java.util.Collections;
+import java.util.List;
+
+import javax.jms.Session;
+
+import junit.framework.TestCase;
+
+import org.apache.qpid.disttest.message.Command;
+import org.apache.qpid.disttest.message.CreateConsumerCommand;
+import org.apache.qpid.disttest.message.CreateProducerCommand;
+import org.apache.qpid.disttest.message.CreateSessionCommand;
+
+public class SessionConfigTest extends TestCase
+{
+    private static final String CONNECTION_NAME = "conn1";
+    private static final String SESSION = "session1";
+
+    public void testSessionHasZeroArgConstructorForGson()
+    {
+        SessionConfig s = new SessionConfig();
+        assertNotNull(s);
+    }
+
+    public void testCreateCommandsForSessionAndChildren()
+    {
+        SessionConfig sessionConfig = createSessionConfigWithChildCommands();
+
+        List<Command> commands = sessionConfig.createCommands(CONNECTION_NAME);
+        assertEquals(3, commands.size());
+
+        assertCommandEquals(commands, 0, CreateSessionCommand.class);
+        assertCommandEquals(commands, 1, CreateProducerCommand.class);
+        assertCommandEquals(commands, 2, CreateConsumerCommand.class);
+
+        CreateSessionCommand createSessionCommand = getCommand(commands, 0);
+        assertEquals(Session.AUTO_ACKNOWLEDGE, createSessionCommand.getAcknowledgeMode());
+        assertEquals(SESSION, createSessionCommand.getSessionName());
+        assertEquals(CONNECTION_NAME, createSessionCommand.getConnectionName());
+    }
+
+    public void testGetTotalNumberOfParticipants()
+    {
+        SessionConfig sessionConfig = createSessionConfigWithOneConsumerAndOneProducer();
+        assertEquals(2, sessionConfig.getTotalNumberOfParticipants());
+    }
+
+    private SessionConfig createSessionConfigWithOneConsumerAndOneProducer()
+    {
+        return createSessionConfigWithChildCommands();
+    }
+
+    private SessionConfig createSessionConfigWithChildCommands()
+    {
+        ProducerConfig producerConfig = mock(ProducerConfig.class);
+        ConsumerConfig consumerConfig = mock(ConsumerConfig.class);
+
+        when(producerConfig.createCommand(SESSION)).thenReturn(mock(CreateProducerCommand.class));
+        when(consumerConfig.createCommand(SESSION)).thenReturn(mock(CreateConsumerCommand.class));
+
+        return new SessionConfig(SESSION,
+                                 Session.AUTO_ACKNOWLEDGE,
+                                 Collections.singletonList(consumerConfig),
+                                 Collections.singletonList(producerConfig));
+    }
+
+
+}
diff --git a/qpid/java/perftests/src/test/java/org/apache/qpid/disttest/controller/config/TestConfigTest.java b/qpid/java/perftests/src/test/java/org/apache/qpid/disttest/controller/config/TestConfigTest.java
new file mode 100644
index 0000000000..1212a57606
--- /dev/null
+++ b/qpid/java/perftests/src/test/java/org/apache/qpid/disttest/controller/config/TestConfigTest.java
@@ -0,0 +1,113 @@
+/*
+ * 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.disttest.controller.config;
+
+import static org.apache.qpid.disttest.controller.config.ConfigTestUtils.assertCommandForClient;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+import java.util.ArrayList;
+import java.util.List;
+
+import junit.framework.TestCase;
+
+import org.apache.qpid.disttest.controller.CommandForClient;
+import org.apache.qpid.disttest.message.NoOpCommand;
+
+public class TestConfigTest extends TestCase
+{
+    private static final QueueConfig[] EMPTY_QUEUES_ARRAY = new QueueConfig[0];
+    private static final String CLIENT1 = "client1";
+    private static final String CLIENT2 = "client2";
+    private static final String TEST1 = "test1";
+
+    public void testConfigHasZeroArgConstructorForGson()
+    {
+        TestConfig c = new TestConfig();
+        assertNotNull(c);
+    }
+
+    public void testCreateCommandsForClient()
+    {
+        TestConfig config = createTestConfigWithClientConfigReturningChildCommands();
+
+        List<CommandForClient> commandsForClients = config.createCommands();
+        assertEquals("Unexpected number of commands for client", 3, commandsForClients.size());
+
+        assertCommandForClient(commandsForClients, 0, CLIENT1, NoOpCommand.class);
+        assertCommandForClient(commandsForClients, 1, CLIENT1, NoOpCommand.class);
+        assertCommandForClient(commandsForClients, 2, CLIENT2, NoOpCommand.class);
+    }
+
+    public void testGetClientNames()
+    {
+        TestConfig config = createTestConfigWithTwoClients();
+
+        assertEquals(2, config.getClientNames().size());
+    }
+
+    public void testGetTotalNumberOfClients()
+    {
+        TestConfig config = createTestConfigWithTwoClients();
+        assertEquals(2, config.getTotalNumberOfClients());
+    }
+
+    public void testGetTotalNumberOfParticipants()
+    {
+        TestConfig config = createTestConfigWithTwoClients();
+        assertEquals(2, config.getTotalNumberOfParticipants());
+    }
+
+    private TestConfig createTestConfigWithClientConfigReturningChildCommands()
+    {
+        ClientConfig clientConfig1 = createClientConfigReturningCommands(CLIENT1, 2);
+        ClientConfig clientConfig2 = createClientConfigReturningCommands(CLIENT2, 1);
+
+        TestConfig config = new TestConfig(TEST1, new ClientConfig[] { clientConfig1, clientConfig2 }, EMPTY_QUEUES_ARRAY);
+        return config;
+    }
+
+    private ClientConfig createClientConfigReturningCommands(final String clientName, int numberOfCommands)
+    {
+        ClientConfig clientConfig = mock(ClientConfig.class);
+
+        List<CommandForClient> commandList = new ArrayList<CommandForClient>();
+
+        for (int i = 1 ; i <= numberOfCommands; i++)
+        {
+            commandList.add(new CommandForClient(clientName, new NoOpCommand()));
+        }
+
+        when(clientConfig.createCommands()).thenReturn(commandList);
+        return clientConfig;
+    }
+
+    private TestConfig createTestConfigWithTwoClients()
+    {
+        ClientConfig clientConfig1 = mock(ClientConfig.class);
+        ClientConfig clientConfig2 = mock(ClientConfig.class);
+
+        when(clientConfig1.getTotalNumberOfParticipants()).thenReturn(1);
+        when(clientConfig2.getTotalNumberOfParticipants()).thenReturn(1);
+
+        TestConfig config = new TestConfig(TEST1, new ClientConfig[] { clientConfig1, clientConfig2 }, EMPTY_QUEUES_ARRAY);
+        return config;
+    }
+}
diff --git a/qpid/java/perftests/src/test/java/org/apache/qpid/disttest/controller/config/TestInstanceTest.java b/qpid/java/perftests/src/test/java/org/apache/qpid/disttest/controller/config/TestInstanceTest.java
new file mode 100644
index 0000000000..928fbe58cf
--- /dev/null
+++ b/qpid/java/perftests/src/test/java/org/apache/qpid/disttest/controller/config/TestInstanceTest.java
@@ -0,0 +1,102 @@
+/*
+ * 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.disttest.controller.config;
+
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+
+import junit.framework.TestCase;
+
+import org.apache.qpid.disttest.controller.CommandForClient;
+import org.apache.qpid.disttest.message.CreateConsumerCommand;
+import org.apache.qpid.disttest.message.CreateProducerCommand;
+import org.apache.qpid.disttest.message.NoOpCommand;
+
+public class TestInstanceTest extends TestCase
+{
+    private static final String CLIENT_NAME = "CLIENT_NAME";
+    private static final int ITERATION_NUMBER = 0;
+
+    private NoOpCommand _noOpCommand;
+    private CreateProducerCommand _createProducerCommand;
+    private CreateConsumerCommand _createConsumerCommand;
+
+    @Override
+    protected void setUp() throws Exception
+    {
+        super.setUp();
+        _noOpCommand = mock(NoOpCommand.class);
+        _createProducerCommand = mock(CreateProducerCommand.class);
+        _createConsumerCommand = mock(CreateConsumerCommand.class);
+    }
+
+    public void testCreateCommandsWithIterationValues()
+    {
+        IterationValue iterationValue = mock(IterationValue.class);
+
+        TestConfig config = createTestConfig();
+
+        TestInstance testInstance = new TestInstance(config, ITERATION_NUMBER, iterationValue);
+
+        List<CommandForClient> commandsForClients = testInstance.createCommands();
+        assertEquals("Unexpected number of commands for client", 3, commandsForClients.size());
+
+        verify(iterationValue).applyToCommand(_noOpCommand);
+        verify(iterationValue).applyToCommand(_createProducerCommand);
+        verify(iterationValue).applyToCommand(_createConsumerCommand);
+    }
+
+    public void testCreateCommandsWithoutIterationValues()
+    {
+        TestConfig config = createTestConfig();
+        TestInstance testInstance = new TestInstance(config);
+
+        List<CommandForClient> commandsForClients = testInstance.createCommands();
+        assertEquals("Unexpected number of commands for client", 3, commandsForClients.size());
+    }
+
+    public void testGetConfiguredClientNames()
+    {
+        TestConfig testConfig = mock(TestConfig.class);
+        when(testConfig.getClientNames()).thenReturn(Collections.singletonList(CLIENT_NAME));
+        TestInstance testInstance = new TestInstance(testConfig);
+
+        List<String> clientNames = testInstance.getClientNames();
+        assertEquals(1, clientNames.size());
+        assertEquals(CLIENT_NAME, clientNames.get(0));
+    }
+
+    private TestConfig createTestConfig()
+    {
+        CommandForClient commandForClient1 = new CommandForClient(CLIENT_NAME, _noOpCommand);
+        CommandForClient commandForClient2 = new CommandForClient(CLIENT_NAME, _createProducerCommand);
+        CommandForClient commandForClient3 = new CommandForClient(CLIENT_NAME, _createConsumerCommand);
+
+        TestConfig config = mock(TestConfig.class);
+        when(config.createCommands()).thenReturn(Arrays.asList(commandForClient1, commandForClient2, commandForClient3));
+
+        return config;
+    }
+
+}
diff --git a/qpid/java/perftests/src/test/java/org/apache/qpid/disttest/controller/config/sampleConfig.json b/qpid/java/perftests/src/test/java/org/apache/qpid/disttest/controller/config/sampleConfig.json
new file mode 100644
index 0000000000..9e1168129b
--- /dev/null
+++ b/qpid/java/perftests/src/test/java/org/apache/qpid/disttest/controller/config/sampleConfig.json
@@ -0,0 +1,72 @@
+{
+  "_tests": [
+    {
+      "_name": "Test 1",
+      "_queues": [
+        {
+          "_name": "Json-Queue-Name",
+          "_durable": false,
+          "_attributes": {}
+        },
+        {
+          "_name": "Json Queue Name 2",
+          "_durable": true,
+          "_attributes": {
+            "x-qpid-priorities": 10.0
+          }
+        }
+      ],
+      "_iterations": [
+        {
+          "_messageSize": 100,
+          "_numberOfMessages": 10
+        },
+        {
+          "_messageSize": 200,
+          "_numberOfMessages": 5
+        }
+      ],
+      "_clients": [
+        {
+          "_connections": [
+            {
+              "_name": "connection1",
+              "_sessions": []
+            }
+          ];
+          "_messageProviders":[
+            {
+              "_name": "testProvider1";
+              "_messageProperties": {
+                 "priority": {"@def": "list"; "_items": [1,2,3,4,4]};
+                "id": {"@def": "random"; "_upper": 10};
+                "test": "test-value"
+              }
+            }
+          ]
+        }
+      ]
+    },
+    {
+      "_name": "Test 2",
+      "_queues": [
+        {
+          "_name": "Json-Queue-Name",
+          "_durable": false,
+          "_attributes": {}
+        }
+      ],
+      "_iterations": [],
+      "_clients": [
+        {
+          "_connections": [
+            {
+              "_name": "connection1",
+              "_sessions": []
+            }
+          ]
+        }
+      ]
+    }
+  ]
+}
\ No newline at end of file
diff --git a/qpid/java/perftests/src/test/java/org/apache/qpid/disttest/jms/JmsMessageAdaptorTest.java b/qpid/java/perftests/src/test/java/org/apache/qpid/disttest/jms/JmsMessageAdaptorTest.java
new file mode 100644
index 0000000000..ab0f52263b
--- /dev/null
+++ b/qpid/java/perftests/src/test/java/org/apache/qpid/disttest/jms/JmsMessageAdaptorTest.java
@@ -0,0 +1,40 @@
+/*
+ * 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.disttest.jms;
+
+import junit.framework.TestCase;
+
+import org.apache.qpid.disttest.message.Command;
+import org.apache.qpid.disttest.message.CommandType;
+
+public class JmsMessageAdaptorTest extends TestCase
+{
+
+    public void testCheckAllCommandTypes()
+    {
+        for (CommandType commandType : CommandType.values())
+        {
+            Class<? extends Command> clazz = JmsMessageAdaptor.getCommandClassFromType(commandType);
+            assertNotNull(clazz);
+
+        }
+
+    }
+
+}
diff --git a/qpid/java/perftests/src/test/java/org/apache/qpid/disttest/message/JsonHandlerTest.java b/qpid/java/perftests/src/test/java/org/apache/qpid/disttest/message/JsonHandlerTest.java
new file mode 100644
index 0000000000..4a56fff8fe
--- /dev/null
+++ b/qpid/java/perftests/src/test/java/org/apache/qpid/disttest/message/JsonHandlerTest.java
@@ -0,0 +1,183 @@
+/*
+ * 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.disttest.message;
+
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+
+import junit.framework.TestCase;
+
+import org.apache.commons.lang.builder.EqualsBuilder;
+import org.apache.qpid.disttest.client.property.ListPropertyValue;
+import org.apache.qpid.disttest.client.property.PropertyValue;
+import org.apache.qpid.disttest.json.JsonHandler;
+
+public class JsonHandlerTest extends TestCase
+{
+    private JsonHandler _jsonHandler = null;
+    private SendChristmasCards _testCommand = null;
+
+    @Override
+    protected void setUp() throws Exception
+    {
+        super.setUp();
+
+        _jsonHandler = new JsonHandler();
+
+        _testCommand = new SendChristmasCards(CommandType.START_TEST, Collections.singletonMap(SendChristmasCards.CardType.FUNNY, 5));
+        _testCommand.persons = Arrays.asList(new Person("Phil"), new Person("Andrew"));
+    }
+
+    public void testMarshallUnmarshall() throws Exception
+    {
+        final String jsonString = _jsonHandler.marshall(_testCommand);
+
+        final SendChristmasCards unmarshalledCommand = _jsonHandler.unmarshall(jsonString, SendChristmasCards.class);
+
+        assertEquals("Unmarshalled command should be equal to the original object", _testCommand, unmarshalledCommand);
+    }
+
+    public void testGeneratorDesrialization()
+    {
+        String json = "{_messageProperties: {test: 1; generator: {'@def': 'list'; _cyclic: false; _items: ['first', " +
+                "{'@def': 'range'; _upper:10; '_type':'int'}]}}}";
+        final TestCommand unmarshalledCommand = _jsonHandler.unmarshall(json, TestCommand.class);
+        Map<String, PropertyValue> properties = unmarshalledCommand.getMessageProperties();
+        assertNotNull("Properties should not be null", properties);
+        assertFalse("Properties should not be empty", properties.isEmpty());
+        assertEquals("Unexpected properties size", 2, properties.size());
+        PropertyValue testProperty = properties.get("test");
+        assertNotNull("Unexpected property test", testProperty);
+        assertTrue("Unexpected property test", testProperty.getValue() instanceof Number);
+        assertEquals("Unexpected property value", 1, ((Number)testProperty.getValue()).intValue());
+        Object generatorObject = properties.get("generator");
+        assertTrue("Unexpected generator object", generatorObject instanceof ListPropertyValue);
+        PropertyValue generator = (PropertyValue)generatorObject;
+        assertEquals("Unexpected generator value", "first", generator.getValue());
+        for (int i = 0; i < 10; i++)
+        {
+            assertEquals("Unexpected generator value", new Integer(i), generator.getValue());
+        }
+        String newJson =_jsonHandler.marshall(unmarshalledCommand);
+        final TestCommand newUnmarshalledCommand = _jsonHandler.unmarshall(newJson, TestCommand.class);
+        assertEquals("Unmarshalled command should be equal to the original object", unmarshalledCommand, newUnmarshalledCommand);
+    }
+
+    /**
+     * A {@link Command} designed to exercise {@link JsonHandler}, e.g does it handle a map of enums?.
+     *
+     * This class is non-private to avoid auto-deletion of "unused" fields/methods
+     */
+    static class SendChristmasCards extends Command
+    {
+        enum CardType {FUNNY, TRADITIONAL}
+
+        private Map<CardType, Integer> _cardTypes;
+        private List<Person> persons;
+
+        public SendChristmasCards(final CommandType type, Map<CardType, Integer> cardTypes)
+        {
+            super(type);
+            _cardTypes = cardTypes;
+        }
+
+        public Map<CardType, Integer> getCardTypes()
+        {
+            return _cardTypes;
+        }
+
+        public List<Person> getPersons()
+        {
+            return persons;
+        }
+
+        @Override
+        public boolean equals(final Object obj)
+        {
+            return EqualsBuilder.reflectionEquals(this, obj);
+        }
+    }
+
+    /**
+     * This class is non-private to avoid auto-deletion of "unused" fields/methods
+     */
+    static class Person
+    {
+        private String _firstName;
+
+        public Person(final String firstName)
+        {
+            _firstName = firstName;
+        }
+
+        public String getFirstName()
+        {
+            return _firstName;
+        }
+
+        @Override
+        public boolean equals(final Object obj)
+        {
+            return EqualsBuilder.reflectionEquals(this, obj);
+        }
+
+    }
+
+    /**
+     * Yet another test class
+     */
+    static class TestCommand extends Command
+    {
+
+        private Map<String, PropertyValue> _messageProperties;
+
+        public TestCommand(CommandType type)
+        {
+            super(type);
+        }
+
+        public Map<String, PropertyValue> getMessageProperties()
+        {
+            return _messageProperties;
+        }
+
+        public void setMessageProperties(Map<String, PropertyValue> _messageProperties)
+        {
+            this._messageProperties = _messageProperties;
+        }
+
+        @Override
+        public boolean equals(final Object obj)
+        {
+            if (obj == null || !(obj instanceof TestCommand))
+            {
+                return false;
+            }
+            TestCommand other = (TestCommand)obj;
+            if (_messageProperties == null && other._messageProperties != null )
+            {
+                return false;
+            }
+            return _messageProperties.equals(other._messageProperties);
+        }
+    }
+}
diff --git a/qpid/java/perftests/src/test/java/org/apache/qpid/disttest/message/ParticipantResultTest.java b/qpid/java/perftests/src/test/java/org/apache/qpid/disttest/message/ParticipantResultTest.java
new file mode 100644
index 0000000000..ee095ce199
--- /dev/null
+++ b/qpid/java/perftests/src/test/java/org/apache/qpid/disttest/message/ParticipantResultTest.java
@@ -0,0 +1,148 @@
+/*
+ * 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.disttest.message;
+
+import static org.apache.qpid.disttest.message.ParticipantAttribute.BATCH_SIZE;
+import static org.apache.qpid.disttest.message.ParticipantAttribute.CONFIGURED_CLIENT_NAME;
+import static org.apache.qpid.disttest.message.ParticipantAttribute.DELIVERY_MODE;
+import static org.apache.qpid.disttest.message.ParticipantAttribute.ERROR_MESSAGE;
+import static org.apache.qpid.disttest.message.ParticipantAttribute.IS_BROWSIING_SUBSCRIPTION;
+import static org.apache.qpid.disttest.message.ParticipantAttribute.IS_DURABLE_SUBSCRIPTION;
+import static org.apache.qpid.disttest.message.ParticipantAttribute.IS_NO_LOCAL;
+import static org.apache.qpid.disttest.message.ParticipantAttribute.IS_SELECTOR;
+import static org.apache.qpid.disttest.message.ParticipantAttribute.IS_SYNCHRONOUS_CONSUMER;
+import static org.apache.qpid.disttest.message.ParticipantAttribute.IS_TOPIC;
+import static org.apache.qpid.disttest.message.ParticipantAttribute.ITERATION_NUMBER;
+import static org.apache.qpid.disttest.message.ParticipantAttribute.MAXIMUM_DURATION;
+import static org.apache.qpid.disttest.message.ParticipantAttribute.PAYLOAD_SIZE;
+import static org.apache.qpid.disttest.message.ParticipantAttribute.NUMBER_OF_MESSAGES_PROCESSED;
+import static org.apache.qpid.disttest.message.ParticipantAttribute.PARTICIPANT_NAME;
+import static org.apache.qpid.disttest.message.ParticipantAttribute.PRIORITY;
+import static org.apache.qpid.disttest.message.ParticipantAttribute.PRODUCER_INTERVAL;
+import static org.apache.qpid.disttest.message.ParticipantAttribute.PRODUCER_START_DELAY;
+import static org.apache.qpid.disttest.message.ParticipantAttribute.TEST_NAME;
+import static org.apache.qpid.disttest.message.ParticipantAttribute.TIME_TAKEN;
+import static org.apache.qpid.disttest.message.ParticipantAttribute.TIME_TO_LIVE;
+
+import java.util.Date;
+
+import javax.jms.DeliveryMode;
+
+import junit.framework.TestCase;
+
+public class ParticipantResultTest extends TestCase
+{
+
+    public void testSharedParticipantResultAttributes() throws Exception
+    {
+        final String participantName = "PARTICIPANT_NAME1";
+        final String testName = "TEST_NAME1";
+        String clientConfiguredName = "CLIENT_CONFIGURED_NAME";
+        String errorMessage = "errorMessage";
+        int iterationNumber = 1;
+
+        ParticipantResult result = new ParticipantResult();
+
+        long numberOfMessages = 500;
+        long timeTaken = 30;
+        int batchSize = 10;
+
+        long startTime = System.currentTimeMillis();
+        long endTime = startTime + timeTaken;
+        long maximumDuration = 1000;
+
+        result.setParticipantName(participantName);
+        result.setTestName(testName);
+        result.setIterationNumber(iterationNumber);
+        result.setConfiguredClientName(clientConfiguredName);
+
+        result.setNumberOfMessagesProcessed(numberOfMessages);
+        result.setConfiguredClientName(clientConfiguredName);
+        result.setBatchSize(10);
+
+        result.setStartDate(new Date(startTime));
+        result.setEndDate(new Date(endTime));
+        result.setMaximumDuration(maximumDuration);
+
+        result.setErrorMessage(errorMessage);
+
+        assertEquals(participantName,      result.getAttributes().get(PARTICIPANT_NAME));
+        assertEquals(testName,             result.getAttributes().get(TEST_NAME));
+        assertEquals(clientConfiguredName,   result.getAttributes().get(CONFIGURED_CLIENT_NAME));
+        assertEquals(numberOfMessages,       result.getAttributes().get(NUMBER_OF_MESSAGES_PROCESSED));
+        assertEquals(timeTaken,              result.getAttributes().get(TIME_TAKEN));
+        assertEquals(batchSize,              result.getAttributes().get(BATCH_SIZE));
+        assertEquals(maximumDuration,        result.getAttributes().get(MAXIMUM_DURATION));
+        assertEquals(errorMessage,           result.getAttributes().get(ERROR_MESSAGE));
+        assertEquals(iterationNumber,         result.getAttributes().get(ITERATION_NUMBER));
+    }
+
+    public void testConsumerParticipantResultAttributes() throws Exception
+    {
+        ConsumerParticipantResult result = new ConsumerParticipantResult();
+
+        boolean topic = true;
+        boolean durable = true;
+        boolean browsingSubscription = false;
+        boolean selector = true;
+        boolean noLocal = false;
+        boolean synchronousConsumer = true;
+
+        result.setTopic(topic);
+        result.setDurableSubscription(durable);
+        result.setBrowsingSubscription(browsingSubscription);
+        result.setSelector(selector);
+        result.setNoLocal(noLocal);
+        result.setSynchronousConsumer(synchronousConsumer);
+
+        assertEquals(topic,                  result.getAttributes().get(IS_TOPIC));
+        assertEquals(durable,                result.getAttributes().get(IS_DURABLE_SUBSCRIPTION));
+        assertEquals(browsingSubscription,   result.getAttributes().get(IS_BROWSIING_SUBSCRIPTION));
+        assertEquals(selector,               result.getAttributes().get(IS_SELECTOR));
+        assertEquals(noLocal,                result.getAttributes().get(IS_NO_LOCAL));
+        assertEquals(synchronousConsumer,    result.getAttributes().get(IS_SYNCHRONOUS_CONSUMER));
+    }
+
+    public void testProducerParticipantResultAttributes() throws Exception
+    {
+        ProducerParticipantResult result = new ProducerParticipantResult();
+
+        int priority = 2;
+        long timeToLive = 30;
+        long producerStartDelay = 40;
+        long producerInterval = 50;
+        int messageSize = 60;
+        int deliveryMode = DeliveryMode.PERSISTENT;
+
+        result.setPriority(priority);
+        result.setTimeToLive(timeToLive);
+        result.setStartDelay(producerStartDelay);
+        result.setInterval(producerInterval);
+        result.setPayloadSize(messageSize);
+        result.setDeliveryMode(deliveryMode);
+
+
+        assertEquals(priority,           result.getAttributes().get(PRIORITY));
+        assertEquals(timeToLive,         result.getAttributes().get(TIME_TO_LIVE));
+        assertEquals(producerStartDelay, result.getAttributes().get(PRODUCER_START_DELAY));
+        assertEquals(producerInterval,   result.getAttributes().get(PRODUCER_INTERVAL));
+        assertEquals(messageSize,        result.getAttributes().get(PAYLOAD_SIZE));
+        assertEquals(deliveryMode,       result.getAttributes().get(DELIVERY_MODE));
+    }
+}
diff --git a/qpid/java/perftests/src/test/java/org/apache/qpid/disttest/results/aggregation/AggregatorTest.java b/qpid/java/perftests/src/test/java/org/apache/qpid/disttest/results/aggregation/AggregatorTest.java
new file mode 100644
index 0000000000..393837b4d5
--- /dev/null
+++ b/qpid/java/perftests/src/test/java/org/apache/qpid/disttest/results/aggregation/AggregatorTest.java
@@ -0,0 +1,61 @@
+/*
+ * 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.disttest.results.aggregation;
+
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+import java.util.Arrays;
+
+import junit.framework.TestCase;
+
+import org.apache.qpid.disttest.controller.ResultsForAllTests;
+
+public class AggregatorTest extends TestCase
+{
+    private Aggregator _aggregator = new Aggregator();
+    private TestResultAggregator _testResultAggregator = mock(TestResultAggregator.class);
+
+    @Override
+    protected void setUp() throws Exception
+    {
+        super.setUp();
+        _aggregator.setTestResultAggregator(_testResultAggregator);
+    }
+
+    public void testAggregrateManyTestResults() throws Exception
+    {
+        ResultsForAllTests resultsForAllTests = mock(ResultsForAllTests.class);
+        ITestResult testResult1 = mock(ITestResult.class);
+        ITestResult testResult2 = mock(ITestResult.class);
+
+        when(resultsForAllTests.getTestResults()).thenReturn(Arrays.asList(testResult1, testResult2));
+        when(_testResultAggregator.aggregateTestResult(testResult1)).thenReturn(mock(AggregatedTestResult.class));
+        when(_testResultAggregator.aggregateTestResult(testResult2)).thenReturn(mock(AggregatedTestResult.class));
+
+        ResultsForAllTests aggregatedResultsForAllTests = _aggregator.aggregateResults(resultsForAllTests);
+        assertEquals(2, aggregatedResultsForAllTests.getTestResults().size());
+
+        verify(_testResultAggregator).aggregateTestResult(testResult1);
+        verify(_testResultAggregator).aggregateTestResult(testResult2);
+
+    }
+
+}
diff --git a/qpid/java/perftests/src/test/java/org/apache/qpid/disttest/results/aggregation/ParticipantResultAggregatorTest.java b/qpid/java/perftests/src/test/java/org/apache/qpid/disttest/results/aggregation/ParticipantResultAggregatorTest.java
new file mode 100644
index 0000000000..5cf84c77f1
--- /dev/null
+++ b/qpid/java/perftests/src/test/java/org/apache/qpid/disttest/results/aggregation/ParticipantResultAggregatorTest.java
@@ -0,0 +1,182 @@
+/*
+ * 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.disttest.results.aggregation;
+
+import java.util.Date;
+
+import org.apache.qpid.disttest.message.ParticipantResult;
+import org.apache.qpid.disttest.results.aggregation.ParticipantResultAggregator;
+
+import junit.framework.TestCase;
+
+public class ParticipantResultAggregatorTest extends TestCase
+{
+    private ParticipantResultAggregator _aggregator = new ParticipantResultAggregator(ParticipantResult.class, AGGREGATED_RESULT_NAME);
+
+    private static final String TEST_NAME = "TEST_NAME";
+    private static final String AGGREGATED_RESULT_NAME = "AGGREGATED_RESULT_NAME";
+    private static final int TEST_ITERATION_NUMBER = 1;
+
+    private static final long PARTICIPANT1_STARTDATE = 50;
+    private static final long PARTICIPANT1_ENDDATE = 20000;
+    private static final long PARTICIPANT1_TOTAL_PROCESSED = 1024;
+
+    private static final long PARTICIPANT2_STARTDATE = 100;
+    private static final long PARTICIPANT2_ENDDATE = 21000;
+    private static final long PARTICIPANT2_TOTAL_PROCESSED = 2048;
+
+    private static final long OVERALL_PROCESSED = PARTICIPANT1_TOTAL_PROCESSED + PARTICIPANT2_TOTAL_PROCESSED;
+    private static final double OVERALL_TIMETAKEN = PARTICIPANT2_ENDDATE - PARTICIPANT1_STARTDATE;
+
+    private static final double EXPECTED_AGGREGATED_ALL_THROUGHPUT = ((OVERALL_PROCESSED)/1024)/((OVERALL_TIMETAKEN)/1000);
+
+    public void testStartAndEndDateForOneParticipantResult()
+    {
+        ParticipantResult result = new ParticipantResult();
+        result.setStartDate(new Date(PARTICIPANT1_STARTDATE));
+        result.setEndDate(new Date(PARTICIPANT1_ENDDATE));
+
+        _aggregator.aggregate(result);
+        ParticipantResult aggregratedResult = _aggregator.getAggregatedResult();
+        assertEquals(PARTICIPANT1_STARTDATE, aggregratedResult.getStartInMillis());
+        assertEquals(PARTICIPANT1_ENDDATE, aggregratedResult.getEndInMillis());
+    }
+
+    public void testStartAndEndDateForTwoParticipantResults()
+    {
+        ParticipantResult result1 = new ParticipantResult();
+        result1.setStartDate(new Date(PARTICIPANT1_STARTDATE));
+        result1.setEndDate(new Date(PARTICIPANT1_ENDDATE));
+
+        ParticipantResult result2 = new ParticipantResult();
+        result2.setStartDate(new Date(PARTICIPANT2_STARTDATE));
+        result2.setEndDate(new Date(PARTICIPANT2_ENDDATE));
+
+        _aggregator.aggregate(result1);
+        _aggregator.aggregate(result2);
+
+        ParticipantResult aggregratedResult = _aggregator.getAggregatedResult();
+        assertEquals(PARTICIPANT1_STARTDATE, aggregratedResult.getStartInMillis());
+        assertEquals(PARTICIPANT2_ENDDATE, aggregratedResult.getEndInMillis());
+    }
+
+    public void testComputeNumberOfMessagesProcessed()
+    {
+        ParticipantResult result1 = new ParticipantResult();
+        result1.setNumberOfMessagesProcessed(10);
+
+        ParticipantResult result2 = new ParticipantResult();
+        result2.setNumberOfMessagesProcessed(15);
+
+        _aggregator.aggregate(result1);
+        _aggregator.aggregate(result2);
+
+        ParticipantResult aggregratedResult = _aggregator.getAggregatedResult();
+        assertEquals(25, aggregratedResult.getNumberOfMessagesProcessed());
+    }
+
+    public void testComputeTotalPayloadProcessed()
+    {
+        ParticipantResult result1 = new ParticipantResult();
+        result1.setTotalPayloadProcessed(PARTICIPANT1_TOTAL_PROCESSED);
+
+        ParticipantResult result2 = new ParticipantResult();
+        result2.setTotalPayloadProcessed(PARTICIPANT2_TOTAL_PROCESSED);
+
+        _aggregator.aggregate(result1);
+        _aggregator.aggregate(result2);
+
+        ParticipantResult aggregratedResult = _aggregator.getAggregatedResult();
+        assertEquals(OVERALL_PROCESSED, aggregratedResult.getTotalPayloadProcessed());
+    }
+
+    public void testComputeThroughput()
+    {
+        ParticipantResult result1 = new ParticipantResult();
+        result1.setStartDate(new Date(PARTICIPANT1_STARTDATE));
+        result1.setEndDate(new Date(PARTICIPANT1_ENDDATE));
+        result1.setTotalPayloadProcessed(PARTICIPANT1_TOTAL_PROCESSED);
+
+        ParticipantResult result2 = new ParticipantResult();
+        result2.setStartDate(new Date(PARTICIPANT2_STARTDATE));
+        result2.setEndDate(new Date(PARTICIPANT2_ENDDATE));
+        result2.setTotalPayloadProcessed(PARTICIPANT2_TOTAL_PROCESSED);
+
+        _aggregator.aggregate(result1);
+        _aggregator.aggregate(result2);
+
+        ParticipantResult aggregratedResult = _aggregator.getAggregatedResult();
+        assertEquals(EXPECTED_AGGREGATED_ALL_THROUGHPUT, aggregratedResult.getThroughput(), 0.1);
+    }
+
+    public void testConstantTestNameAndIterationNumberRolledUp() throws Exception
+    {
+        ParticipantResult result1 = new ParticipantResult();
+        result1.setTestName(TEST_NAME);
+        result1.setIterationNumber(TEST_ITERATION_NUMBER);
+
+        ParticipantResult result2 = new ParticipantResult();
+        result2.setTestName(TEST_NAME);
+        result2.setIterationNumber(TEST_ITERATION_NUMBER);
+
+        _aggregator.aggregate(result1);
+        _aggregator.aggregate(result2);
+
+        ParticipantResult aggregratedResult = _aggregator.getAggregatedResult();
+        assertEquals(TEST_ITERATION_NUMBER, aggregratedResult.getIterationNumber());
+        assertEquals(TEST_NAME, aggregratedResult.getTestName());
+    }
+
+    public void testConstantPayloadSizesRolledUp() throws Exception
+    {
+        final int payloadSize = 1024;
+
+        ParticipantResult result1 = new ParticipantResult();
+        result1.setPayloadSize(payloadSize);
+
+        ParticipantResult result2 = new ParticipantResult();
+        result2.setPayloadSize(payloadSize);
+
+        _aggregator.aggregate(result1);
+        _aggregator.aggregate(result2);
+
+        ParticipantResult aggregratedResult = _aggregator.getAggregatedResult();
+        assertEquals(payloadSize, aggregratedResult.getPayloadSize());
+    }
+
+    public void testDifferingPayloadSizesNotRolledUp() throws Exception
+    {
+        final int payload1Size = 1024;
+        final int payload2Size = 2048;
+
+        ParticipantResult result1 = new ParticipantResult();
+        result1.setPayloadSize(payload1Size);
+
+        ParticipantResult result2 = new ParticipantResult();
+        result2.setPayloadSize(payload2Size);
+
+        _aggregator.aggregate(result1);
+        _aggregator.aggregate(result2);
+
+        ParticipantResult aggregratedResult = _aggregator.getAggregatedResult();
+        assertEquals(0, aggregratedResult.getPayloadSize());
+    }
+
+}
diff --git a/qpid/java/perftests/src/test/java/org/apache/qpid/disttest/results/aggregation/TestResultAggregatorTest.java b/qpid/java/perftests/src/test/java/org/apache/qpid/disttest/results/aggregation/TestResultAggregatorTest.java
new file mode 100644
index 0000000000..7cf900ca79
--- /dev/null
+++ b/qpid/java/perftests/src/test/java/org/apache/qpid/disttest/results/aggregation/TestResultAggregatorTest.java
@@ -0,0 +1,144 @@
+/*
+ * 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.disttest.results.aggregation;
+
+import java.util.Date;
+
+import junit.framework.TestCase;
+
+import org.apache.qpid.disttest.controller.TestResult;
+import org.apache.qpid.disttest.message.ConsumerParticipantResult;
+import org.apache.qpid.disttest.message.ParticipantResult;
+import org.apache.qpid.disttest.message.ProducerParticipantResult;
+import org.apache.qpid.disttest.results.aggregation.AggregatedTestResult;
+import org.apache.qpid.disttest.results.aggregation.TestResultAggregator;
+
+public class TestResultAggregatorTest extends TestCase
+{
+
+    private static final String TEST1_NAME = "TEST1_NAME";
+    private static final int TEST1_ITERATION_NUMBER = 1;
+
+
+    private static final String CONSUMER_PARTICIPANT_NAME1 = "CONSUMER_PARTICIPANT_NAME1";
+    private static final String CONSUMER_PARTICIPANT_NAME2 = "CONSUMER_PARTICIPANT_NAME2";
+
+    private static final String PRODUCER_PARTICIPANT_NAME = "PRODUCER_PARTICIPANT_NAME";
+
+
+    private static final long CONSUMER1_STARTDATE = 50;
+    private static final long CONSUMER1_ENDDATE = 20000;
+
+    private static final long CONSUMER2_STARTDATE = 100;
+    private static final long CONSUMER2_ENDDATE = 21000;
+
+    private static final long PRODUCER_STARTDATE = 0;
+    private static final long PRODUCER_ENDDATE = 19000;
+
+    private static final long NUMBER_OF_MESSAGES_PROCESSED_PER_CONSUMER = 50;
+    private static final long NUMBER_OF_MESSAGES_CONSUMED_IN_TOTAL = NUMBER_OF_MESSAGES_PROCESSED_PER_CONSUMER * 2;
+    private static final long NUMBER_OF_MESSAGES_PRODUCED = NUMBER_OF_MESSAGES_PROCESSED_PER_CONSUMER * 2;
+
+    private static final int PAYLOAD_SIZE = 1024;
+    private static final long TOTAL_PAYLOAD_PROCESSED_PER_CONSUMER = NUMBER_OF_MESSAGES_PROCESSED_PER_CONSUMER * PAYLOAD_SIZE;
+    private static final long TOTAL_PAYLOAD_PRODUCED_IN_TOTAL =  TOTAL_PAYLOAD_PROCESSED_PER_CONSUMER * 2;
+
+    private static final int EXPECTED_NUMBER_OF_AGGREGATED_RESULTS = 3;
+
+    private TestResultAggregator _aggregator = new TestResultAggregator();
+
+    public void testAggregateResultsForTwoConsumerAndOneProducer() throws Exception
+    {
+        TestResult originalTestResult = createResultsFromTest();
+
+        int numberOfOriginalParticipantResults = originalTestResult.getParticipantResults().size();
+        int expectedNumberOfResults = numberOfOriginalParticipantResults + EXPECTED_NUMBER_OF_AGGREGATED_RESULTS;
+
+        AggregatedTestResult aggregatedTestResult = _aggregator.aggregateTestResult(originalTestResult);
+
+        aggregatedTestResult.getAllConsumerParticipantResult().getTotalPayloadProcessed();
+        assertEquals(expectedNumberOfResults, aggregatedTestResult.getParticipantResults().size());
+
+        assertMinimalAggregatedResults(
+                aggregatedTestResult.getAllParticipantResult(),
+                TEST1_NAME, TEST1_ITERATION_NUMBER,
+                NUMBER_OF_MESSAGES_CONSUMED_IN_TOTAL);
+
+        assertMinimalAggregatedResults(
+                aggregatedTestResult.getAllConsumerParticipantResult(),
+                TEST1_NAME, TEST1_ITERATION_NUMBER,
+                NUMBER_OF_MESSAGES_CONSUMED_IN_TOTAL);
+
+        assertMinimalAggregatedResults(
+                aggregatedTestResult.getAllProducerParticipantResult(),
+                TEST1_NAME, TEST1_ITERATION_NUMBER,
+                NUMBER_OF_MESSAGES_PRODUCED);
+    }
+
+    public void testAggregateResultsWhenParticipantErrored()
+    {
+        ParticipantResult failedParticipantResult = new ParticipantResult();
+        failedParticipantResult.setParticipantName(PRODUCER_PARTICIPANT_NAME);
+        failedParticipantResult.setErrorMessage("error");
+        TestResult result = new TestResult(TEST1_NAME);
+        result.addParticipantResult(failedParticipantResult);
+
+        AggregatedTestResult aggregatedTestResult = _aggregator.aggregateTestResult(result);
+        assertEquals(TestResultAggregator.AGGREGATED_ERROR_MESSAGE, aggregatedTestResult.getAllParticipantResult().getErrorMessage());
+    }
+
+    private void assertMinimalAggregatedResults(ParticipantResult result, String expectedTestName, int expectedIterationNumber, long expectedNumberOfMessagesProcessed)
+    {
+        assertEquals("Unexpected test name in " + result.getParticipantName(), expectedTestName, result.getTestName());
+        assertEquals("Unexpected iteration number in " + result.getParticipantName(), expectedIterationNumber, result.getIterationNumber());
+        assertEquals("Unexpected number of messages processed in " + result.getParticipantName(), expectedNumberOfMessagesProcessed, result.getNumberOfMessagesProcessed());
+    }
+
+    private TestResult createResultsFromTest()
+    {
+        TestResult testResult = new TestResult(TEST1_NAME);
+
+        ConsumerParticipantResult consumerResult1 = new ConsumerParticipantResult();
+        setPropertiesOn(consumerResult1, TEST1_NAME, TEST1_ITERATION_NUMBER, CONSUMER_PARTICIPANT_NAME1, NUMBER_OF_MESSAGES_PROCESSED_PER_CONSUMER, PAYLOAD_SIZE, TOTAL_PAYLOAD_PROCESSED_PER_CONSUMER, CONSUMER1_STARTDATE, CONSUMER1_ENDDATE);
+        testResult.addParticipantResult(consumerResult1);
+
+        ConsumerParticipantResult consumerResult2 = new ConsumerParticipantResult();
+        setPropertiesOn(consumerResult2, TEST1_NAME, TEST1_ITERATION_NUMBER, CONSUMER_PARTICIPANT_NAME2, NUMBER_OF_MESSAGES_PROCESSED_PER_CONSUMER, PAYLOAD_SIZE, TOTAL_PAYLOAD_PROCESSED_PER_CONSUMER, CONSUMER2_STARTDATE, CONSUMER2_ENDDATE);
+        testResult.addParticipantResult(consumerResult2);
+
+        ParticipantResult producerResult = new ProducerParticipantResult();
+        setPropertiesOn(producerResult, TEST1_NAME, TEST1_ITERATION_NUMBER, PRODUCER_PARTICIPANT_NAME, NUMBER_OF_MESSAGES_PRODUCED, PAYLOAD_SIZE, TOTAL_PAYLOAD_PRODUCED_IN_TOTAL, PRODUCER_STARTDATE, PRODUCER_ENDDATE);
+        testResult.addParticipantResult(producerResult);
+
+        return testResult;
+    }
+
+    private void setPropertiesOn(ParticipantResult participantResult, String testName, int iterationNumber, String participantName, long numberOfMessagesProcessed, int payloadSize, long totalPayloadProcessed, long start, long end)
+    {
+        participantResult.setParticipantName(participantName);
+        participantResult.setTestName(testName);
+        participantResult.setIterationNumber(iterationNumber);
+
+        participantResult.setNumberOfMessagesProcessed(numberOfMessagesProcessed);
+        participantResult.setPayloadSize(payloadSize);
+        participantResult.setTotalPayloadProcessed(totalPayloadProcessed);
+        participantResult.setStartDate(new Date(start));
+        participantResult.setEndDate(new Date(end));
+    }
+}
diff --git a/qpid/java/perftests/src/test/java/org/apache/qpid/disttest/results/formatting/CSVFormaterTest.java b/qpid/java/perftests/src/test/java/org/apache/qpid/disttest/results/formatting/CSVFormaterTest.java
new file mode 100644
index 0000000000..c8f5b3a231
--- /dev/null
+++ b/qpid/java/perftests/src/test/java/org/apache/qpid/disttest/results/formatting/CSVFormaterTest.java
@@ -0,0 +1,140 @@
+/*
+ * 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.disttest.results.formatting;
+
+import static org.apache.qpid.disttest.message.ParticipantAttribute.BATCH_SIZE;
+import static org.apache.qpid.disttest.message.ParticipantAttribute.CONFIGURED_CLIENT_NAME;
+import static org.apache.qpid.disttest.message.ParticipantAttribute.DELIVERY_MODE;
+import static org.apache.qpid.disttest.message.ParticipantAttribute.ERROR_MESSAGE;
+import static org.apache.qpid.disttest.message.ParticipantAttribute.IS_BROWSIING_SUBSCRIPTION;
+import static org.apache.qpid.disttest.message.ParticipantAttribute.IS_DURABLE_SUBSCRIPTION;
+import static org.apache.qpid.disttest.message.ParticipantAttribute.IS_NO_LOCAL;
+import static org.apache.qpid.disttest.message.ParticipantAttribute.IS_SELECTOR;
+import static org.apache.qpid.disttest.message.ParticipantAttribute.IS_SYNCHRONOUS_CONSUMER;
+import static org.apache.qpid.disttest.message.ParticipantAttribute.IS_TOPIC;
+import static org.apache.qpid.disttest.message.ParticipantAttribute.ITERATION_NUMBER;
+import static org.apache.qpid.disttest.message.ParticipantAttribute.MAXIMUM_DURATION;
+import static org.apache.qpid.disttest.message.ParticipantAttribute.PAYLOAD_SIZE;
+import static org.apache.qpid.disttest.message.ParticipantAttribute.NUMBER_OF_MESSAGES_PROCESSED;
+import static org.apache.qpid.disttest.message.ParticipantAttribute.PARTICIPANT_NAME;
+import static org.apache.qpid.disttest.message.ParticipantAttribute.PRIORITY;
+import static org.apache.qpid.disttest.message.ParticipantAttribute.PRODUCER_INTERVAL;
+import static org.apache.qpid.disttest.message.ParticipantAttribute.PRODUCER_START_DELAY;
+import static org.apache.qpid.disttest.message.ParticipantAttribute.TEST_NAME;
+import static org.apache.qpid.disttest.message.ParticipantAttribute.TIME_TAKEN;
+import static org.apache.qpid.disttest.message.ParticipantAttribute.TIME_TO_LIVE;
+import static org.apache.qpid.disttest.message.ParticipantAttribute.THROUGHPUT;
+import static org.apache.qpid.disttest.message.ParticipantAttribute.TOTAL_PAYLOAD_PROCESSED;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+import java.io.BufferedReader;
+import java.io.InputStream;
+import java.io.InputStreamReader;
+import java.util.HashMap;
+import java.util.Map;
+
+import junit.framework.TestCase;
+
+import org.apache.qpid.disttest.controller.ResultsForAllTests;
+import org.apache.qpid.disttest.controller.TestResult;
+import org.apache.qpid.disttest.message.ParticipantAttribute;
+import org.apache.qpid.disttest.message.ParticipantResult;
+import org.apache.qpid.disttest.results.formatting.CSVFormater;
+
+public class CSVFormaterTest extends TestCase
+{
+    private static final String TEST1 = "TEST1";
+    private static final String PARTICIPANT = "PARTICIPANT";
+    private static final String CONFIGURED_CLIENT1 = "CONFIGURED_CLIENT1";
+
+    private CSVFormater _formatter = new CSVFormater();
+
+    public void testResultsFileWithWithOneRow() throws Exception
+    {
+        ParticipantResult participantResult = mock(ParticipantResult.class);
+        Map<ParticipantAttribute, Object> participantAttributes = getParticipantAttributes();
+
+        when(participantResult.getAttributes()).thenReturn(participantAttributes);
+
+        TestResult testResult = new TestResult(TEST1);
+        testResult.addParticipantResult(participantResult);
+
+        ResultsForAllTests resultsForAllTests = new ResultsForAllTests();
+        resultsForAllTests.add(testResult);
+
+        String output = _formatter.format(resultsForAllTests);
+
+        String expectedOutput = readCsvOutputFileAsString("expectedOutput.csv");
+
+        assertEquals(expectedOutput, output);
+    }
+
+    private Map<ParticipantAttribute, Object> getParticipantAttributes()
+    {
+        Map<ParticipantAttribute, Object> participantAttributes = new HashMap<ParticipantAttribute, Object>();
+
+        participantAttributes.put(TEST_NAME, TEST1);
+        participantAttributes.put(ITERATION_NUMBER, 0);
+        participantAttributes.put(CONFIGURED_CLIENT_NAME, CONFIGURED_CLIENT1);
+        participantAttributes.put(PARTICIPANT_NAME, PARTICIPANT);
+        participantAttributes.put(NUMBER_OF_MESSAGES_PROCESSED, 1);
+        participantAttributes.put(PAYLOAD_SIZE, 2);
+        participantAttributes.put(PRIORITY, 3);
+        participantAttributes.put(TIME_TO_LIVE, 4);
+        participantAttributes.put(DELIVERY_MODE, 5);
+        participantAttributes.put(BATCH_SIZE, 6);
+        participantAttributes.put(MAXIMUM_DURATION, 7);
+        participantAttributes.put(PRODUCER_START_DELAY, 8);
+        participantAttributes.put(PRODUCER_INTERVAL, 9);
+        participantAttributes.put(IS_TOPIC, true);
+        participantAttributes.put(IS_DURABLE_SUBSCRIPTION, false);
+        participantAttributes.put(IS_BROWSIING_SUBSCRIPTION, true);
+        participantAttributes.put(IS_SELECTOR, false);
+        participantAttributes.put(IS_NO_LOCAL, true);
+        participantAttributes.put(IS_SYNCHRONOUS_CONSUMER, false);
+        participantAttributes.put(TOTAL_PAYLOAD_PROCESSED, 1024);
+        participantAttributes.put(THROUGHPUT, 2048);
+        participantAttributes.put(TIME_TAKEN, 1000);
+        participantAttributes.put(ERROR_MESSAGE, "error");
+
+        return participantAttributes;
+    }
+
+    private String readCsvOutputFileAsString(String filename) throws Exception
+    {
+        InputStream is = getClass().getResourceAsStream(filename);
+        assertNotNull(is);
+
+        StringBuilder output = new StringBuilder();
+
+        BufferedReader br = new BufferedReader(new InputStreamReader(is));
+        String line = null;
+        while((line = br.readLine()) != null)
+        {
+            output.append(line);
+            output.append("\n");
+        }
+
+        return output.toString();
+    }
+
+
+
+}
diff --git a/qpid/java/perftests/src/test/java/org/apache/qpid/disttest/results/formatting/CSVOrderParticipantResultComparatorTest.java b/qpid/java/perftests/src/test/java/org/apache/qpid/disttest/results/formatting/CSVOrderParticipantResultComparatorTest.java
new file mode 100644
index 0000000000..6cec4b5245
--- /dev/null
+++ b/qpid/java/perftests/src/test/java/org/apache/qpid/disttest/results/formatting/CSVOrderParticipantResultComparatorTest.java
@@ -0,0 +1,89 @@
+/*
+ * 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.disttest.results.formatting;
+
+
+import junit.framework.TestCase;
+
+import org.apache.qpid.disttest.message.ConsumerParticipantResult;
+import org.apache.qpid.disttest.message.ParticipantResult;
+import org.apache.qpid.disttest.message.ProducerParticipantResult;
+import org.apache.qpid.disttest.results.aggregation.TestResultAggregator;
+import org.apache.qpid.disttest.results.formatting.CSVOrderParticipantResultComparator;
+
+public class CSVOrderParticipantResultComparatorTest extends TestCase
+{
+    CSVOrderParticipantResultComparator _comparator = new CSVOrderParticipantResultComparator();
+
+    public void testOrderedConsumerParticipants() throws Exception
+    {
+        assertCompare(
+                new ConsumerParticipantResult("apple"),
+                new ConsumerParticipantResult("banana"));
+
+    }
+    public void testProducerPrecedesConsumerParticipants() throws Exception
+    {
+        assertCompare(
+                new ProducerParticipantResult(),
+                new ConsumerParticipantResult());
+    }
+
+    public void testProducerPrecedesAllProducersResult()
+    {
+        assertCompare(
+                new ProducerParticipantResult("participantName"),
+                new ParticipantResult(TestResultAggregator.ALL_PRODUCER_PARTICIPANTS_NAME));
+    }
+
+    public void testConsumerPrecedesAllConsumersResult()
+    {
+        assertCompare(
+                new ConsumerParticipantResult("participantName"),
+                new ParticipantResult(TestResultAggregator.ALL_CONSUMER_PARTICIPANTS_NAME));
+    }
+
+    public void testAllParticipantsPrecedesAllConsumersResult()
+    {
+        assertCompare(
+                new ParticipantResult(TestResultAggregator.ALL_PARTICIPANTS_NAME),
+                new ParticipantResult(TestResultAggregator.ALL_CONSUMER_PARTICIPANTS_NAME));
+    }
+
+    public void testAllParticipantsPrecedesAllProducersResult()
+    {
+        assertCompare(
+                new ParticipantResult(TestResultAggregator.ALL_PARTICIPANTS_NAME),
+                new ParticipantResult(TestResultAggregator.ALL_PRODUCER_PARTICIPANTS_NAME));
+    }
+
+    private void assertCompare(ParticipantResult smaller, ParticipantResult bigger)
+    {
+        assertEquals("Expected " + smaller + " to 'equal' itself",
+                0,
+                _comparator.compare(smaller, smaller));
+
+        String failureMsg = "Expected " + smaller + " to be smaller than " + bigger;
+
+        assertTrue(failureMsg, _comparator.compare(smaller, bigger) < 0);
+        assertTrue(failureMsg, _comparator.compare(bigger, smaller) > 0);
+    }
+
+}
+// <ParticipantResult>
diff --git a/qpid/java/perftests/src/test/java/org/apache/qpid/disttest/results/formatting/expectedOutput.csv b/qpid/java/perftests/src/test/java/org/apache/qpid/disttest/results/formatting/expectedOutput.csv
new file mode 100644
index 0000000000..08cb2275b5
--- /dev/null
+++ b/qpid/java/perftests/src/test/java/org/apache/qpid/disttest/results/formatting/expectedOutput.csv
@@ -0,0 +1,2 @@
+Test Name,Iteration number,Client Name,Participant name,Number of messages,Payload size (bytes),Priority,Time to live (ms),Delivery mode,Batch size,Maximum duration (ms),Producer start delay (ms),Producer interval (ms),Is topic,Is durable subscription,Is browsing subscription,Is selector,Is no local,Is synchronous consumer,Total payload processed (bytes),Throughput (kbytes/s),Time taken (ms),Error message
+TEST1,0,CONFIGURED_CLIENT1,PARTICIPANT,1,2,3,4,5,6,7,8,9,true,false,true,false,true,false,1024,2048,1000,error
\ No newline at end of file
diff --git a/qpid/java/perftests/src/test/java/org/apache/qpid/systest/disttest/DistributedTestSystemTestBase.java b/qpid/java/perftests/src/test/java/org/apache/qpid/systest/disttest/DistributedTestSystemTestBase.java
new file mode 100644
index 0000000000..96daf64526
--- /dev/null
+++ b/qpid/java/perftests/src/test/java/org/apache/qpid/systest/disttest/DistributedTestSystemTestBase.java
@@ -0,0 +1,72 @@
+/*
+ * 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.systest.disttest;
+
+import java.util.Properties;
+
+import javax.jms.Connection;
+import javax.jms.ConnectionFactory;
+import javax.jms.JMSException;
+import javax.naming.Context;
+import javax.naming.InitialContext;
+import javax.naming.NamingException;
+
+import org.apache.qpid.test.utils.QpidBrokerTestCase;
+
+public class DistributedTestSystemTestBase extends QpidBrokerTestCase
+{
+    protected Context _context;
+
+    protected Connection _connection;
+
+    @Override
+    protected void setUp() throws Exception
+    {
+        super.setUp();
+
+        final Properties properties = new Properties();
+        properties.load(DistributedTestSystemTestBase.class.getResourceAsStream("perftests.systests.properties"));
+        _context = new InitialContext(properties);
+
+        _connection = getConnection();
+        _connection.start();
+    }
+
+    @Override
+    protected void tearDown() throws Exception
+    {
+        // no need to close connections - this is done by superclass
+
+        super.tearDown();
+    }
+
+    public Context getContext()
+    {
+        return _context;
+    }
+
+    @Override
+    public Connection getConnection() throws JMSException, NamingException
+    {
+        final ConnectionFactory connectionFactory = (ConnectionFactory) _context.lookup("connectionfactory");
+        final Connection connection = connectionFactory.createConnection();
+        _connections.add(connection);
+        return connection;
+    }
+}
diff --git a/qpid/java/perftests/src/test/java/org/apache/qpid/systest/disttest/QpidQueueCreatorTest.java b/qpid/java/perftests/src/test/java/org/apache/qpid/systest/disttest/QpidQueueCreatorTest.java
new file mode 100644
index 0000000000..3fa2b9836e
--- /dev/null
+++ b/qpid/java/perftests/src/test/java/org/apache/qpid/systest/disttest/QpidQueueCreatorTest.java
@@ -0,0 +1,117 @@
+/*
+ * 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.systest.disttest;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import javax.jms.Connection;
+import javax.jms.Session;
+
+import org.apache.qpid.client.AMQDestination;
+import org.apache.qpid.client.AMQSession;
+import org.apache.qpid.disttest.DistributedTestException;
+import org.apache.qpid.disttest.controller.config.QueueConfig;
+import org.apache.qpid.disttest.jms.QpidQueueCreator;
+
+public class QpidQueueCreatorTest extends DistributedTestSystemTestBase
+{
+    private static final Map<String, Object> EMPTY_ATTRIBUTES = Collections.emptyMap();
+
+    private QpidQueueCreator _creator;
+    private Connection _connection;
+    private Session _session;
+    private List<QueueConfig> _configs;
+    private String _queueName;
+
+    @Override
+    public void setUp() throws Exception
+    {
+        super.setUp();
+        _connection = getConnection();
+        _session = _connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
+        _creator = new QpidQueueCreator();
+        _configs = new ArrayList<QueueConfig>();
+        _queueName = "direct://amq.direct//" + getTestQueueName();
+    }
+
+    public void testCreateQueueWithoutAttributes() throws Exception
+    {
+        _configs.add(new QueueConfig(_queueName, true, EMPTY_ATTRIBUTES));
+
+        assertQueueBound(_queueName, false);
+
+        _creator.createQueues(_connection, _configs);
+
+        assertQueueBound(_queueName, true);
+    }
+
+    public void testCreateWithAttributes() throws Exception
+    {
+        Map<String, Object> attributes = new HashMap<String, Object>();
+        attributes.put("x-qpid-priorities", Integer.valueOf(5));
+        _configs.add(new QueueConfig(_queueName, true, attributes));
+
+        assertQueueBound(_queueName, false);
+
+        _creator.createQueues(_connection, _configs);
+
+        assertQueueBound(_queueName, true);
+    }
+
+    public void testDeleteQueues() throws Exception
+    {
+        _configs.add(new QueueConfig(_queueName, true, EMPTY_ATTRIBUTES));
+
+        assertQueueBound(_queueName, false);
+
+        _creator.createQueues(_connection, _configs);
+        assertQueueBound(_queueName, true);
+
+        _creator.deleteQueues(_connection, _configs);
+        assertQueueBound(_queueName, false);
+    }
+
+    public void testDeleteQueueThatDoesNotExist() throws Exception
+    {
+        String queueThatDoesNotExist = _queueName;
+        List<QueueConfig> configs = new ArrayList<QueueConfig>();
+        Map<String, Object> attributes = Collections.emptyMap();
+        configs.add(new QueueConfig(queueThatDoesNotExist, true, attributes));
+
+        try
+        {
+            _creator.deleteQueues(_connection, configs);
+            fail("Exception not thrown");
+        }
+        catch (DistributedTestException e)
+        {
+            // PASS
+        }
+    }
+
+    private void assertQueueBound(String queueName, boolean isBound) throws Exception
+    {
+        AMQDestination destination = (AMQDestination)_session.createQueue(queueName);
+        assertEquals("Queue is not in expected bound state", isBound, ((AMQSession<?, ?>)_session).isQueueBound(destination));
+    }
+}
diff --git a/qpid/java/perftests/src/test/java/org/apache/qpid/systest/disttest/SystemTestConstants.java b/qpid/java/perftests/src/test/java/org/apache/qpid/systest/disttest/SystemTestConstants.java
new file mode 100644
index 0000000000..808b428bc9
--- /dev/null
+++ b/qpid/java/perftests/src/test/java/org/apache/qpid/systest/disttest/SystemTestConstants.java
@@ -0,0 +1,28 @@
+/*
+ * 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.systest.disttest;
+
+public abstract class SystemTestConstants
+{
+    public static final long REGISTRATION_TIMEOUT = 5000;
+    public static final long COMMAND_RESPONSE_TIMEOUT = 10000;
+    public static final long TEST_RESULT_TIMEOUT = 5000;
+
+}
diff --git a/qpid/java/perftests/src/test/java/org/apache/qpid/systest/disttest/clientonly/BasicDistributedClientTest.java b/qpid/java/perftests/src/test/java/org/apache/qpid/systest/disttest/clientonly/BasicDistributedClientTest.java
new file mode 100644
index 0000000000..d599bdc5c4
--- /dev/null
+++ b/qpid/java/perftests/src/test/java/org/apache/qpid/systest/disttest/clientonly/BasicDistributedClientTest.java
@@ -0,0 +1,186 @@
+/*
+ * 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.systest.disttest.clientonly;
+
+import javax.jms.Destination;
+import javax.jms.JMSException;
+import javax.jms.Message;
+import javax.jms.MessageProducer;
+import javax.jms.Queue;
+import javax.jms.Session;
+
+import org.apache.qpid.disttest.client.Client;
+import org.apache.qpid.disttest.client.ClientState;
+import org.apache.qpid.disttest.jms.ClientJmsDelegate;
+import org.apache.qpid.disttest.jms.JmsMessageAdaptor;
+import org.apache.qpid.disttest.message.Command;
+import org.apache.qpid.disttest.message.CommandType;
+import org.apache.qpid.disttest.message.CreateConnectionCommand;
+import org.apache.qpid.disttest.message.CreateSessionCommand;
+import org.apache.qpid.disttest.message.NoOpCommand;
+import org.apache.qpid.disttest.message.RegisterClientCommand;
+import org.apache.qpid.disttest.message.Response;
+import org.apache.qpid.disttest.message.StopClientCommand;
+import org.apache.qpid.systest.disttest.DistributedTestSystemTestBase;
+
+public class BasicDistributedClientTest extends DistributedTestSystemTestBase
+{
+    private Session _session = null;
+    private MessageProducer _clientQueueProducer;
+    private Client _client;
+    private ControllerQueue _controllerQueue;
+    private ClientJmsDelegate _clientJmsDelegate = null;
+
+    @Override
+    protected void setUp() throws Exception
+    {
+        super.setUp();
+
+        _controllerQueue = new ControllerQueue(_connection, _context);
+        _session = _connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
+
+        _clientJmsDelegate = new ClientJmsDelegate(_context);
+        _client = new Client(_clientJmsDelegate);
+        _client.start();
+    }
+
+    @Override
+    protected void tearDown() throws Exception
+    {
+        try
+        {
+            _controllerQueue.close();
+            if (_session != null)
+            {
+                _session.close();
+            }
+        }
+        finally
+        {
+            super.tearDown();
+        }
+    }
+
+    public void testClientSendsRegistrationMessage() throws Exception
+    {
+        final RegisterClientCommand regClientCommand = _controllerQueue.getNext();
+
+        assertNotNull("Client must have a non-null name", regClientCommand.getClientName());
+        assertEquals("Unexpected client name", _clientJmsDelegate.getClientName(), regClientCommand.getClientName());
+        assertNotNull("Client queue name should not be null", regClientCommand.getClientQueueName());
+    }
+
+    public void testClientSendsCommandResponses() throws Exception
+    {
+        final RegisterClientCommand registrationCommand = _controllerQueue.getNext();
+        createClientQueueProducer(registrationCommand);
+
+        sendCommandToClient(new NoOpCommand());
+
+        final Response responseCommand = _controllerQueue.getNext();
+        assertEquals("Incorrect client message type", CommandType.RESPONSE, responseCommand.getType());
+    }
+
+    public void testClientCanBeStoppedViaCommand() throws Exception
+    {
+        assertEquals("Expected client to be in STARTED state", ClientState.READY, _client.getState());
+
+        final RegisterClientCommand registrationCommand = _controllerQueue.getNext();
+        createClientQueueProducer(registrationCommand);
+
+        final Command stopClientCommand = new StopClientCommand();
+        sendCommandToClient(stopClientCommand);
+
+        _client.waitUntilStopped(1000);
+
+        Response response = _controllerQueue.getNext();
+        assertNotNull(response);
+        assertFalse("response shouldn't contain error", response.hasError());
+
+        assertEquals("Expected client to be in STOPPED state", ClientState.STOPPED, _client.getState());
+    }
+
+    public void testClientCanCreateTestConnection() throws Exception
+    {
+        assertEquals("Unexpected number of test connections", 0, _clientJmsDelegate.getNoOfTestConnections());
+
+        final RegisterClientCommand registration = _controllerQueue.getNext();
+        createClientQueueProducer(registration);
+
+        final CreateConnectionCommand createConnectionCommand = new CreateConnectionCommand();
+        createConnectionCommand.setConnectionName("newTestConnection");
+        createConnectionCommand.setConnectionFactoryName("connectionfactory");
+
+        sendCommandToClient(createConnectionCommand);
+        Response response = _controllerQueue.getNext();
+
+        assertFalse("Response message should not have indicated an error", response.hasError());
+        assertEquals("Unexpected number of test connections", 1, _clientJmsDelegate.getNoOfTestConnections());
+    }
+
+    public void testClientCanCreateTestSession() throws Exception
+    {
+        assertEquals("Unexpected number of test sessions", 0, _clientJmsDelegate.getNoOfTestSessions());
+
+        final RegisterClientCommand registration = _controllerQueue.getNext();
+        createClientQueueProducer(registration);
+
+        final CreateConnectionCommand createConnectionCommand = new CreateConnectionCommand();
+        createConnectionCommand.setConnectionName("newTestConnection");
+        createConnectionCommand.setConnectionFactoryName("connectionfactory");
+
+        sendCommandToClient(createConnectionCommand);
+        Response response = _controllerQueue.getNext();
+        assertFalse("Response message should not have indicated an error", response.hasError());
+
+        final CreateSessionCommand createSessionCommand = new CreateSessionCommand();
+        createSessionCommand.setConnectionName("newTestConnection");
+        createSessionCommand.setSessionName("newTestSession");
+        createSessionCommand.setAcknowledgeMode(Session.AUTO_ACKNOWLEDGE);
+
+        sendCommandToClient(createSessionCommand);
+        response = _controllerQueue.getNext();
+
+        assertFalse("Response message should not have indicated an error", response.hasError());
+        assertEquals("Unexpected number of test sessions", 1, _clientJmsDelegate.getNoOfTestSessions());
+    }
+
+    private void sendCommandToClient(final Command command) throws JMSException
+    {
+        final Message message = JmsMessageAdaptor.commandToMessage(_session, command);
+        _clientQueueProducer.send(message);
+    }
+
+    private void createClientQueueProducer(
+            final RegisterClientCommand registration) throws JMSException
+    {
+        final Destination clientCommandQueue = createDestinationFromRegistration(registration);
+        _clientQueueProducer  = _session.createProducer(clientCommandQueue);
+    }
+
+    private Queue createDestinationFromRegistration(
+            final RegisterClientCommand registrationCommand)
+            throws JMSException
+    {
+        String clientQueueName = registrationCommand.getClientQueueName();
+        assertNotNull("Null client queue in register message", clientQueueName);
+        return _session.createQueue(clientQueueName);
+    }
+}
diff --git a/qpid/java/perftests/src/test/java/org/apache/qpid/systest/disttest/clientonly/ConsumerParticipantTest.java b/qpid/java/perftests/src/test/java/org/apache/qpid/systest/disttest/clientonly/ConsumerParticipantTest.java
new file mode 100644
index 0000000000..a3c0430865
--- /dev/null
+++ b/qpid/java/perftests/src/test/java/org/apache/qpid/systest/disttest/clientonly/ConsumerParticipantTest.java
@@ -0,0 +1,156 @@
+/*
+ * 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.systest.disttest.clientonly;
+
+import javax.jms.Message;
+import javax.jms.MessageConsumer;
+import javax.jms.MessageProducer;
+import javax.jms.Session;
+
+import org.apache.qpid.disttest.client.Client;
+import org.apache.qpid.disttest.client.ConsumerParticipant;
+import org.apache.qpid.disttest.client.ParticipantExecutor;
+import org.apache.qpid.disttest.message.CreateConsumerCommand;
+import org.apache.qpid.disttest.message.ParticipantResult;
+import org.apache.qpid.systest.disttest.DistributedTestSystemTestBase;
+import org.apache.qpid.systest.disttest.clientonly.ProducerParticipantTest.TestClientJmsDelegate;
+
+public class ConsumerParticipantTest  extends DistributedTestSystemTestBase
+{
+    private MessageProducer _producer;
+    private Session _session;
+    private TestClientJmsDelegate _delegate;
+    private Client _client;
+    private ControllerQueue _controllerQueue;
+
+    @Override
+    protected void setUp() throws Exception
+    {
+        super.setUp();
+
+        _controllerQueue = new ControllerQueue(_connection, _context);
+        _session = _connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
+        _producer = _session.createProducer(getTestQueue());
+
+        _delegate = new TestClientJmsDelegate(getContext());
+        _client = new Client(_delegate);
+    }
+
+
+    @Override
+    protected void tearDown() throws Exception
+    {
+        _controllerQueue.close();
+        super.tearDown();
+    }
+
+    public void testConsumeNumberOfMessagesSynchronously() throws Exception
+    {
+        runTest(Session.AUTO_ACKNOWLEDGE, 10, 0, true);
+    }
+
+    public void testConsumeNumberOfMessagesAsynchronously() throws Exception
+    {
+        runTest(Session.AUTO_ACKNOWLEDGE, 10, 0, false);
+    }
+
+    public void testSelectors() throws Exception
+    {
+        final CreateConsumerCommand command = new CreateConsumerCommand();
+        command.setNumberOfMessages(10);
+        command.setSessionName("testSession");
+        command.setDestinationName(getTestQueueName());
+        command.setSelector("id=1");
+        Session session = _connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
+
+        _delegate.addConnection("name-does-not-matter", _connection);
+        _delegate.addSession(command.getSessionName(), session);
+
+        ConsumerParticipant consumerParticipant =  new ConsumerParticipant(_delegate, command);
+        _delegate.createConsumer(command);
+
+        for (int i = 0; i < 20; i++)
+        {
+            Message message = _session.createMessage();
+            if (i % 2 == 0)
+            {
+                message.setIntProperty("id", 0);
+            }
+            else
+            {
+                message.setIntProperty("id", 1);
+            }
+            _producer.send(message);
+        }
+
+        new ParticipantExecutor(consumerParticipant).start(_client);
+
+        ParticipantResult results = _controllerQueue.getNext();
+        assertNotNull("No results message recieved", results);
+        assertEquals("Unexpected number of messages received", 10, results.getNumberOfMessagesProcessed());
+
+        Session testQueueConsumerSession = _connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
+        final MessageConsumer testQueueConsumer = testQueueConsumerSession.createConsumer(getTestQueue());
+        for (int i = 0; i < 10; i++)
+        {
+            Message message = testQueueConsumer.receive(2000);
+            assertNotNull("Message is not received: " + message, message);
+            assertEquals("Unexpected id value", 0, message.getIntProperty("id"));
+        }
+        Message message = testQueueConsumer.receive(2000);
+        assertNull("Unexpected message remaining on test queue: " + message, message);
+
+        _connection.stop();
+    }
+
+    protected void runTest(int acknowledgeMode, int numberOfMessages, int batchSize, boolean synchronous) throws Exception
+    {
+        final CreateConsumerCommand command = new CreateConsumerCommand();
+        command.setNumberOfMessages(numberOfMessages);
+        command.setBatchSize(batchSize);
+        command.setSessionName("testSession");
+        command.setDestinationName(getTestQueueName());
+        command.setSynchronous(synchronous);
+
+        Session session = _connection.createSession(Session.SESSION_TRANSACTED == acknowledgeMode, acknowledgeMode);
+
+        _delegate.addConnection("name-does-not-matter", _connection);
+        _delegate.addSession(command.getSessionName(), session);
+
+        ConsumerParticipant consumerParticipant =  new ConsumerParticipant(_delegate, command);
+        _delegate.createConsumer(command);
+
+        for (int i = 0; i < numberOfMessages; i++)
+        {
+            _producer.send(_session.createMessage());
+        }
+
+        new ParticipantExecutor(consumerParticipant).start(_client);
+
+        ParticipantResult results = _controllerQueue.getNext();
+        assertNotNull("No results message recieved", results);
+
+        Session testQueueConsumerSession = _connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
+        final MessageConsumer testQueueConsumer = testQueueConsumerSession.createConsumer(getTestQueue());
+        Message message = testQueueConsumer.receive(2000);
+        assertNull("Unexpected message remaining on test queue: " + message, message);
+
+        _connection.stop();
+    }
+}
diff --git a/qpid/java/perftests/src/test/java/org/apache/qpid/systest/disttest/clientonly/ControllerQueue.java b/qpid/java/perftests/src/test/java/org/apache/qpid/systest/disttest/clientonly/ControllerQueue.java
new file mode 100644
index 0000000000..7f0c23eb38
--- /dev/null
+++ b/qpid/java/perftests/src/test/java/org/apache/qpid/systest/disttest/clientonly/ControllerQueue.java
@@ -0,0 +1,90 @@
+package org.apache.qpid.systest.disttest.clientonly;
+
+import java.util.Map;
+
+import javax.jms.Connection;
+import javax.jms.Destination;
+import javax.jms.JMSException;
+import javax.jms.Message;
+import javax.jms.MessageConsumer;
+import javax.jms.Session;
+import javax.naming.Context;
+
+import junit.framework.Assert;
+
+import org.apache.qpid.disttest.DistributedTestConstants;
+import org.apache.qpid.disttest.jms.JmsMessageAdaptor;
+import org.apache.qpid.disttest.message.Command;
+import org.apache.qpid.disttest.message.CommandType;
+
+/**
+ * Helper for unit tests to simplify access to the Controller Queue.
+ *
+ * Implicitly creates the queue, so you must create a {@link ControllerQueue} object before
+ * trying to use the underlying queue.
+ */
+public class ControllerQueue
+{
+    private MessageConsumer _controllerQueueMessageConsumer;
+    private Session _controllerQueueSession;
+
+    /**
+     * Implicitly creates the queue, so you must create a {@link ControllerQueue} object before
+     * trying to use the underlying queue.
+     *
+     * @param context used for looking up the controller queue {@link Destination}
+     */
+    public ControllerQueue(Connection connection, Context context) throws Exception
+    {
+        _controllerQueueSession = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
+        Destination controllerQueue = (Destination) context.lookup(DistributedTestConstants.CONTROLLER_QUEUE_JNDI_NAME);
+        _controllerQueueMessageConsumer = _controllerQueueSession.createConsumer(controllerQueue);
+    }
+
+    public <T extends Command> T getNext(long timeout) throws JMSException
+    {
+        final Message message = _controllerQueueMessageConsumer.receive(timeout);
+        if(message == null)
+        {
+            return null;
+        }
+
+        return (T) JmsMessageAdaptor.messageToCommand(message);
+    }
+
+    public void addNextResponse(Map<CommandType, Command> responses) throws JMSException
+    {
+        Command nextResponse = getNext();
+        responses.put(nextResponse.getType(), nextResponse);
+    }
+
+    @SuppressWarnings("unchecked")
+    public <T extends Command> T getNext() throws JMSException
+    {
+        return (T)getNext(true);
+    }
+
+    public <T extends Command> T getNext(boolean assertMessageExists) throws JMSException
+    {
+        final Message message = _controllerQueueMessageConsumer.receive(1000);
+        if(assertMessageExists)
+        {
+            Assert.assertNotNull("No message received from control queue", message);
+        }
+
+        if(message == null)
+        {
+            return null;
+        }
+
+        T command = (T) JmsMessageAdaptor.messageToCommand(message);
+
+        return command;
+    }
+
+    public void close() throws Exception
+    {
+        _controllerQueueMessageConsumer.close();
+        _controllerQueueSession.close();
+    }
+}
diff --git a/qpid/java/perftests/src/test/java/org/apache/qpid/systest/disttest/clientonly/DistributedClientTest.java b/qpid/java/perftests/src/test/java/org/apache/qpid/systest/disttest/clientonly/DistributedClientTest.java
new file mode 100644
index 0000000000..4a872a7ee2
--- /dev/null
+++ b/qpid/java/perftests/src/test/java/org/apache/qpid/systest/disttest/clientonly/DistributedClientTest.java
@@ -0,0 +1,323 @@
+/*
+ * 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.systest.disttest.clientonly;
+
+import static org.apache.qpid.disttest.client.ClientState.READY;
+import static org.apache.qpid.disttest.client.ClientState.RUNNING_TEST;
+
+import java.util.HashMap;
+import java.util.Map;
+
+import javax.jms.Destination;
+import javax.jms.JMSException;
+import javax.jms.Message;
+import javax.jms.MessageProducer;
+import javax.jms.Queue;
+import javax.jms.Session;
+
+import org.apache.qpid.disttest.client.Client;
+import org.apache.qpid.disttest.client.ClientState;
+import org.apache.qpid.disttest.jms.ClientJmsDelegate;
+import org.apache.qpid.disttest.jms.JmsMessageAdaptor;
+import org.apache.qpid.disttest.message.Command;
+import org.apache.qpid.disttest.message.CommandType;
+import org.apache.qpid.disttest.message.CreateConnectionCommand;
+import org.apache.qpid.disttest.message.CreateConsumerCommand;
+import org.apache.qpid.disttest.message.CreateProducerCommand;
+import org.apache.qpid.disttest.message.CreateSessionCommand;
+import org.apache.qpid.disttest.message.ParticipantResult;
+import org.apache.qpid.disttest.message.RegisterClientCommand;
+import org.apache.qpid.disttest.message.Response;
+import org.apache.qpid.disttest.message.StartTestCommand;
+import org.apache.qpid.disttest.message.TearDownTestCommand;
+import org.apache.qpid.systest.disttest.DistributedTestSystemTestBase;
+
+public class DistributedClientTest extends DistributedTestSystemTestBase
+{
+    private static final String TEST_CONSUMER = "newTestConsumer";
+    private static final String TEST_DESTINATION = "newDestination";
+    private static final String TEST_PRODUCER_NAME = "newTestProducer";
+    private static final String TEST_SESSION_NAME = "newTestSession";
+    private static final String TEST_CONNECTION_NAME = "newTestConnection";
+
+    private Session _session = null;
+    private MessageProducer _clientQueueProducer;
+    private Client _client;
+    private ControllerQueue _controllerQueue;
+    protected ClientJmsDelegate _clientJmsDelegate;
+
+    @Override
+    protected void setUp() throws Exception
+    {
+        super.setUp();
+
+        _session = _connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
+        _controllerQueue = new ControllerQueue(_connection, _context);
+
+        _clientJmsDelegate = new ClientJmsDelegate(_context);
+        _client = new Client(_clientJmsDelegate);
+        _client.start();
+
+        final RegisterClientCommand registrationCommand = _controllerQueue.getNext();
+        createClientQueueProducer(registrationCommand);
+
+        createTestConnection(TEST_CONNECTION_NAME);
+        createTestSession(TEST_CONNECTION_NAME, TEST_SESSION_NAME);
+
+        assertEquals("Expected no test producers at start of test", 0, _clientJmsDelegate.getNoOfTestProducers());
+        assertEquals("Expected no test consumers at start of test", 0, _clientJmsDelegate.getNoOfTestConsumers());
+    }
+
+    @Override
+    protected void tearDown() throws Exception
+    {
+        try
+        {
+            _controllerQueue.close();
+            if (_session != null)
+            {
+                _session.close();
+            }
+        }
+        finally
+        {
+            super.tearDown();
+        }
+    }
+
+    public void testClientCanCreateTestProducer() throws Exception
+    {
+        assertEquals("Should initially have zero producers", 0, _clientJmsDelegate.getNoOfTestProducers());
+
+        createTestProducer(TEST_SESSION_NAME, TEST_PRODUCER_NAME, TEST_DESTINATION);
+
+        assertEquals("Should now have one test producer", 1, _clientJmsDelegate.getNoOfTestProducers());
+    }
+
+    public void testClientCanCreateTestConsumer() throws Exception
+    {
+        assertEquals("Should initially have no test consumers", 0, _clientJmsDelegate.getNoOfTestConsumers());
+
+        createTestConsumer(TEST_SESSION_NAME, TEST_CONSUMER, TEST_DESTINATION);
+
+        assertEquals("Should now have one test consumer", 1, _clientJmsDelegate.getNoOfTestConsumers());
+    }
+
+    public void testClientFailsToCreateSessionUsingInvalidConnection() throws Exception
+    {
+        int initialNoOfTestSessions = _clientJmsDelegate.getNoOfTestSessions();
+
+        createTestSession("nonExistentConnection", TEST_SESSION_NAME, false /* shouldSucceed */);
+
+        assertEquals("Number of test sessions should not have changed", initialNoOfTestSessions, _clientJmsDelegate.getNoOfTestSessions());
+    }
+
+    public void testClientFailsToCreateProducerUsingInvalidSession() throws Exception
+    {
+        int initialNoOfTestProducers = _clientJmsDelegate.getNoOfTestProducers();
+
+        createTestProducer("invalidSessionName", TEST_PRODUCER_NAME, TEST_DESTINATION, false /* shouldSucceed */);
+
+        assertEquals("Number of test producers should not have changed", initialNoOfTestProducers, _clientJmsDelegate.getNoOfTestProducers());
+    }
+
+    public void testClientFailsToCreateConsumerUsingInvalidSession() throws Exception
+    {
+        int initialNoOfTestConsumers = _clientJmsDelegate.getNoOfTestConsumers();
+
+        createTestConsumer("invalidSessionName", TEST_CONSUMER, TEST_DESTINATION, false /* shouldSucceed */);
+
+        assertEquals("Number of test consumers should not have changed", initialNoOfTestConsumers, _clientJmsDelegate.getNoOfTestConsumers());
+    }
+
+    public void testClientCanStartPerformingTests() throws Exception
+    {
+        createTestProducer(TEST_SESSION_NAME, TEST_PRODUCER_NAME, TEST_DESTINATION);
+
+        sendCommandToClient(new StartTestCommand());
+
+        validateStartTestResponseAndParticipantResults(CommandType.PRODUCER_PARTICIPANT_RESULT);
+
+        assertState(_client, RUNNING_TEST);
+    }
+
+    public void testParticipantsSendResults() throws JMSException
+    {
+        createTestProducer(TEST_SESSION_NAME, TEST_PRODUCER_NAME, TEST_DESTINATION);
+
+        sendCommandToClient(new StartTestCommand());
+
+        validateStartTestResponseAndParticipantResults(CommandType.PRODUCER_PARTICIPANT_RESULT);
+    }
+
+    /**
+     * Need to validate both of these responses together because their order is non-deterministic
+     * @param expectedParticipantResultCommandType TODO
+     */
+    private void validateStartTestResponseAndParticipantResults(CommandType expectedParticipantResultCommandType) throws JMSException
+    {
+        Map<CommandType, Command> responses = new HashMap<CommandType, Command>();
+        _controllerQueue.addNextResponse(responses);
+        _controllerQueue.addNextResponse(responses);
+
+        ParticipantResult results = (ParticipantResult) responses.get(expectedParticipantResultCommandType);
+        validateResponse(null, results, true);
+
+        Response startTestResponse = (Response) responses.get(CommandType.RESPONSE);
+        validateResponse(CommandType.START_TEST, startTestResponse, true);
+    }
+
+    public void testClientCannotStartPerformingTestsInNonReadyState() throws Exception
+    {
+        assertState(_client, READY);
+        sendCommandAndValidateResponse(new StartTestCommand(), true);
+        assertState(_client, RUNNING_TEST);
+
+        // Send another start test command
+        sendCommandAndValidateResponse(new StartTestCommand(), false /*should reject duplicate start command*/);
+        assertState(_client, RUNNING_TEST);
+    }
+
+    public void testNonRunningClientIsUnaffectedByStopTestCommand() throws Exception
+    {
+        assertState(_client, READY);
+
+        sendCommandAndValidateResponse(new TearDownTestCommand(), false);
+
+        assertState(_client, READY);
+    }
+
+    private void sendCommandToClient(final Command command) throws JMSException
+    {
+        final Message message = JmsMessageAdaptor.commandToMessage(_session, command);
+        _clientQueueProducer.send(message);
+    }
+
+    private void sendCommandAndValidateResponse(final Command command, boolean shouldSucceed) throws JMSException
+    {
+        sendCommandToClient(command);
+        Response response = _controllerQueue.getNext();
+        validateResponse(command.getType(), response, shouldSucceed);
+    }
+
+    private void sendCommandAndValidateResponse(final Command command) throws JMSException
+    {
+        sendCommandAndValidateResponse(command, true);
+    }
+
+    private void createTestConnection(String connectionName) throws Exception
+    {
+        int initialNumberOfConnections = _clientJmsDelegate.getNoOfTestConnections();
+
+        final CreateConnectionCommand createConnectionCommand = new CreateConnectionCommand();
+        createConnectionCommand.setConnectionName(connectionName);
+        createConnectionCommand.setConnectionFactoryName("connectionfactory");
+
+        sendCommandAndValidateResponse(createConnectionCommand);
+
+        int expectedNumberOfConnections = initialNumberOfConnections + 1;
+
+        assertEquals("unexpected number of test connections", expectedNumberOfConnections, _clientJmsDelegate.getNoOfTestConnections());
+    }
+
+    private void createTestSession(String connectionName, String sessionName, boolean shouldSucceed) throws Exception
+    {
+        int initialNumberOfSessions = _clientJmsDelegate.getNoOfTestSessions();
+
+        final CreateSessionCommand createSessionCommand = new CreateSessionCommand();
+        createSessionCommand.setConnectionName(connectionName);
+        createSessionCommand.setSessionName(sessionName);
+        createSessionCommand.setAcknowledgeMode(Session.AUTO_ACKNOWLEDGE);
+
+        sendCommandAndValidateResponse(createSessionCommand, shouldSucceed);
+
+        int expectedNumberOfSessions = initialNumberOfSessions + (shouldSucceed ? 1 : 0);
+
+        assertEquals("unexpected number of test sessions", expectedNumberOfSessions, _clientJmsDelegate.getNoOfTestSessions());
+    }
+
+    private void createTestSession(String connectionName, String sessionName) throws Exception
+    {
+        createTestSession(connectionName, sessionName, true);
+    }
+
+    private void createTestProducer(String sessionName, String producerName, String destinationName, boolean shouldSucceed) throws JMSException
+    {
+        final CreateProducerCommand createProducerCommand = new CreateProducerCommand();
+        createProducerCommand.setParticipantName(producerName);
+        createProducerCommand.setSessionName(sessionName);
+        createProducerCommand.setDestinationName(destinationName);
+        createProducerCommand.setNumberOfMessages(100);
+
+        sendCommandAndValidateResponse(createProducerCommand, shouldSucceed);
+    }
+
+    private void createTestProducer(String sessionName, String producerName, String destinationName) throws JMSException
+    {
+        createTestProducer(sessionName, producerName, destinationName, true);
+    }
+
+    private void createTestConsumer(String sessionName, String consumerName, String destinationName, boolean shouldSucceed) throws JMSException
+    {
+        final CreateConsumerCommand createConsumerCommand = new CreateConsumerCommand();
+        createConsumerCommand.setSessionName(sessionName);
+        createConsumerCommand.setDestinationName(destinationName);
+        createConsumerCommand.setParticipantName(consumerName);
+        createConsumerCommand.setNumberOfMessages(1);
+
+        sendCommandAndValidateResponse(createConsumerCommand, shouldSucceed);
+    }
+
+    private void createTestConsumer(String sessionName, String consumerName, String destinationName) throws JMSException
+    {
+        createTestConsumer(sessionName, consumerName, destinationName, true);
+    }
+
+    private void validateResponse(CommandType originatingCommandType, Response response, boolean shouldSucceed) throws JMSException
+    {
+        assertEquals("Response is a reply to the wrong command: " + response,
+                originatingCommandType,
+                response.getInReplyToCommandType());
+
+        boolean shouldHaveError = !shouldSucceed;
+        assertEquals("Response message " + response + " should have indicated hasError=" + shouldHaveError,
+                shouldHaveError,
+                response.hasError());
+    }
+
+    private void createClientQueueProducer(final RegisterClientCommand registration) throws JMSException
+    {
+        final Destination clientCommandQueue = createDestinationFromRegistration(registration);
+        _clientQueueProducer = _session.createProducer(clientCommandQueue);
+    }
+
+    private Queue createDestinationFromRegistration(final RegisterClientCommand registrationCommand) throws JMSException
+    {
+        String clientQueueName = registrationCommand.getClientQueueName();
+        assertNotNull("Null client queue in register message", clientQueueName);
+        return _session.createQueue(clientQueueName);
+    }
+
+    private static void assertState(Client client, ClientState expectedState)
+    {
+        ClientState clientState = client.getState();
+        assertEquals("Client should be in state: " + expectedState + " but is in state " + clientState, expectedState, clientState);
+    }
+}
diff --git a/qpid/java/perftests/src/test/java/org/apache/qpid/systest/disttest/clientonly/MessageProviderTest.java b/qpid/java/perftests/src/test/java/org/apache/qpid/systest/disttest/clientonly/MessageProviderTest.java
new file mode 100644
index 0000000000..dcbff6518b
--- /dev/null
+++ b/qpid/java/perftests/src/test/java/org/apache/qpid/systest/disttest/clientonly/MessageProviderTest.java
@@ -0,0 +1,119 @@
+/*
+ * 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.systest.disttest.clientonly;
+
+import java.util.HashMap;
+import java.util.Map;
+
+import javax.jms.DeliveryMode;
+import javax.jms.JMSException;
+import javax.jms.Message;
+import javax.jms.MessageConsumer;
+import javax.jms.Session;
+import javax.jms.TextMessage;
+
+import org.apache.qpid.disttest.client.MessageProvider;
+import org.apache.qpid.disttest.client.property.PropertyValue;
+import org.apache.qpid.disttest.client.property.SimplePropertyValue;
+import org.apache.qpid.disttest.message.CreateMessageProviderCommand;
+import org.apache.qpid.disttest.message.CreateProducerCommand;
+import org.apache.qpid.systest.disttest.DistributedTestSystemTestBase;
+import org.apache.qpid.systest.disttest.clientonly.ProducerParticipantTest.TestClientJmsDelegate;
+
+public class MessageProviderTest extends DistributedTestSystemTestBase
+{
+    private MessageConsumer _consumer;
+    private Session _session;
+    private TestClientJmsDelegate _delegate;
+
+    @Override
+    protected void setUp() throws Exception
+    {
+        super.setUp();
+        _session = _connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
+        _consumer = _session.createConsumer(getTestQueue());
+        _delegate = new TestClientJmsDelegate(getContext());
+    }
+
+    public void testMessageSize() throws Exception
+    {
+        runSizeTest(0);
+        runSizeTest(5);
+        runSizeTest(512);
+    }
+
+    public void runSizeTest(int size) throws Exception
+    {
+        CreateProducerCommand command = new CreateProducerCommand();
+        command.setMessageSize(size);
+        MessageProvider messageProvider = new MessageProvider(null);
+        Message message = messageProvider.nextMessage(_session, command);
+        assertNotNull("Message is not generated", message);
+        assertTrue("Wrong message type", message instanceof TextMessage);
+        TextMessage textMessage = (TextMessage)message;
+        String text = textMessage.getText();
+        assertNotNull("Message payload is not generated", text);
+        assertEquals("Message payload size is incorrect", size, text.length());
+    }
+
+    public void testCreateMessageProviderAndSendMessage() throws Exception
+    {
+        final CreateMessageProviderCommand messageProviderCommand = new CreateMessageProviderCommand();
+        messageProviderCommand.setProviderName("test1");
+        Map<String, PropertyValue> messageProperties = new HashMap<String, PropertyValue>();
+        messageProperties.put("test", new SimplePropertyValue("testValue"));
+        messageProperties.put("priority", new SimplePropertyValue(new Integer(9)));
+        messageProviderCommand.setMessageProperties(messageProperties);
+        _delegate.createMessageProvider(messageProviderCommand);
+
+        final CreateProducerCommand producerCommand = new CreateProducerCommand();
+        producerCommand.setNumberOfMessages(1);
+        producerCommand.setDeliveryMode(DeliveryMode.PERSISTENT);
+        producerCommand.setPriority(6);
+        producerCommand.setParticipantName("test");
+        producerCommand.setMessageSize(10);
+        producerCommand.setSessionName("testSession");
+        producerCommand.setDestinationName(getTestQueueName());
+        producerCommand.setMessageProviderName(messageProviderCommand.getProviderName());
+
+        Session session = _connection.createSession(true, Session.SESSION_TRANSACTED);
+        _delegate.addConnection("name-does-not-matter", _connection);
+        _delegate.addSession(producerCommand.getSessionName(), session);
+        _delegate.createProducer(producerCommand);
+
+        Message message = _delegate.sendNextMessage(producerCommand);
+        session.commit();
+        assertMessage(message);
+
+        _connection.start();
+        Message receivedMessage = _consumer.receive(1000l);
+        assertMessage(receivedMessage);
+     }
+
+    protected void assertMessage(Message message) throws JMSException
+    {
+        assertNotNull("Message should not be null", message);
+        assertEquals("Unexpected test property", "testValue", message.getStringProperty("test"));
+        assertEquals("Unexpected priority property", 9, message.getJMSPriority());
+        assertTrue("Unexpected message type", message instanceof TextMessage);
+        String text = ((TextMessage)message).getText();
+        assertNotNull("Message text should not be null", text);
+        assertNotNull("Unexpected message size ", text.length());
+    }
+}
diff --git a/qpid/java/perftests/src/test/java/org/apache/qpid/systest/disttest/clientonly/ProducerParticipantTest.java b/qpid/java/perftests/src/test/java/org/apache/qpid/systest/disttest/clientonly/ProducerParticipantTest.java
new file mode 100644
index 0000000000..54bb9efa98
--- /dev/null
+++ b/qpid/java/perftests/src/test/java/org/apache/qpid/systest/disttest/clientonly/ProducerParticipantTest.java
@@ -0,0 +1,132 @@
+/*
+ * 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.systest.disttest.clientonly;
+
+import javax.jms.Connection;
+import javax.jms.DeliveryMode;
+import javax.jms.Message;
+import javax.jms.MessageConsumer;
+import javax.jms.Session;
+import javax.jms.TextMessage;
+import javax.naming.Context;
+
+import org.apache.qpid.disttest.client.Client;
+import org.apache.qpid.disttest.client.ParticipantExecutor;
+import org.apache.qpid.disttest.client.ProducerParticipant;
+import org.apache.qpid.disttest.jms.ClientJmsDelegate;
+import org.apache.qpid.disttest.message.CreateProducerCommand;
+import org.apache.qpid.disttest.message.ParticipantResult;
+import org.apache.qpid.systest.disttest.DistributedTestSystemTestBase;
+
+public class ProducerParticipantTest extends DistributedTestSystemTestBase
+{
+    private MessageConsumer _consumer;
+    private TestClientJmsDelegate _delegate;
+    private Client _client;
+    private ControllerQueue _controllerQueue;
+
+    @Override
+    protected void setUp() throws Exception
+    {
+        super.setUp();
+
+        _controllerQueue = new ControllerQueue(_connection, _context);
+        Session session = _connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
+        _consumer = session.createConsumer(getTestQueue());
+
+        _delegate = new TestClientJmsDelegate(getContext());
+        _client = new Client(_delegate);
+    }
+
+
+
+    @Override
+    protected void tearDown() throws Exception
+    {
+        _controllerQueue.close();
+        super.tearDown();
+    }
+
+
+
+    public void testProduceNumberOfMessages() throws Exception
+    {
+        runTest(Session.AUTO_ACKNOWLEDGE, 100, 10, 0, 0);
+    }
+
+    protected void runTest(int acknowledgeMode, int messageSize, int numberOfMessages, int batchSize, long publishInterval) throws Exception
+    {
+        final CreateProducerCommand command = new CreateProducerCommand();
+        command.setNumberOfMessages(numberOfMessages);
+        command.setDeliveryMode(DeliveryMode.PERSISTENT);
+        command.setParticipantName("test");
+        command.setMessageSize(messageSize);
+        command.setBatchSize(batchSize);
+        command.setInterval(publishInterval);
+        command.setSessionName("testSession");
+        command.setDestinationName(getTestQueueName());
+
+        Session session = _connection.createSession(Session.SESSION_TRANSACTED == acknowledgeMode, acknowledgeMode);
+
+        _delegate.addConnection("name-does-not-matter", _connection);
+        _delegate.addSession(command.getSessionName(), session);
+        _delegate.createProducer(command);
+
+        final ProducerParticipant producer = new ProducerParticipant(_delegate, command);
+
+        new ParticipantExecutor(producer).start(_client);
+
+        _connection.start();
+        for (int i = 0; i < numberOfMessages; i++)
+        {
+            final Message m = _consumer.receive(1000l);
+            assertNotNull("Expected message [" + i + "] is not received", m);
+            assertTrue("Unexpected message", m instanceof TextMessage);
+        }
+        Message m = _consumer.receive(500l);
+        assertNull("Unexpected message", m);
+
+        ParticipantResult results = _controllerQueue.getNext();
+
+        assertNotNull("no results", results);
+        assertFalse(results.getStartInMillis() == 0);
+        assertFalse(results.getEndInMillis() == 0);
+    }
+
+    static class TestClientJmsDelegate extends ClientJmsDelegate
+    {
+
+        public TestClientJmsDelegate(Context context)
+        {
+            super(context);
+        }
+
+        @Override
+        public void addSession(final String sessionName, final Session newSession)
+        {
+            super.addSession(sessionName, newSession);
+        }
+
+        @Override
+        public void addConnection(final String connectionName, final Connection newConnection)
+        {
+            super.addConnection(connectionName, newConnection);
+        }
+    }
+}
diff --git a/qpid/java/perftests/src/test/java/org/apache/qpid/systest/disttest/controllerandclient/ControllerAndClientTest.java b/qpid/java/perftests/src/test/java/org/apache/qpid/systest/disttest/controllerandclient/ControllerAndClientTest.java
new file mode 100644
index 0000000000..e8790ea7f6
--- /dev/null
+++ b/qpid/java/perftests/src/test/java/org/apache/qpid/systest/disttest/controllerandclient/ControllerAndClientTest.java
@@ -0,0 +1,237 @@
+/*
+ * 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.systest.disttest.controllerandclient;
+
+import static org.apache.qpid.systest.disttest.SystemTestConstants.COMMAND_RESPONSE_TIMEOUT;
+import static org.apache.qpid.systest.disttest.SystemTestConstants.REGISTRATION_TIMEOUT;
+import static org.apache.qpid.systest.disttest.SystemTestConstants.TEST_RESULT_TIMEOUT;
+
+import java.util.List;
+
+import javax.jms.Message;
+import javax.jms.MessageConsumer;
+import javax.jms.Queue;
+import javax.jms.Session;
+import javax.naming.NamingException;
+
+import org.apache.qpid.disttest.ConfigFileHelper;
+import org.apache.qpid.disttest.client.Client;
+import org.apache.qpid.disttest.client.ClientState;
+import org.apache.qpid.disttest.controller.Controller;
+import org.apache.qpid.disttest.controller.ResultsForAllTests;
+import org.apache.qpid.disttest.controller.TestResult;
+import org.apache.qpid.disttest.controller.config.Config;
+import org.apache.qpid.disttest.jms.ClientJmsDelegate;
+import org.apache.qpid.disttest.jms.ControllerJmsDelegate;
+import org.apache.qpid.disttest.message.ConsumerParticipantResult;
+import org.apache.qpid.disttest.message.ParticipantResult;
+import org.apache.qpid.disttest.message.ProducerParticipantResult;
+import org.apache.qpid.disttest.results.aggregation.ITestResult;
+import org.apache.qpid.systest.disttest.DistributedTestSystemTestBase;
+
+public class ControllerAndClientTest extends DistributedTestSystemTestBase
+{
+    private Controller _controller;
+
+    @Override
+    protected void setUp() throws Exception
+    {
+        super.setUp();
+
+        _controller = new Controller(new ControllerJmsDelegate(_context), REGISTRATION_TIMEOUT, COMMAND_RESPONSE_TIMEOUT);
+        _controller.setTestResultTimeout(TEST_RESULT_TIMEOUT);
+    }
+
+    public void testProducerAndConsumerInSeparateClients() throws Exception
+    {
+        List<TestResult> resultList = runTestsForTwoClients("producerAndConsumerInSeparateClients.json", 1);
+
+        TestResult testResult1 = resultList.get(0);
+        assertEquals("Unexpected test name", "Test 1", testResult1.getName());
+        List<ParticipantResult> test1ParticipantResults = testResult1.getParticipantResults();
+        assertEquals("Unexpected number of participant results for test 1", 2, test1ParticipantResults.size());
+        assertParticipantNames(test1ParticipantResults, "participantConsumer1", "participantProducer1");
+    }
+
+    public void testProducerClient() throws Exception
+    {
+        Session session = _connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
+        Queue queue = session.createQueue("producerClient");
+        MessageConsumer consumer = session.createConsumer(queue);
+
+        // queue is not declared in configuration
+        // controller is not able to clean it
+        // cleaning manually
+        while(consumer.receive(1000l) != null);
+
+        final Config config = ConfigFileHelper.getConfigFromResource(getClass(), "produceClient.json");
+        _controller.setConfig(config);
+        final Client client1 = new Client(new ClientJmsDelegate(_context));
+        final Thread client1Thread = createBackgroundClientThread(client1);
+        _controller.awaitClientRegistrations();
+
+        ResultsForAllTests results = _controller.runAllTests();
+        _controller.stopAllRegisteredClients();
+
+        assertClientThreadsShutdown(client1Thread);
+        assertClientsStopped(ClientState.STOPPED, client1);
+        assertFalse("Test should have no errors", results.hasErrors());
+        List<ITestResult> allTestResults = results.getTestResults();
+        assertEquals("Unexpected number of test results", 1, allTestResults.size());
+        ITestResult testResult1 = allTestResults.get(0);
+        assertEquals("Unexpected test name", "Test 1", testResult1.getName());
+        List<ParticipantResult> test1ParticipantResults = testResult1.getParticipantResults();
+        assertEquals("Unexpected number of participant results for test 1", 1, test1ParticipantResults.size());
+        assertParticipantNames(test1ParticipantResults, "participantProducer1");
+
+        // check message properties
+        for (int i=0; i< 10; i++)
+        {
+            Message message = consumer.receive(1000l);
+            assertNotNull("Message " + i + " is not received", message);
+            assertEquals("Unexpected priority", i, message.getJMSPriority());
+            assertEquals("Unexpected id", i, message.getIntProperty("id"));
+            assertEquals("Unexpected test", "test-value", message.getStringProperty("test"));
+        }
+    }
+
+    public void testProducerAndThreeConsumersInSeparateClients() throws Exception
+    {
+        List<TestResult> resultList = runTestsForTwoClients("producerAndThreeConsumersInSeparateClients.json", 1);
+
+        TestResult testResult1 = resultList.get(0);
+        List<ParticipantResult> test1ParticipantResults = testResult1.getParticipantResults();
+        assertEquals("Unexpected number of participant results for test", 4, test1ParticipantResults.size());
+
+        assertParticipantNames(test1ParticipantResults, "participantConsumer1", "participantConsumer2", "participantConsumer3", "participantProducer1");
+
+        ConsumerParticipantResult consumer1 = (ConsumerParticipantResult) test1ParticipantResults.get(0);
+        assertEquals(3, consumer1.getNumberOfMessagesProcessed());
+        assertEquals(true, consumer1.isSynchronousConsumer());
+
+        ProducerParticipantResult producer1 = (ProducerParticipantResult) test1ParticipantResults.get(3);
+        assertEquals(9, producer1.getNumberOfMessagesProcessed());
+        assertEquals(2, producer1.getBatchSize());
+        assertEquals(50, producer1.getInterval());
+    }
+
+    public void testIteratingFeature() throws Exception
+    {
+        List<TestResult> resultList = runTestsForTwoClients("iteratingFeature.json", 2);
+
+        assertTestResultMessageSize(resultList.get(0), 0, 100, 10);
+        assertTestResultMessageSize(resultList.get(1), 1, 200, 5);
+
+    }
+
+    private void assertTestResultMessageSize(TestResult testResult, int iterationNumber, int expectedMessageSize, int expectedNumberOfMessages)
+    {
+        List<ParticipantResult> test1ParticipantResults = testResult.getParticipantResults();
+        assertEquals("Unexpected number of participant results for test", 2, test1ParticipantResults.size());
+
+        ParticipantResult producer1 = (ParticipantResult) test1ParticipantResults.get(1);
+
+        assertEquals(expectedMessageSize, producer1.getPayloadSize());
+        assertEquals(iterationNumber, producer1.getIterationNumber());
+    }
+
+    public void testTwoTests() throws Exception
+    {
+        List<TestResult> resultList = runTestsForTwoClients("testWithTwoTests.json", 2);
+
+        assertEquals("Test 1", resultList.get(0).getName());
+        assertEquals("Test 2", resultList.get(1).getName());
+    }
+
+    private List<TestResult> runTestsForTwoClients(String jsonConfigFile, int expectedNumberOfTests) throws NamingException, InterruptedException
+    {
+        final Config config = ConfigFileHelper.getConfigFromResource(getClass(), jsonConfigFile);
+        _controller.setConfig(config);
+
+        final Client client1 = new Client(new ClientJmsDelegate(_context));
+        final Client client2 = new Client(new ClientJmsDelegate(_context));
+
+        final Thread client1Thread = createBackgroundClientThread(client1);
+        final Thread client2Thread = createBackgroundClientThread(client2);
+
+        _controller.awaitClientRegistrations();
+
+        ResultsForAllTests results = _controller.runAllTests();
+        _controller.stopAllRegisteredClients();
+
+        assertClientThreadsShutdown(client1Thread, client2Thread);
+        assertClientsStopped(ClientState.STOPPED, client1, client2);
+
+        assertFalse("Test should have no errors", results.hasErrors());
+
+        List<TestResult> allTestResults = (List)results.getTestResults();
+        assertEquals("Unexpected number of test results", expectedNumberOfTests, allTestResults.size());
+
+        return allTestResults;
+    }
+
+
+    private void assertParticipantNames(List<ParticipantResult> participants, String... expectedOrderedParticipantNames)
+    {
+        assertEquals("Size of list of expected participant names is different from actual", expectedOrderedParticipantNames.length, participants.size());
+
+        for (int i = 0; i < expectedOrderedParticipantNames.length; i++)
+        {
+            String expectedParticipantName = expectedOrderedParticipantNames[i];
+            ParticipantResult participant = participants.get(i);
+            assertEquals(expectedParticipantName, participant.getParticipantName());
+        }
+    }
+
+    private void assertClientsStopped(ClientState expectedState, final Client... clients)
+    {
+        for (Client client : clients)
+        {
+            assertEquals(client.getClientName() + " in unexpected state", expectedState, client.getState());
+        }
+    }
+
+    private void assertClientThreadsShutdown(final Thread... clientThreads)
+            throws InterruptedException
+    {
+        for (Thread clientThread : clientThreads)
+        {
+            clientThread.join(2000);
+            assertFalse(clientThread.getName() + " should have shutdown", clientThread.isAlive());
+        }
+    }
+
+    private Thread createBackgroundClientThread(final Client client) throws NamingException
+    {
+        final String clientThreadName = client.getClientName() + "-thread";
+        final Thread clientThread = new Thread(new Runnable()
+        {
+            @Override
+            public void run()
+            {
+                client.start();
+                client.waitUntilStopped(1000);
+            }
+        }, clientThreadName);
+        clientThread.start();
+        return clientThread;
+    }
+
+}
diff --git a/qpid/java/perftests/src/test/java/org/apache/qpid/systest/disttest/controllerandclient/iteratingFeature.json b/qpid/java/perftests/src/test/java/org/apache/qpid/systest/disttest/controllerandclient/iteratingFeature.json
new file mode 100644
index 0000000000..89123302b7
--- /dev/null
+++ b/qpid/java/perftests/src/test/java/org/apache/qpid/systest/disttest/controllerandclient/iteratingFeature.json
@@ -0,0 +1,63 @@
+{
+    "_tests":[
+    {
+      "_name": "Test iteration feature",
+      "_iterations":[ 
+        {
+          "_messageSize": 100,
+          "_numberOfMessages": 10
+        },
+        {
+          "_messageSize": 200,
+          "_numberOfMessages": 5
+        }
+      ],
+      "_queues":[
+        {
+          "_name": "direct://amq.direct//testQueue"
+        }
+      ],
+      "_clients":[
+        {
+          "_name": "producingClient",
+          "_connections":[
+            {
+              "_name": "connection1",
+              "_factory": "connectionfactory",
+              "_sessions": [
+                {
+                  "_sessionName": "session1",
+                  "_producers": [
+                    {
+                      "_name": "participantProducer1",
+                      "_destinationName": "direct://amq.direct//testQueue"
+                    }
+                  ]
+                }
+              ]
+            }
+          ]
+        },
+        {
+          "_name": "consumingClient",
+          "_connections":[
+            {
+              "_name": "connection1",
+              "_factory": "connectionfactory",
+              "_sessions": [
+                {
+                  "_sessionName": "session1",
+                  "_consumers": [
+                    {
+                      "_name": "participantConsumer1",
+                      "_destinationName": "direct://amq.direct//testQueue"
+                    }
+                  ]
+                }
+              ]
+            }
+          ]
+        }
+      ]
+    }]
+}
\ No newline at end of file
diff --git a/qpid/java/perftests/src/test/java/org/apache/qpid/systest/disttest/controllerandclient/produceClient.json b/qpid/java/perftests/src/test/java/org/apache/qpid/systest/disttest/controllerandclient/produceClient.json
new file mode 100644
index 0000000000..605e5cb585
--- /dev/null
+++ b/qpid/java/perftests/src/test/java/org/apache/qpid/systest/disttest/controllerandclient/produceClient.json
@@ -0,0 +1,41 @@
+{
+    "_tests":[
+        {
+            "_name": "Test 1";
+            "_clients":[
+                {
+                    "_name": "producingClient",
+                    "_connections":[
+                        {
+                            "_name": "connection1",
+                            "_factory": "connectionfactory",
+                            "_sessions": [
+                                {
+                                    "_sessionName": "session1",
+                                    "_producers": [
+                                        {
+                                            "_name": "participantProducer1",
+                                            "_destinationName": "direct://amq.direct//producerClient",
+                                            "_numberOfMessages": 10;
+                                            "_messageProviderName": "testProvider1"
+                                        }
+                                    ]
+                                }
+                            ]
+                        }
+                    ];
+                    "_messageProviders":[
+                        {
+                            "_name": "testProvider1";
+                            "_messageProperties": {
+                                "priority": {"@def": "list"; "_items": [0,1,2,3,4,5,6,7,8,9]};
+                                "id": {"@def": "range"; "_upper": 10; "_type": "int"};
+                                "test": "test-value"
+                            }
+                        }
+                    ]
+                }
+            ]
+        }
+    ]
+}
\ No newline at end of file
diff --git a/qpid/java/perftests/src/test/java/org/apache/qpid/systest/disttest/controllerandclient/producerAndConsumerInSeparateClients.json b/qpid/java/perftests/src/test/java/org/apache/qpid/systest/disttest/controllerandclient/producerAndConsumerInSeparateClients.json
new file mode 100644
index 0000000000..8d210dce84
--- /dev/null
+++ b/qpid/java/perftests/src/test/java/org/apache/qpid/systest/disttest/controllerandclient/producerAndConsumerInSeparateClients.json
@@ -0,0 +1,55 @@
+{
+    "_tests":[
+    {
+      "_name": "Test 1";
+      "_queues":[
+        {
+          "_name": "direct://amq.direct//testQueue"
+        }
+      ];
+      "_clients":[
+        {
+          "_name": "producingClient",
+          "_connections":[
+            {
+              "_name": "connection1",
+              "_factory": "connectionfactory",
+              "_sessions": [
+                {
+                  "_sessionName": "session1",
+                  "_producers": [
+                    {
+                      "_name": "participantProducer1",
+                      "_destinationName": "direct://amq.direct//testQueue",
+                      "_numberOfMessages": 1
+                    }
+                  ]
+                }
+              ]
+            }
+          ]
+        },
+        {
+          "_name": "consumingClient",
+          "_connections":[
+            {
+              "_name": "connection1",
+              "_factory": "connectionfactory",
+              "_sessions": [
+                {
+                  "_sessionName": "session1",
+                  "_consumers": [
+                    {
+                      "_name": "participantConsumer1",
+                      "_destinationName": "direct://amq.direct//testQueue",
+                      "_numberOfMessages": 1
+                    }
+                  ]
+                }
+              ]
+            }
+          ]
+        }
+      ]
+    }]
+}
\ No newline at end of file
diff --git a/qpid/java/perftests/src/test/java/org/apache/qpid/systest/disttest/controllerandclient/producerAndThreeConsumersInSeparateClients.json b/qpid/java/perftests/src/test/java/org/apache/qpid/systest/disttest/controllerandclient/producerAndThreeConsumersInSeparateClients.json
new file mode 100644
index 0000000000..f94c4f0ae0
--- /dev/null
+++ b/qpid/java/perftests/src/test/java/org/apache/qpid/systest/disttest/controllerandclient/producerAndThreeConsumersInSeparateClients.json
@@ -0,0 +1,77 @@
+{
+    "_tests":[
+    {
+      "_name": "ProducerAndThreeConsumersInSeparateClients";
+      "_queues":[
+        {
+          "_name": "direct://amq.direct//testQueue"
+        }
+      ];
+      "_clients":[
+        {
+          "_name": "producingClient",
+          "_connections":[
+            {
+              "_name": "connection1",
+              "_factory": "connectionfactory",
+              "_sessions": [
+                {
+                  "_sessionName": "session1",
+                  "_producers": [
+                    {
+                      "_name": "participantProducer1",
+                      "_destinationName": "direct://amq.direct//testQueue",
+                      "_numberOfMessages": 9,
+                      "_batchSize": 2,
+                      "_interval": 50
+                    }
+                  ]
+                }
+              ]
+            }
+          ]
+        },
+        {
+          "_name": "consumingClient",
+          "_connections":[
+            {
+              "_name": "connection1",
+              "_factory": "connectionfactory",
+              "_sessions": [
+                {
+                  "_sessionName": "session1",
+                  "_consumers": [
+                    {
+                      "_name": "participantConsumer1",
+                      "_destinationName": "direct://amq.direct//testQueue",
+                      "_numberOfMessages": 3
+                    }
+                  ]
+                },
+                {
+                  "_sessionName": "session2",
+                  "_consumers": [
+                    {
+                      "_name": "participantConsumer2",
+                      "_destinationName": "direct://amq.direct//testQueue",
+                      "_numberOfMessages": 3
+                    }
+                  ]
+                },
+                {
+                  "_sessionName": "session3",
+                  "_consumers": [
+                    {
+                      "_name": "participantConsumer3",
+                      "_destinationName": "direct://amq.direct//testQueue",
+                      "_numberOfMessages": 3
+                    }
+                  ]
+                }
+              ]
+            }
+          ]
+        }
+      ]
+    }]
+}
\ No newline at end of file
diff --git a/qpid/java/perftests/src/test/java/org/apache/qpid/systest/disttest/controllerandclient/testWithTwoTests.json b/qpid/java/perftests/src/test/java/org/apache/qpid/systest/disttest/controllerandclient/testWithTwoTests.json
new file mode 100644
index 0000000000..4abd7f4feb
--- /dev/null
+++ b/qpid/java/perftests/src/test/java/org/apache/qpid/systest/disttest/controllerandclient/testWithTwoTests.json
@@ -0,0 +1,107 @@
+{
+    "_tests":[
+    {
+      "_name": "Test 1";
+      "_queues":[
+        {
+          "_name": "direct://amq.direct//testQueue"
+        }
+      ];
+      "_clients":[
+        {
+          "_name": "producingClient",
+          "_connections":[
+            {
+              "_name": "connection1",
+              "_factory": "connectionfactory",
+              "_sessions": [
+                {
+                  "_sessionName": "session1",
+                  "_producers": [
+                    {
+                      "_name": "participantProducer1",
+                      "_destinationName": "direct://amq.direct//testQueue",
+                      "_numberOfMessages": 1
+                    }
+                  ]
+                }
+              ]
+            }
+          ]
+        },
+        {
+          "_name": "consumingClient",
+          "_connections":[
+            {
+              "_name": "connection1",
+              "_factory": "connectionfactory",
+              "_sessions": [
+                {
+                  "_sessionName": "session1",
+                  "_consumers": [
+                    {
+                      "_name": "participantconsumer1",
+                      "_destinationName": "direct://amq.direct//testQueue",
+                      "_numberOfMessages": 1
+                    }
+                  ]
+                }
+              ]
+            }
+          ]
+        }
+      ]
+    },
+    {
+        "_name": "Test 2";
+        "_queues":[
+          {
+            "_name": "direct://amq.direct//testQueue2"
+          }
+        ];
+        "_clients":[
+          {
+            "_name": "producingClient",
+            "_connections":[
+              {
+                "_name": "connection1",
+                "_factory": "connectionfactory",
+                "_sessions": [
+                  {
+                    "_sessionName": "session1",
+                    "_producers": [
+                      {
+                        "_name": "participantProducer2",
+                        "_destinationName": "direct://amq.direct//testQueue2",
+                        "_numberOfMessages": 1
+                      }
+                    ]
+                  }
+                ]
+              }
+            ]
+          },
+          {
+            "_name": "consumingClient",
+            "_connections":[
+              {
+                "_name": "connection1",
+                "_factory": "connectionfactory",
+                "_sessions": [
+                  {
+                    "_sessionName": "session1",
+                    "_consumers": [
+                      {
+                        "_name": "participantConsumer2",
+                        "_destinationName": "direct://amq.direct//testQueue2",
+                        "_numberOfMessages": 1
+                      }
+                    ]
+                  }
+                ]
+              }
+            ]
+          }
+        ]
+      }]
+}
\ No newline at end of file
diff --git a/qpid/java/perftests/src/test/java/org/apache/qpid/systest/disttest/controlleronly/DistributedControllerTest.java b/qpid/java/perftests/src/test/java/org/apache/qpid/systest/disttest/controlleronly/DistributedControllerTest.java
new file mode 100644
index 0000000000..ad7f0e0682
--- /dev/null
+++ b/qpid/java/perftests/src/test/java/org/apache/qpid/systest/disttest/controlleronly/DistributedControllerTest.java
@@ -0,0 +1,157 @@
+/*
+ * 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.systest.disttest.controlleronly;
+
+import static org.apache.qpid.systest.disttest.SystemTestConstants.COMMAND_RESPONSE_TIMEOUT;
+import static org.apache.qpid.systest.disttest.SystemTestConstants.REGISTRATION_TIMEOUT;
+
+import java.util.concurrent.ArrayBlockingQueue;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicReference;
+
+import javax.jms.Connection;
+import javax.jms.ConnectionFactory;
+import javax.jms.Destination;
+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.TemporaryQueue;
+
+import org.apache.qpid.disttest.ConfigFileHelper;
+import org.apache.qpid.disttest.controller.Controller;
+import org.apache.qpid.disttest.controller.config.Config;
+import org.apache.qpid.disttest.jms.ControllerJmsDelegate;
+import org.apache.qpid.disttest.jms.JmsMessageAdaptor;
+import org.apache.qpid.disttest.message.Command;
+import org.apache.qpid.disttest.message.CommandType;
+import org.apache.qpid.disttest.message.RegisterClientCommand;
+import org.apache.qpid.disttest.message.Response;
+import org.apache.qpid.systest.disttest.DistributedTestSystemTestBase;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class DistributedControllerTest extends DistributedTestSystemTestBase
+{
+    private static final Logger LOGGER = LoggerFactory.getLogger(DistributedControllerTest.class);
+
+    private static final String CLIENT1 = "client1";
+    private Controller _controller = null;
+    private Session _session = null;
+    private Connection _connection = null;
+    private Destination _controllerQueue = null;
+    private TemporaryQueue _clientQueue = null;
+
+    @Override
+    protected void setUp() throws Exception
+    {
+        super.setUp();
+
+        _controllerQueue = (Destination) _context.lookup("controllerqueue");
+
+        final ConnectionFactory connectionFactory = (ConnectionFactory) _context.lookup("connectionfactory");
+        _connection = connectionFactory.createConnection();
+        _connection.start();
+        _session = _connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
+        _clientQueue = _session.createTemporaryQueue();
+
+        _controller = new Controller(new ControllerJmsDelegate(_context), REGISTRATION_TIMEOUT, COMMAND_RESPONSE_TIMEOUT);
+    }
+
+    @Override
+    protected void tearDown() throws Exception
+    {
+        try
+        {
+            if (_connection != null)
+            {
+                _connection.close();
+            }
+        }
+        finally
+        {
+            super.tearDown();
+        }
+    }
+
+    public void testControllerSendsOneCommandToSingleClient() throws Exception
+    {
+        Config config = ConfigFileHelper.getConfigFromResource(getClass(), "distributedControllerTest.json");
+        _controller.setConfig(config);
+
+        sendRegistration(CLIENT1);
+        _controller.awaitClientRegistrations();
+
+        final ArrayBlockingQueue<Command> commandList = new ArrayBlockingQueue<Command>(4);
+        final MessageConsumer clientConsumer = _session.createConsumer(_clientQueue);
+        final AtomicReference<Exception> listenerException = new AtomicReference<Exception>();
+        final MessageProducer producer = _session.createProducer(_controllerQueue);
+        clientConsumer.setMessageListener(new MessageListener()
+        {
+            @Override
+            public void onMessage(Message message)
+            {
+                try
+                {
+                    Command command = JmsMessageAdaptor.messageToCommand(message);
+                    LOGGER.debug("Test client received " + command);
+                    commandList.add(command);
+                    producer.send(JmsMessageAdaptor.commandToMessage(_session, new Response(CLIENT1, command.getType())));
+                }
+                catch(Exception e)
+                {
+                    listenerException.set(e);
+                }
+            }
+        });
+
+        _controller.runAllTests();
+        assertCommandType(CommandType.CREATE_CONNECTION, commandList);
+        assertCommandType(CommandType.START_TEST, commandList);
+        assertCommandType(CommandType.TEAR_DOWN_TEST, commandList);
+
+        _controller.stopAllRegisteredClients();
+        assertCommandType(CommandType.STOP_CLIENT, commandList);
+        assertNull("Unexpected exception occured", listenerException.get());
+        Command command = commandList.poll(1l, TimeUnit.SECONDS);
+        assertNull("Unexpected command is received", command);
+    }
+
+    private void assertCommandType(CommandType expectedType, BlockingQueue<Command> commandList) throws InterruptedException
+    {
+        Command command = commandList.poll(1l, TimeUnit.SECONDS);
+        assertNotNull("Command of type " + expectedType + " is not received", command);
+        assertEquals("Unexpected command type", expectedType, command.getType());
+    }
+
+    private void sendRegistration(final String clientId) throws JMSException
+    {
+        final MessageProducer registrationProducer = _session.createProducer(_controllerQueue);
+
+        final Command command = new RegisterClientCommand(clientId, _clientQueue.getQueueName());
+        final Message registrationMessage = JmsMessageAdaptor.commandToMessage(_session, command);
+        registrationProducer.send(registrationMessage);
+        LOGGER.debug("sent registrationMessage: " + registrationMessage);
+    }
+
+}
diff --git a/qpid/java/perftests/src/test/java/org/apache/qpid/systest/disttest/controlleronly/distributedControllerTest.json b/qpid/java/perftests/src/test/java/org/apache/qpid/systest/disttest/controlleronly/distributedControllerTest.json
new file mode 100644
index 0000000000..b49603ef23
--- /dev/null
+++ b/qpid/java/perftests/src/test/java/org/apache/qpid/systest/disttest/controlleronly/distributedControllerTest.json
@@ -0,0 +1,17 @@
+{
+    "_tests":[
+    {
+      "_name": "Test 1";
+      "_clients":[
+        {
+          "_name": "client1",
+          "_connections":[
+            {
+              "_name": "connection1",
+              "_factory": "connectionfactory"
+            }
+          ]
+        }
+      ]
+    }]
+}
\ No newline at end of file
diff --git a/qpid/java/perftests/src/test/java/org/apache/qpid/systest/disttest/endtoend/EndToEndTest.java b/qpid/java/perftests/src/test/java/org/apache/qpid/systest/disttest/endtoend/EndToEndTest.java
new file mode 100644
index 0000000000..c2ec1d6291
--- /dev/null
+++ b/qpid/java/perftests/src/test/java/org/apache/qpid/systest/disttest/endtoend/EndToEndTest.java
@@ -0,0 +1,94 @@
+/*
+ * 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.systest.disttest.endtoend;
+
+import static org.apache.qpid.disttest.AbstractRunner.JNDI_CONFIG_PROP;
+import static org.apache.qpid.disttest.ControllerRunner.OUTPUT_FILE_PROP;
+import static org.apache.qpid.disttest.ControllerRunner.TEST_CONFIG_PROP;
+
+import java.io.File;
+import java.io.IOException;
+
+import org.apache.qpid.disttest.ControllerRunner;
+import org.apache.qpid.disttest.message.ParticipantAttribute;
+import org.apache.qpid.disttest.results.aggregation.TestResultAggregator;
+import org.apache.qpid.test.utils.QpidBrokerTestCase;
+import org.apache.qpid.util.FileUtils;
+
+public class EndToEndTest extends QpidBrokerTestCase
+{
+    private ControllerRunner _runner;
+    private static final String TEST_CONFIG = "perftests/src/test/java/org/apache/qpid/systest/disttest/endtoend/endtoend.json";
+    private static final String JNDI_CONFIG_FILE = "perftests/src/test/java/org/apache/qpid/systest/disttest/perftests.systests.properties";
+
+    public void testRunner() throws Exception
+    {
+        File csvOutputFile = createTemporaryCsvFile();
+        assertFalse("CSV output file must exist",csvOutputFile.exists());
+
+        final String[] args = new String[] {TEST_CONFIG_PROP + "=" + TEST_CONFIG,
+                                            JNDI_CONFIG_PROP + "=" + JNDI_CONFIG_FILE,
+                                            OUTPUT_FILE_PROP  + "=" + csvOutputFile.getAbsolutePath()};
+        _runner = new ControllerRunner();
+        _runner.parseArgumentsIntoConfig(args);
+        _runner.runController();
+
+        assertTrue("CSV output file must exist", csvOutputFile.exists());
+        final String csvContents = FileUtils.readFileAsString(csvOutputFile);
+        final String[] csvLines = csvContents.split("\n");
+
+        int numberOfHeaders = 1;
+        int numberOfParticipants = 2;
+        int numberOfSummaries = 3;
+
+        int numberOfExpectedRows = numberOfHeaders + numberOfParticipants + numberOfSummaries;
+        assertEquals("Unexpected number of lines in CSV", numberOfExpectedRows, csvLines.length);
+
+        assertDataRowsHaveCorrectTestAndClientName("End To End 1", "producingClient", "participantProducer1", csvLines[1], 1);
+        assertDataRowsHaveCorrectTestAndClientName("End To End 1", "consumingClient", "participantConsumer1", csvLines[2], 1);
+
+        assertDataRowsHaveCorrectTestAndClientName("End To End 1", "", TestResultAggregator.ALL_PARTICIPANTS_NAME, csvLines[3], 1);
+        assertDataRowsHaveCorrectTestAndClientName("End To End 1", "", TestResultAggregator.ALL_CONSUMER_PARTICIPANTS_NAME, csvLines[4], 1);
+        assertDataRowsHaveCorrectTestAndClientName("End To End 1", "", TestResultAggregator.ALL_PRODUCER_PARTICIPANTS_NAME, csvLines[5], 1);
+
+    }
+
+    private void assertDataRowsHaveCorrectTestAndClientName(String testName, String clientName, String participantName, String csvLine, int expectedNumberOfMessagesProcessed)
+    {
+        final int DONT_STRIP_EMPTY_LAST_FIELD_FLAG = -1;
+        String[] cells = csvLine.split(",", DONT_STRIP_EMPTY_LAST_FIELD_FLAG);
+        // All attributes become cells in the CSV, so this will be true
+        assertEquals("Unexpected number of cells in CSV line " + csvLine, ParticipantAttribute.values().length, cells.length);
+        assertEquals("Unexpected test name in CSV line " + csvLine, testName, cells[0]);
+        assertEquals("Unexpected client name in CSV line " + csvLine, clientName, cells[2]);
+        assertEquals("Unexpected participant name in CSV line " + csvLine, participantName, cells[3]);
+        assertEquals("Unexpected number of messages processed in CSV line " + csvLine, String.valueOf(expectedNumberOfMessagesProcessed), cells[4]);
+
+    }
+
+    private File createTemporaryCsvFile() throws IOException
+    {
+        File csvFile = File.createTempFile(getName(), ".csv");
+        csvFile.delete();
+        csvFile.deleteOnExit();
+
+        return csvFile;
+    }
+
+}
diff --git a/qpid/java/perftests/src/test/java/org/apache/qpid/systest/disttest/endtoend/endtoend.json b/qpid/java/perftests/src/test/java/org/apache/qpid/systest/disttest/endtoend/endtoend.json
new file mode 100644
index 0000000000..1b7cc51265
--- /dev/null
+++ b/qpid/java/perftests/src/test/java/org/apache/qpid/systest/disttest/endtoend/endtoend.json
@@ -0,0 +1,65 @@
+{
+    "_tests":[
+    {
+      "_name": "End To End 1";
+      "_queues":[
+        {
+          "_name": "direct://amq.direct//testQueue"
+        }
+      ];
+      "_clients":[
+        {
+          "_name": "producingClient",
+          "_connections":[
+            {
+              "_name": "connection1",
+              "_factory": "connectionfactory",
+              "_sessions": [
+                {
+                  "_sessionName": "session1",
+                  "_producers": [
+                    {
+                      "_name": "participantProducer1",
+                      "_destinationName": "direct://amq.direct//testQueue",
+                      "_numberOfMessages": 1
+                    }
+                  ]
+                }
+              ];
+              "_messageProviders":[
+                {
+                  "_name": "testProvider1";
+                  "_messageProperties": {
+                     "priority": {"@def": "list"; "_items": [1,2,3,4,4]};
+                    "id": {"@def": "random"; "_upper": 10};
+                    "test": "test-value"
+                  }
+                }
+              ]
+            }
+          ]
+        },
+        {
+          "_name": "consumingClient",
+          "_connections":[
+            {
+              "_name": "connection1",
+              "_factory": "connectionfactory",
+              "_sessions": [
+                {
+                  "_sessionName": "session1",
+                  "_consumers": [
+                    {
+                      "_name": "participantConsumer1",
+                      "_destinationName": "direct://amq.direct//testQueue",
+                      "_numberOfMessages": 1
+                    }
+                  ]
+                }
+              ]
+            }
+          ]
+        }
+      ]
+    }]
+}
\ No newline at end of file
diff --git a/qpid/java/perftests/src/test/java/org/apache/qpid/systest/disttest/perftests.systests.properties b/qpid/java/perftests/src/test/java/org/apache/qpid/systest/disttest/perftests.systests.properties
new file mode 100644
index 0000000000..b5d053227c
--- /dev/null
+++ b/qpid/java/perftests/src/test/java/org/apache/qpid/systest/disttest/perftests.systests.properties
@@ -0,0 +1,26 @@
+# 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.
+
+# this file is used for running system tests of the performance test framework,
+# (i.e. not for running the performance tests themselves!)
+
+java.naming.factory.initial = org.apache.qpid.jndi.PropertiesFileInitialContextFactory
+
+# use QpidBrokerTestCase's default port
+connectionfactory.connectionfactory = amqp://guest:guest@clientid/test?brokerlist='tcp://localhost:15672'
+
+destination.controllerqueue = direct://amq.direct//controllerqueue