QPID-3346: move message group feature into trunk.

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

8 files changed, 1224 insertions, 27 deletions

diff --git a/qpid/cpp/src/tests/CMakeLists.txt b/qpid/cpp/src/tests/CMakeLists.txt
index cc33478114..7d781e5eb3 100644
--- a/qpid/cpp/src/tests/CMakeLists.txt
+++ b/qpid/cpp/src/tests/CMakeLists.txt
@@ -272,6 +272,11 @@ add_executable (datagen datagen.cpp ${platform_test_additions})
 target_link_libraries (datagen qpidclient)
 remember_location(datagen)
 
+add_executable (msg_group_test msg_group_test.cpp ${platform_test_additions})
+target_link_libraries (msg_group_test qpidmessaging)
+remember_location(msg_group_test)
+
+
 # qpid-perftest and qpid-latency-test are generally useful so install them
 install (TARGETS qpid-perftest qpid-latency-test RUNTIME
          DESTINATION ${QPID_INSTALL_BINDIR})
diff --git a/qpid/cpp/src/tests/Makefile.am b/qpid/cpp/src/tests/Makefile.am
index cd569e901c..78ac6db5f1 100644
--- a/qpid/cpp/src/tests/Makefile.am
+++ b/qpid/cpp/src/tests/Makefile.am
@@ -291,13 +291,19 @@ qpid_stream_INCLUDES=$(PUBLIC_INCLUDES)
 qpid_stream_SOURCES=qpid-stream.cpp
 qpid_stream_LDADD=$(lib_messaging)
 
+check_PROGRAMS+=msg_group_test
+msg_group_test_INCLUDES=$(PUBLIC_INCLUDES)
+msg_group_test_SOURCES=msg_group_test.cpp
+msg_group_test_LDADD=$(lib_messaging)
+
 TESTS_ENVIRONMENT = \
     VALGRIND=$(VALGRIND) \
     LIBTOOL="$(LIBTOOL)" \
     QPID_DATA_DIR= \
     $(srcdir)/run_test 
 
-system_tests = qpid-client-test quick_perftest quick_topictest run_header_test quick_txtest
+system_tests = qpid-client-test quick_perftest quick_topictest run_header_test quick_txtest \
+  run_msg_group_tests
 TESTS += start_broker $(system_tests) python_tests stop_broker run_federation_tests run_federation_sys_tests \
   run_acl_tests run_cli_tests replication_test dynamic_log_level_test \
   run_queue_flow_limit_tests ipv6_test
@@ -342,7 +348,8 @@ EXTRA_DIST +=								\
   start_broker.ps1							\
   stop_broker.ps1							\
   topictest.ps1                                                         \
-  run_queue_flow_limit_tests
+  run_queue_flow_limit_tests						\
+  run_msg_group_tests
 
 check_LTLIBRARIES += libdlclose_noop.la
 libdlclose_noop_la_LDFLAGS = -module -rpath $(abs_builddir)
@@ -353,7 +360,10 @@ CLEANFILES+=valgrind.out *.log *.vglog* dummy_test qpidd.port $(unit_wrappers)
 # Longer running stability tests, not run by default check: target.
 # Not run under valgrind, too slow
 
-LONG_TESTS+=start_broker fanout_perftest shared_perftest multiq_perftest topic_perftest run_ring_queue_test stop_broker \
+LONG_TESTS+=start_broker \
+ fanout_perftest shared_perftest multiq_perftest topic_perftest run_ring_queue_test \
+ run_msg_groups_tests_soak \
+ stop_broker \
  run_long_federation_sys_tests \
  run_failover_soak reliable_replication_test \
  federated_cluster_test_with_node_failure
@@ -366,7 +376,8 @@ EXTRA_DIST+=						\
 	run_failover_soak				\
 	reliable_replication_test			\
 	federated_cluster_test_with_node_failure        \
-	sasl_test_setup.sh
+	sasl_test_setup.sh                              \
+	run_msg_groups_tests_soak
 
 check-long:
 	$(MAKE) check TESTS="$(LONG_TESTS)" VALGRIND=
diff --git a/qpid/cpp/src/tests/QueueTest.cpp b/qpid/cpp/src/tests/QueueTest.cpp
index d94a5cab7f..7bf061ff54 100644
--- a/qpid/cpp/src/tests/QueueTest.cpp
+++ b/qpid/cpp/src/tests/QueueTest.cpp
@@ -56,12 +56,12 @@ class TestConsumer : public virtual Consumer{
 public:
     typedef boost::shared_ptr<TestConsumer> shared_ptr;
 
-    intrusive_ptr<Message> last;
+    QueuedMessage last;
     bool received;
-    TestConsumer(bool acquire = true):Consumer(acquire), received(false) {};
+    TestConsumer(std::string name="test", bool acquire = true):Consumer(name, acquire), received(false) {};
 
     virtual bool deliver(QueuedMessage& msg){
-        last = msg.payload;
+        last = msg;
         received = true;
         return true;
     };
@@ -149,16 +149,16 @@ QPID_AUTO_TEST_CASE(testConsumers){
 
     queue->deliver(msg1);
     BOOST_CHECK(queue->dispatch(c1));
-    BOOST_CHECK_EQUAL(msg1.get(), c1->last.get());
+    BOOST_CHECK_EQUAL(msg1.get(), c1->last.payload.get());
 
     queue->deliver(msg2);
     BOOST_CHECK(queue->dispatch(c2));
-    BOOST_CHECK_EQUAL(msg2.get(), c2->last.get());
+    BOOST_CHECK_EQUAL(msg2.get(), c2->last.payload.get());
 
     c1->received = false;
     queue->deliver(msg3);
     BOOST_CHECK(queue->dispatch(c1));
-    BOOST_CHECK_EQUAL(msg3.get(), c1->last.get());
+    BOOST_CHECK_EQUAL(msg3.get(), c1->last.payload.get());
 
     //Test cancellation:
     queue->cancel(c1);
@@ -214,7 +214,7 @@ QPID_AUTO_TEST_CASE(testDequeue){
     if (!consumer->received)
         sleep(2);
 
-    BOOST_CHECK_EQUAL(msg3.get(), consumer->last.get());
+    BOOST_CHECK_EQUAL(msg3.get(), consumer->last.payload.get());
     BOOST_CHECK_EQUAL(uint32_t(0), queue->getMessageCount());
 
     received = queue->get().payload;
@@ -298,14 +298,14 @@ QPID_AUTO_TEST_CASE(testSeek){
     queue->deliver(msg2);
     queue->deliver(msg3);
 
-    TestConsumer::shared_ptr consumer(new TestConsumer(false));
+    TestConsumer::shared_ptr consumer(new TestConsumer("test", false));
     SequenceNumber seq(2);
     consumer->position = seq;
 
     QueuedMessage qm;
     queue->dispatch(consumer);
 
-    BOOST_CHECK_EQUAL(msg3.get(), consumer->last.get());
+    BOOST_CHECK_EQUAL(msg3.get(), consumer->last.payload.get());
     queue->dispatch(consumer);
     queue->dispatch(consumer); // make sure over-run is safe
 
@@ -325,14 +325,18 @@ QPID_AUTO_TEST_CASE(testSearch){
     queue->deliver(msg3);
 
     SequenceNumber seq(2);
-    QueuedMessage qm = queue->find(seq);
+    QueuedMessage qm;
+    TestConsumer::shared_ptr c1(new TestConsumer());
+
+    BOOST_CHECK(queue->find(seq, qm));
 
     BOOST_CHECK_EQUAL(seq.getValue(), qm.position.getValue());
 
-    queue->acquire(qm);
+    queue->acquire(qm, c1->getName());
     BOOST_CHECK_EQUAL(queue->getMessageCount(), 2u);
     SequenceNumber seq1(3);
-    QueuedMessage qm1 = queue->find(seq1);
+    QueuedMessage qm1;
+    BOOST_CHECK(queue->find(seq1, qm1));
     BOOST_CHECK_EQUAL(seq1.getValue(), qm1.position.getValue());
 
 }
@@ -552,12 +556,13 @@ QPID_AUTO_TEST_CASE(testLVQAcquire){
     QueuedMessage qmsg2(queue.get(), msg2, ++sequence);
     framing::SequenceNumber sequence1(10);
     QueuedMessage qmsg3(queue.get(), 0, sequence1);
+    TestConsumer::shared_ptr dummy(new TestConsumer());
 
-    BOOST_CHECK(!queue->acquire(qmsg));
-    BOOST_CHECK(queue->acquire(qmsg2));
+    BOOST_CHECK(!queue->acquire(qmsg, dummy->getName()));
+    BOOST_CHECK(queue->acquire(qmsg2, dummy->getName()));
     // Acquire the massage again to test failure case.
-    BOOST_CHECK(!queue->acquire(qmsg2));
-    BOOST_CHECK(!queue->acquire(qmsg3));
+    BOOST_CHECK(!queue->acquire(qmsg2, dummy->getName()));
+    BOOST_CHECK(!queue->acquire(qmsg3, dummy->getName()));
 
     BOOST_CHECK_EQUAL(queue->getMessageCount(), 2u);
 
@@ -567,7 +572,7 @@ QPID_AUTO_TEST_CASE(testLVQAcquire){
     // set mode to no browse and check
     args.setOrdering(client::LVQ_NO_BROWSE);
     queue->configure(args);
-    TestConsumer::shared_ptr c1(new TestConsumer(false));
+    TestConsumer::shared_ptr c1(new TestConsumer("test", false));
 
     queue->dispatch(c1);
     queue->dispatch(c1);
@@ -696,6 +701,280 @@ QPID_AUTO_TEST_CASE(testQueueCleaner) {
     BOOST_CHECK_EQUAL(queue->getMessageCount(), 0u);
 }
 
+
+namespace {
+    // helper for group tests
+    void verifyAcquire( Queue::shared_ptr queue,
+                        TestConsumer::shared_ptr c,
+                        std::deque<QueuedMessage>& results,
+                        const std::string& expectedGroup,
+                        const int expectedId )
+    {
+        queue->dispatch(c);
+        results.push_back(c->last);
+        std::string group = c->last.payload->getProperties<MessageProperties>()->getApplicationHeaders().getAsString("GROUP-ID");
+        int id = c->last.payload->getProperties<MessageProperties>()->getApplicationHeaders().getAsInt("MY-ID");
+        BOOST_CHECK_EQUAL( group, expectedGroup );
+        BOOST_CHECK_EQUAL( id, expectedId );
+    }
+}
+
+QPID_AUTO_TEST_CASE(testGroupsMultiConsumer) {
+    //
+    // Verify that consumers of grouped messages own the groups once a message is acquired,
+    // and release the groups once all acquired messages have been dequeued or requeued
+    //
+    FieldTable args;
+    Queue::shared_ptr queue(new Queue("my_queue", true));
+    args.setString("qpid.group_header_key", "GROUP-ID");
+    args.setInt("qpid.shared_msg_group", 1);
+    queue->configure(args);
+
+    std::string groups[] = { std::string("a"), std::string("a"), std::string("a"),
+                             std::string("b"), std::string("b"), std::string("b"),
+                             std::string("c"), std::string("c"), std::string("c") };
+    for (int i = 0; i < 9; ++i) {
+        intrusive_ptr<Message> msg = create_message("e", "A");
+        msg->insertCustomProperty("GROUP-ID", groups[i]);
+        msg->insertCustomProperty("MY-ID", i);
+        queue->deliver(msg);
+    }
+
+    // Queue = a-0, a-1, a-2, b-3, b-4, b-5, c-6, c-7, c-8...
+    // Owners= ---, ---, ---, ---, ---, ---, ---, ---, ---,
+
+    BOOST_CHECK_EQUAL(uint32_t(9), queue->getMessageCount());
+
+    TestConsumer::shared_ptr c1(new TestConsumer("C1"));
+    TestConsumer::shared_ptr c2(new TestConsumer("C2"));
+
+    queue->consume(c1);
+    queue->consume(c2);
+
+    std::deque<QueuedMessage> dequeMeC1;
+    std::deque<QueuedMessage> dequeMeC2;
+
+
+    verifyAcquire(queue, c1, dequeMeC1, "a", 0 );  // c1 now owns group "a" (acquire a-0)
+    verifyAcquire(queue, c2, dequeMeC2, "b", 3 );  // c2 should now own group "b" (acquire b-3)
+
+    // now let c1 complete the 'a-0' message - this should free the 'a' group
+    queue->dequeue( 0, dequeMeC1.front() );
+    dequeMeC1.pop_front();
+
+    // Queue = a-1, a-2, b-3, b-4, b-5, c-6, c-7, c-8...
+    // Owners= ---, ---, ^C2, ^C2, ^C2, ---, ---, ---
+
+    // now c2 should pick up the next 'a-1', since it is oldest free
+    verifyAcquire(queue, c2, dequeMeC2, "a", 1 ); // c2 should now own groups "a" and "b"
+
+    // Queue = a-1, a-2, b-3, b-4, b-5, c-6, c-7, c-8...
+    // Owners= ^C2, ^C2, ^C2, ^C2, ^C2, ---, ---, ---
+
+    // c1 should only be able to snarf up the first "c" message now...
+    verifyAcquire(queue, c1, dequeMeC1, "c", 6 );    // should skip to the first "c"
+
+    // Queue = a-1, a-2, b-3, b-4, b-5, c-6, c-7, c-8...
+    // Owners= ^C2, ^C2, ^C2, ^C2, ^C2, ^C1, ^C1, ^C1
+
+    // hmmm... what if c2 now dequeues "b-3"?  (now only has a-1 acquired)
+    queue->dequeue( 0, dequeMeC2.front() );
+    dequeMeC2.pop_front();
+
+    // Queue = a-1, a-2, b-4, b-5, c-6, c-7, c-8...
+    // Owners= ^C2, ^C2, ---, ---, ^C1, ^C1, ^C1
+
+    // b group is free, c is owned by c1 - c1's next get should grab 'b-4'
+    verifyAcquire(queue, c1, dequeMeC1, "b", 4 );
+
+    // Queue = a-1, a-2, b-4, b-5, c-6, c-7, c-8...
+    // Owners= ^C2, ^C2, ^C1, ^C1, ^C1, ^C1, ^C1
+
+    // c2 can now only grab a-2, and that's all
+    verifyAcquire(queue, c2, dequeMeC2, "a", 2 );
+
+    // now C2 can't get any more, since C1 owns "b" and "c" group...
+    bool gotOne = queue->dispatch(c2);
+    BOOST_CHECK( !gotOne );
+
+    // hmmm... what if c1 now dequeues "c-6"?  (now only own's b-4)
+    queue->dequeue( 0, dequeMeC1.front() );
+    dequeMeC1.pop_front();
+
+    // Queue = a-1, a-2, b-4, b-5, c-7, c-8...
+    // Owners= ^C2, ^C2, ^C1, ^C1, ---, ---
+
+    // c2 can now grab c-7
+    verifyAcquire(queue, c2, dequeMeC2, "c", 7 );
+
+    // Queue = a-1, a-2, b-4, b-5, c-7, c-8...
+    // Owners= ^C2, ^C2, ^C1, ^C1, ^C2, ^C2
+
+    // what happens if C-2 "requeues" a-1 and a-2?
+    queue->requeue( dequeMeC2.front() );
+    dequeMeC2.pop_front();
+    queue->requeue( dequeMeC2.front() );
+    dequeMeC2.pop_front();  // now just has c-7 acquired
+
+    // Queue = a-1, a-2, b-4, b-5, c-7, c-8...
+    // Owners= ---, ---, ^C1, ^C1, ^C2, ^C2
+
+    // now c1 will grab a-1 and a-2...
+    verifyAcquire(queue, c1, dequeMeC1, "a", 1 );
+    verifyAcquire(queue, c1, dequeMeC1, "a", 2 );
+
+    // Queue = a-1, a-2, b-4, b-5, c-7, c-8...
+    // Owners= ^C1, ^C1, ^C1, ^C1, ^C2, ^C2
+
+    // c2 can now acquire c-8 only
+    verifyAcquire(queue, c2, dequeMeC2, "c", 8 );
+
+    // and c1 can get b-5
+    verifyAcquire(queue, c1, dequeMeC1, "b", 5 );
+
+    // should be no more acquire-able for anyone now:
+    gotOne = queue->dispatch(c1);
+    BOOST_CHECK( !gotOne );
+    gotOne = queue->dispatch(c2);
+    BOOST_CHECK( !gotOne );
+
+    // requeue all of C1's acquired messages, then cancel C1
+    while (!dequeMeC1.empty()) {
+        queue->requeue(dequeMeC1.front());
+        dequeMeC1.pop_front();
+    }
+    queue->cancel(c1);
+
+    // Queue = a-1, a-2, b-4, b-5, c-7, c-8...
+    // Owners= ---, ---, ---, ---, ^C2, ^C2
+
+    // b-4, a-1, a-2, b-5 all should be available, right?
+    verifyAcquire(queue, c2, dequeMeC2, "a", 1 );
+
+    while (!dequeMeC2.empty()) {
+        queue->dequeue(0, dequeMeC2.front());
+        dequeMeC2.pop_front();
+    }
+
+    // Queue = a-2, b-4, b-5
+    // Owners= ---, ---, ---
+
+    TestConsumer::shared_ptr c3(new TestConsumer("C3"));
+    std::deque<QueuedMessage> dequeMeC3;
+
+    verifyAcquire(queue, c3, dequeMeC3, "a", 2 );
+    verifyAcquire(queue, c2, dequeMeC2, "b", 4 );
+
+    // Queue = a-2, b-4, b-5
+    // Owners= ^C3, ^C2, ^C2
+
+    gotOne = queue->dispatch(c3);
+    BOOST_CHECK( !gotOne );
+
+    verifyAcquire(queue, c2, dequeMeC2, "b", 5 );
+
+    while (!dequeMeC2.empty()) {
+        queue->dequeue(0, dequeMeC2.front());
+        dequeMeC2.pop_front();
+    }
+
+    // Queue = a-2,
+    // Owners= ^C3,
+
+    intrusive_ptr<Message> msg = create_message("e", "A");
+    msg->insertCustomProperty("GROUP-ID", "a");
+    msg->insertCustomProperty("MY-ID", 9);
+    queue->deliver(msg);
+
+    // Queue = a-2, a-9
+    // Owners= ^C3, ^C3
+
+    gotOne = queue->dispatch(c2);
+    BOOST_CHECK( !gotOne );
+
+    msg = create_message("e", "A");
+    msg->insertCustomProperty("GROUP-ID", "b");
+    msg->insertCustomProperty("MY-ID", 10);
+    queue->deliver(msg);
+
+    // Queue = a-2, a-9, b-10
+    // Owners= ^C3, ^C3, ----
+
+    verifyAcquire(queue, c2, dequeMeC2, "b", 10 );
+    verifyAcquire(queue, c3, dequeMeC3, "a", 9 );
+
+    gotOne = queue->dispatch(c3);
+    BOOST_CHECK( !gotOne );
+
+    queue->cancel(c2);
+    queue->cancel(c3);
+}
+
+
+QPID_AUTO_TEST_CASE(testGroupsMultiConsumerDefaults) {
+    //
+    // Verify that the same default group name is automatically applied to messages that
+    // do not specify a group name.
+    //
+    FieldTable args;
+    Queue::shared_ptr queue(new Queue("my_queue", true));
+    args.setString("qpid.group_header_key", "GROUP-ID");
+    args.setInt("qpid.shared_msg_group", 1);
+    queue->configure(args);
+
+    for (int i = 0; i < 3; ++i) {
+        intrusive_ptr<Message> msg = create_message("e", "A");
+        // no "GROUP-ID" header
+        msg->insertCustomProperty("MY-ID", i);
+        queue->deliver(msg);
+    }
+
+    // Queue = 0, 1, 2
+
+    BOOST_CHECK_EQUAL(uint32_t(3), queue->getMessageCount());
+
+    TestConsumer::shared_ptr c1(new TestConsumer("C1"));
+    TestConsumer::shared_ptr c2(new TestConsumer("C2"));
+
+    queue->consume(c1);
+    queue->consume(c2);
+
+    std::deque<QueuedMessage> dequeMeC1;
+    std::deque<QueuedMessage> dequeMeC2;
+
+    queue->dispatch(c1);    // c1 now owns default group (acquired 0)
+    dequeMeC1.push_back(c1->last);
+    int id = c1->last.payload->getProperties<MessageProperties>()->getApplicationHeaders().getAsInt("MY-ID");
+    BOOST_CHECK_EQUAL( id, 0 );
+
+    bool gotOne = queue->dispatch(c2);  // c2 should get nothing
+    BOOST_CHECK( !gotOne );
+
+    queue->dispatch(c1);    // c1 now acquires 1
+    dequeMeC1.push_back(c1->last);
+    id = c1->last.payload->getProperties<MessageProperties>()->getApplicationHeaders().getAsInt("MY-ID");
+    BOOST_CHECK_EQUAL( id, 1 );
+
+    gotOne = queue->dispatch(c2);  // c2 should still get nothing
+    BOOST_CHECK( !gotOne );
+
+    while (!dequeMeC1.empty()) {
+        queue->dequeue(0, dequeMeC1.front());
+        dequeMeC1.pop_front();
+    }
+
+    // now default group should be available...
+    queue->dispatch(c2);    // c2 now owns default group (acquired 2)
+    id = c2->last.payload->getProperties<MessageProperties>()->getApplicationHeaders().getAsInt("MY-ID");
+    BOOST_CHECK_EQUAL( id, 2 );
+
+    gotOne = queue->dispatch(c1);  // c1 should get nothing
+    BOOST_CHECK( !gotOne );
+
+    queue->cancel(c1);
+    queue->cancel(c2);
+}
+
 QPID_AUTO_TEST_CASE(testMultiQueueLastNode){
 
     TestMessageStoreOC  testStore;
diff --git a/qpid/cpp/src/tests/cluster_tests.py b/qpid/cpp/src/tests/cluster_tests.py
index 4e3339650b..d217f9fbde 100755
--- a/qpid/cpp/src/tests/cluster_tests.py
+++ b/qpid/cpp/src/tests/cluster_tests.py
@@ -1418,6 +1418,76 @@ class LongTests(BrokerTest):
             if receiver: receiver.connection.detach()
             logger.setLevel(log_level)
 
+    def test_msg_group_failover(self):
+        """Test fail-over during continuous send-receive of grouped messages.
+        """
+
+        class GroupedTrafficGenerator(Thread):
+            def __init__(self, url, queue, group_key):
+                Thread.__init__(self)
+                self.url = url
+                self.queue = queue
+                self.group_key = group_key
+                self.status = -1
+
+            def run(self):
+                # generate traffic for approx 10 seconds (2011msgs / 200 per-sec)
+                cmd = ["msg_group_test",
+                       "--broker=%s" % self.url,
+                       "--address=%s" % self.queue,
+                       "--connection-options={%s}" % (Cluster.CONNECTION_OPTIONS),
+                       "--group-key=%s" % self.group_key,
+                       "--receivers=2",
+                       "--senders=3",
+                       "--messages=2011",
+                       "--send-rate=200",
+                       "--capacity=11",
+                       "--ack-frequency=23",
+                       "--allow-duplicates",
+                       "--group-size=37",
+                       "--randomize-group-size",
+                       "--interleave=13"]
+                #      "--trace"]
+                self.generator = Popen( cmd );
+                self.status = self.generator.wait()
+                return self.status
+
+            def results(self):
+                self.join(timeout=30)  # 3x assumed duration
+                if self.isAlive(): return -1
+                return self.status
+
+        # Original cluster will all be killed so expect exit with failure
+        cluster = self.cluster(3, expect=EXPECT_EXIT_FAIL, args=["-t"])
+        for b in cluster: b.ready()     # Wait for brokers to be ready
+
+        # create a queue with rather draconian flow control settings
+        ssn0 = cluster[0].connect().session()
+        q_args = "{'qpid.group_header_key':'group-id', 'qpid.shared_msg_group':1}"
+        s0 = ssn0.sender("test-group-q; {create:always, node:{type:queue, x-declare:{arguments:%s}}}" % q_args)
+
+        # Kill original brokers, start new ones for the duration.
+        endtime = time.time() + self.duration();
+        i = 0
+        while time.time() < endtime:
+            traffic = GroupedTrafficGenerator( cluster[i].host_port(),
+                                               "test-group-q", "group-id" )
+            traffic.start()
+            time.sleep(1)
+
+            for x in range(2):
+                for b in cluster[i:]: b.ready() # Check if any broker crashed.
+                cluster[i].kill()
+                i += 1
+                b = cluster.start(expect=EXPECT_EXIT_FAIL)
+                time.sleep(1)
+
+            # wait for traffic to finish, verify success
+            self.assertEqual(0, traffic.results())
+
+        for i in range(i, len(cluster)): cluster[i].kill()
+
+
 class StoreTests(BrokerTest):
     """
     Cluster tests that can only be run if there is a store available.
diff --git a/qpid/cpp/src/tests/msg_group_test.cpp b/qpid/cpp/src/tests/msg_group_test.cpp
new file mode 100644
index 0000000000..6b9d09b89a
--- /dev/null
+++ b/qpid/cpp/src/tests/msg_group_test.cpp
@@ -0,0 +1,618 @@
+/*
+ *
+ * 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.
+ *
+ */
+
+#include <qpid/messaging/Address.h>
+#include <qpid/messaging/Connection.h>
+#include <qpid/messaging/Receiver.h>
+#include <qpid/messaging/Sender.h>
+#include <qpid/messaging/Session.h>
+#include <qpid/messaging/Message.h>
+#include <qpid/messaging/FailoverUpdates.h>
+#include <qpid/Options.h>
+#include <qpid/log/Logger.h>
+#include <qpid/log/Options.h>
+#include "qpid/log/Statement.h"
+#include "qpid/sys/Time.h"
+#include "qpid/sys/Runnable.h"
+#include "qpid/sys/Thread.h"
+#include "qpid/sys/SystemInfo.h"
+
+#include <iostream>
+#include <memory>
+#include <stdlib.h>
+
+using namespace qpid::messaging;
+using namespace qpid::types;
+using namespace std;
+
+namespace qpid {
+namespace tests {
+
+struct Options : public qpid::Options
+{
+    bool help;
+    std::string url;
+    std::string address;
+    std::string connectionOptions;
+    uint messages;
+    uint capacity;
+    uint ackFrequency;
+    bool failoverUpdates;
+    qpid::log::Options log;
+    uint senders;
+    uint receivers;
+    uint groupSize;
+    bool printReport;
+    std::string groupKey;
+    bool durable;
+    bool allowDuplicates;
+    bool randomizeSize;
+    bool stickyConsumer;
+    uint timeout;
+    uint interleave;
+    std::string prefix;
+    uint sendRate;
+
+    Options(const std::string& argv0=std::string())
+        : qpid::Options("Options"),
+          help(false),
+          url("amqp:tcp:127.0.0.1"),
+          messages(10000),
+          capacity(1000),
+          ackFrequency(100),
+          failoverUpdates(false),
+          log(argv0),
+          senders(2),
+          receivers(2),
+          groupSize(10),
+          printReport(false),
+          groupKey("qpid.no_group"),
+          durable(false),
+          allowDuplicates(false),
+          randomizeSize(false),
+          stickyConsumer(false),
+          timeout(10),
+          interleave(1),
+          sendRate(0)
+    {
+        addOptions()
+          ("ack-frequency", qpid::optValue(ackFrequency, "N"), "Ack frequency (0 implies none of the messages will get accepted)")
+          ("address,a", qpid::optValue(address, "ADDRESS"), "address to send and receive from")
+          ("allow-duplicates", qpid::optValue(allowDuplicates), "Ignore the delivery of duplicated messages")
+          ("broker,b", qpid::optValue(url, "URL"), "url of broker to connect to")
+          ("capacity", qpid::optValue(capacity, "N"), "Pre-fetch window (0 implies no pre-fetch)")
+          ("connection-options", qpid::optValue(connectionOptions, "OPTIONS"), "options for the connection")
+          ("durable", qpid::optValue(durable, "yes|no"), "Mark messages as durable.")
+          ("failover-updates", qpid::optValue(failoverUpdates), "Listen for membership updates distributed via amq.failover")
+          ("group-key", qpid::optValue(groupKey, "KEY"), "Key of the message header containing the group identifier.")
+          ("group-prefix", qpid::optValue(prefix, "STRING"), "Add 'prefix' to the start of all generated group identifiers.")
+          ("group-size", qpid::optValue(groupSize, "N"), "Number of messages per a group.")
+          ("interleave", qpid::optValue(interleave, "N"), "Simultaineously interleave messages from N different groups.")
+          ("messages,m", qpid::optValue(messages, "N"), "Number of messages to send per each sender.")
+          ("receivers,r", qpid::optValue(receivers, "N"), "Number of message consumers.")
+          ("randomize-group-size", qpid::optValue(randomizeSize), "Randomize the number of messages per group to [1...group-size].")
+          ("send-rate", qpid::optValue(sendRate,"N"), "Send at rate of N messages/second. 0 means send as fast as possible.")
+          ("senders,s", qpid::optValue(senders, "N"), "Number of message producers.")
+          ("sticky-consumers", qpid::optValue(stickyConsumer), "If set, verify that all messages in a group are consumed by the same client [TBD].")
+          ("timeout", qpid::optValue(timeout, "N"), "Fail with a stall error should all consumers remain idle for timeout seconds.")
+          ("print-report", qpid::optValue(printReport), "Dump message group statistics to stdout.")
+          ("help", qpid::optValue(help), "print this usage statement");
+        add(log);
+        //("check-redelivered", qpid::optValue(checkRedelivered), "Fails with exception if a duplicate is not marked as redelivered (only relevant when ignore-duplicates is selected)")
+        //("tx", qpid::optValue(tx, "N"), "batch size for transactions (0 implies transaction are not used)")
+        //("rollback-frequency", qpid::optValue(rollbackFrequency, "N"), "rollback frequency (0 implies no transaction will be rolledback)")
+    }
+
+    bool parse(int argc, char** argv)
+    {
+        try {
+            qpid::Options::parse(argc, argv);
+            if (address.empty()) throw qpid::Exception("Address must be specified!");
+            qpid::log::Logger::instance().configure(log);
+            if (help) {
+                std::ostringstream msg;
+                std::cout << msg << *this << std::endl << std::endl
+                          << "Verifies the behavior of grouped messages." << std::endl;
+                return false;
+            } else {
+                return true;
+            }
+        } catch (const std::exception& e) {
+            std::cerr << *this << std::endl << std::endl << e.what() << std::endl;
+            return false;
+        }
+    }
+};
+
+const string EOS("eos");
+const string SN("sn");
+
+
+// class that monitors group state across all publishers and consumers.  tracks the next
+// expected sequence for each group, and total messages consumed.
+class GroupChecker
+{
+    qpid::sys::Mutex lock;
+
+    const uint totalMsgs;
+    uint totalMsgsConsumed;
+    uint totalMsgsPublished;
+    bool allowDuplicates;
+    uint duplicateMsgs;
+
+    typedef std::map<std::string, uint> SequenceMap;
+    SequenceMap sequenceMap;
+
+    // Statistics - for each group, store the names of all clients that consumed messages
+    // from that group, and the number of messages consumed per client.
+    typedef std::map<std::string, uint> ClientCounter;
+    typedef std::map<std::string, ClientCounter> GroupStatistics;
+    GroupStatistics statistics;
+
+public:
+
+    GroupChecker( uint t, bool d ) :
+        totalMsgs(t), totalMsgsConsumed(0), totalMsgsPublished(0), allowDuplicates(d),
+        duplicateMsgs(0) {}
+
+    bool checkSequence( const std::string& groupId,
+                        uint sequence, const std::string& client )
+    {
+        qpid::sys::Mutex::ScopedLock l(lock);
+
+        QPID_LOG(debug, "Client " << client << " has received " << groupId << ":" << sequence);
+
+        GroupStatistics::iterator gs = statistics.find(groupId);
+        if (gs == statistics.end()) {
+            statistics[groupId][client] = 1;
+        } else {
+            gs->second[client]++;
+        }
+        // now verify
+        SequenceMap::iterator s = sequenceMap.find(groupId);
+        if (s == sequenceMap.end()) {
+            QPID_LOG(debug,  "Client " << client << " thinks this is the first message from group " << groupId << ":" << sequence);
+            // if duplication allowed, it is possible that the last msg(s) of an old sequence are redelivered on reconnect.
+            // in this case, set the sequence from the first msg.
+            sequenceMap[groupId] = (allowDuplicates) ? sequence : 0;
+            s = sequenceMap.find(groupId);
+        } else if (sequence < s->second) {
+            duplicateMsgs++;
+            QPID_LOG(debug, "Client " << client << " thinks this message is a duplicate! " << groupId << ":" << sequence);
+            return allowDuplicates;
+        }
+        totalMsgsConsumed++;
+        return sequence == s->second++;
+    }
+
+    void sendingSequence( const std::string& groupId,
+                          uint sequence, bool eos,
+                          const std::string& client )
+    {
+        qpid::sys::Mutex::ScopedLock l(lock);
+        ++totalMsgsPublished;
+
+        QPID_LOG(debug, "Client " << client << " sending " << groupId << ":" << sequence <<
+                 ((eos) ? " (last)" : ""));
+    }
+
+    bool eraseGroup( const std::string& groupId, const std::string& name )
+    {
+        qpid::sys::Mutex::ScopedLock l(lock);
+        QPID_LOG(debug, "Deleting group " << groupId << " (by client " << name << ")");
+        return sequenceMap.erase( groupId ) == 1;
+    }
+
+    uint getNextExpectedSequence( const std::string& groupId )
+    {
+        qpid::sys::Mutex::ScopedLock l(lock);
+        return sequenceMap[groupId];
+    }
+
+    bool allMsgsConsumed()  // true when done processing msgs
+    {
+        qpid::sys::Mutex::ScopedLock l(lock);
+        return (totalMsgsPublished >= totalMsgs) &&
+          (totalMsgsConsumed >= totalMsgsPublished) &&
+          sequenceMap.size() == 0;
+    }
+
+    uint getConsumedTotal()
+    {
+        qpid::sys::Mutex::ScopedLock l(lock);
+        return totalMsgsConsumed;
+    }
+
+    uint getPublishedTotal()
+    {
+        qpid::sys::Mutex::ScopedLock l(lock);
+        return totalMsgsPublished;
+    }
+
+    ostream& print(ostream& out)
+    {
+        qpid::sys::Mutex::ScopedLock l(lock);
+        out << "Total Published: " << totalMsgsPublished << ", Total Consumed: " << totalMsgsConsumed <<
+          ", Duplicates detected: " << duplicateMsgs << std::endl;
+        out << "Total Groups: " << statistics.size() << std::endl;
+        unsigned long consumers = 0;
+        for (GroupStatistics::iterator gs = statistics.begin(); gs != statistics.end(); ++gs) {
+            out << "  GroupId: " << gs->first;
+            consumers += gs->second.size(); // # of consumers that processed this group
+            if (gs->second.size() == 1)
+                out << " completely consumed by a single client." << std::endl;
+            else
+                out << " consumed by " << gs->second.size() << " different clients." << std::endl;
+
+            for (ClientCounter::iterator cc = gs->second.begin(); cc != gs->second.end(); ++cc) {
+                out << "    Client: " << cc->first << " consumed " << cc->second << " messages from the group." << std::endl;
+            }
+        }
+        out << "Average # of consumers per group: " << ((statistics.size() != 0) ? (double(consumers)/statistics.size()) : 0) << std::endl;
+        return out;
+    }
+};
+
+
+namespace {
+    // rand() is not thread safe.  Create a singleton obj to hold a lock while calling
+    // rand() so it can be called safely by multiple concurrent clients.
+    class Randomizer {
+        qpid::sys::Mutex lock;
+    public:
+        uint operator()(uint max) {
+            qpid::sys::Mutex::ScopedLock l(lock);
+            return (rand() % max) + 1;
+        }
+    };
+
+    static Randomizer randomizer;
+}
+
+
+// tag each generated message with a group identifer
+//
+class GroupGenerator {
+
+    const std::string groupPrefix;
+    const uint groupSize;
+    const bool randomizeSize;
+    const uint interleave;
+
+    uint groupSuffix;
+    uint total;
+
+    struct GroupState {
+        std::string id;
+        const uint size;
+        uint count;
+        GroupState( const std::string& i, const uint s )
+            : id(i), size(s), count(0) {}
+    };
+    typedef std::list<GroupState> GroupList;
+    GroupList groups;
+    GroupList::iterator current;
+
+    // add a new group identifier to the list
+    void newGroup() {
+        std::ostringstream groupId(groupPrefix, ios_base::out|ios_base::ate);
+        groupId << std::string(":") << groupSuffix++;
+        uint size = (randomizeSize) ? randomizer(groupSize) : groupSize;
+        QPID_LOG(trace, "New group: GROUPID=[" << groupId.str() << "] size=" << size << " this=" << this);
+        GroupState group( groupId.str(), size );
+        groups.push_back( group );
+    }
+
+public:
+    GroupGenerator( const std::string& prefix,
+                    const uint t,
+                    const uint size,
+                    const bool randomize,
+                    const uint i)
+        : groupPrefix(prefix), groupSize(size),
+          randomizeSize(randomize), interleave(i), groupSuffix(0), total(t)
+    {
+        QPID_LOG(trace, "New group generator: PREFIX=[" << prefix << "] total=" << total << " size=" << size << " rand=" << randomize << " interleave=" << interleave << " this=" << this);
+        for (uint i = 0; i < 1 || i < interleave; ++i) {
+            newGroup();
+        }
+        current = groups.begin();
+    }
+
+    bool genGroup(std::string& groupId, uint& seq, bool& eos)
+    {
+        if (!total) return false;
+        --total;
+        if (current == groups.end())
+            current = groups.begin();
+        groupId = current->id;
+        seq = current->count++;
+        if (current->count == current->size) {
+            QPID_LOG(trace, "Last msg for " << current->id << ", " << current->count << " this=" << this);
+            eos = true;
+            if (total >= interleave) {  // need a new group to replace this one
+                newGroup();
+                groups.erase(current++);
+            } else ++current;
+        } else {
+            ++current;
+            eos = total < interleave;   // mark eos on the last message of each group
+        }
+        QPID_LOG(trace, "SENDING GROUPID=[" << groupId << "] seq=" << seq << " eos=" << eos << " this=" << this);
+        return true;
+    }
+};
+
+
+
+class Client : public qpid::sys::Runnable
+{
+public:
+    typedef boost::shared_ptr<Client> shared_ptr;
+    enum State {ACTIVE, DONE, FAILURE};
+    Client( const std::string& n, const Options& o ) : name(n), opts(o), state(ACTIVE), stopped(false) {}
+    virtual ~Client() {}
+    State getState() { return state; }
+    void testFailed( const std::string& reason ) { state = FAILURE; error << "Client '" << name << "' failed: " << reason; }
+    void clientDone() { if (state == ACTIVE) state = DONE; }
+    qpid::sys::Thread& getThread() { return thread; }
+    const std::string getErrorMsg() { return error.str(); }
+    void stop() {stopped = true;}
+    const std::string& getName() { return name; }
+
+protected:
+    const std::string name;
+    const Options& opts;
+    qpid::sys::Thread thread;
+    ostringstream error;
+    State state;
+    bool stopped;
+};
+
+
+class Consumer : public Client
+{
+    GroupChecker& checker;
+
+public:
+    Consumer(const std::string& n, const Options& o, GroupChecker& c ) : Client(n, o), checker(c) {};
+    virtual ~Consumer() {};
+
+    void run()
+    {
+        Connection connection;
+        try {
+            connection = Connection(opts.url, opts.connectionOptions);
+            connection.open();
+            std::auto_ptr<FailoverUpdates> updates(opts.failoverUpdates ? new FailoverUpdates(connection) : 0);
+            Session session = connection.createSession();
+            Receiver receiver = session.createReceiver(opts.address);
+            receiver.setCapacity(opts.capacity);
+            Message msg;
+            uint count = 0;
+
+            while (!stopped) {
+                if (receiver.fetch(msg, Duration::SECOND)) { // msg retrieved
+                    qpid::types::Variant::Map& properties = msg.getProperties();
+                    std::string groupId = properties[opts.groupKey];
+                    uint groupSeq = properties[SN];
+                    bool eof = properties[EOS];
+
+                    QPID_LOG(trace, "RECVING GROUPID=[" << groupId << "] seq=" << groupSeq << " eos=" << eof << " name=" << name);
+
+                    qpid::sys::usleep(10);
+
+                    if (!checker.checkSequence( groupId, groupSeq, name )) {
+                        ostringstream msg;
+                        msg << "Check sequence failed.  Group=" << groupId << " rcvd seq=" << groupSeq << " expected=" << checker.getNextExpectedSequence( groupId );
+                        testFailed( msg.str() );
+                        break;
+                    } else if (eof) {
+                        if (!checker.eraseGroup( groupId, name )) {
+                            ostringstream msg;
+                            msg << "Erase group failed.  Group=" << groupId << " rcvd seq=" << groupSeq;
+                            testFailed( msg.str() );
+                            break;
+                        }
+                    }
+
+                    ++count;
+                    if (opts.ackFrequency && (count % opts.ackFrequency == 0)) {
+                        session.acknowledge();
+                    }
+                    // Clear out message properties & content for next iteration.
+                    msg = Message(); // TODO aconway 2010-12-01: should be done by fetch
+                } else if (checker.allMsgsConsumed())   // timed out, nothing else to do?
+                    break;
+            }
+            session.acknowledge();
+            session.close();
+            connection.close();
+        } catch(const std::exception& error) {
+            ostringstream msg;
+            msg << "consumer error: " << error.what();
+            testFailed( msg.str() );
+            connection.close();
+        }
+        clientDone();
+        QPID_LOG(trace, "Consuming client " << name << " completed.");
+    }
+};
+
+
+
+class Producer : public Client
+{
+    GroupChecker& checker;
+    GroupGenerator generator;
+
+public:
+    Producer(const std::string& n, const Options& o, GroupChecker& c)
+        : Client(n, o), checker(c),
+          generator( n, o.messages, o.groupSize, o.randomizeSize, o.interleave )
+    {};
+    virtual ~Producer() {};
+
+    void run()
+    {
+        Connection connection;
+        try {
+            connection = Connection(opts.url, opts.connectionOptions);
+            connection.open();
+            std::auto_ptr<FailoverUpdates> updates(opts.failoverUpdates ? new FailoverUpdates(connection) : 0);
+            Session session = connection.createSession();
+            Sender sender = session.createSender(opts.address);
+            if (opts.capacity) sender.setCapacity(opts.capacity);
+            Message msg;
+            msg.setDurable(opts.durable);
+            std::string groupId;
+            uint seq;
+            bool eos;
+            uint sent = 0;
+
+            qpid::sys::AbsTime start = qpid::sys::now();
+            int64_t interval = 0;
+            if (opts.sendRate) interval = qpid::sys::TIME_SEC/opts.sendRate;
+
+            while (!stopped && generator.genGroup(groupId, seq, eos)) {
+                msg.getProperties()[opts.groupKey] = groupId;
+                msg.getProperties()[SN] = seq;
+                msg.getProperties()[EOS] = eos;
+                checker.sendingSequence( groupId, seq, eos, name );
+
+                sender.send(msg);
+                ++sent;
+
+                if (opts.sendRate) {
+                    qpid::sys::AbsTime waitTill(start, sent*interval);
+                    int64_t delay = qpid::sys::Duration(qpid::sys::now(), waitTill);
+                    if (delay > 0) qpid::sys::usleep(delay/qpid::sys::TIME_USEC);
+                }
+            }
+            session.sync();
+            session.close();
+            connection.close();
+        } catch(const std::exception& error) {
+            ostringstream msg;
+            msg << "producer '" << name << "' error: " << error.what();
+            testFailed(msg.str());
+            connection.close();
+        }
+        clientDone();
+        QPID_LOG(trace, "Producing client " << name << " completed.");
+    }
+};
+
+
+}} // namespace qpid::tests
+
+using namespace qpid::tests;
+
+int main(int argc, char ** argv)
+{
+    int status = 0;
+    try {
+        Options opts;
+        if (opts.parse(argc, argv)) {
+
+            GroupChecker state( opts.senders * opts.messages,
+                                opts.allowDuplicates);
+            std::vector<Client::shared_ptr> clients;
+
+            if (opts.randomizeSize) srand((unsigned int)qpid::sys::SystemInfo::getProcessId());
+
+            // fire off the producers && consumers
+            for (size_t j = 0; j < opts.senders; ++j)  {
+                ostringstream name;
+                name << opts.prefix << "P_" << j;
+                clients.push_back(Client::shared_ptr(new Producer( name.str(), opts, state )));
+                clients.back()->getThread() = qpid::sys::Thread(*clients.back());
+            }
+            for (size_t j = 0; j < opts.receivers; ++j)  {
+                ostringstream name;
+                name << opts.prefix << "C_" << j;
+                clients.push_back(Client::shared_ptr(new Consumer( name.str(), opts, state )));
+                clients.back()->getThread() = qpid::sys::Thread(*clients.back());
+            }
+
+            // wait for all pubs/subs to finish.... or for consumers to fail or stall.
+            uint stalledTime = 0;
+            bool done;
+            bool clientFailed = false;
+            do {
+                uint lastCount = state.getConsumedTotal();
+                qpid::sys::usleep( 1000000 );
+
+                // check each client for status
+                done = true;
+                for (std::vector<Client::shared_ptr>::iterator i = clients.begin();
+                     i != clients.end(); ++i) {
+                    QPID_LOG(debug, "Client " << (*i)->getName() << " state=" << (*i)->getState());
+                    if ((*i)->getState() == Client::FAILURE) {
+                        QPID_LOG(error, argv[0] << ": test failed with client error: " << (*i)->getErrorMsg());
+                        clientFailed = true;
+                        done = true;
+                        break;  // exit test.
+                    } else if ((*i)->getState() != Client::DONE) {
+                        done = false;
+                    }
+                }
+
+                if (!done) {
+                    // check that consumers are still receiving messages
+                    if (lastCount == state.getConsumedTotal())
+                        stalledTime++;
+                    else {
+                        lastCount = state.getConsumedTotal();
+                        stalledTime = 0;
+                    }
+                }
+
+                QPID_LOG(debug, "Consumed to date = " << state.getConsumedTotal() <<
+                         " Published to date = " << state.getPublishedTotal() <<
+                         " total=" << opts.senders * opts.messages );
+
+            } while (!done && stalledTime < opts.timeout);
+
+            if (clientFailed) {
+                status = 1;
+            } else if (stalledTime >= opts.timeout) {
+                QPID_LOG(error, argv[0] << ": test failed due to stalled consumer." );
+                status = 2;
+            }
+
+            // Wait for started threads.
+            for (std::vector<Client::shared_ptr>::iterator i = clients.begin();
+                 i != clients.end(); ++i) {
+                (*i)->stop();
+                (*i)->getThread().join();
+            }
+
+            if (opts.printReport && !status) state.print(std::cout);
+        } else status = 4;
+    } catch(const std::exception& error) {
+        QPID_LOG(error, argv[0] << ": " << error.what());
+        status = 3;
+    }
+    QPID_LOG(trace, "TEST DONE [" << status << "]");
+
+    return status;
+}
diff --git a/qpid/cpp/src/tests/qpid-send.cpp b/qpid/cpp/src/tests/qpid-send.cpp
index ef5e98e2a0..b1213a484f 100644
--- a/qpid/cpp/src/tests/qpid-send.cpp
+++ b/qpid/cpp/src/tests/qpid-send.cpp
@@ -28,6 +28,7 @@
 #include <qpid/messaging/FailoverUpdates.h>
 #include <qpid/sys/Time.h>
 #include <qpid/sys/Monitor.h>
+#include <qpid/sys/SystemInfo.h>
 #include "TestOptions.h"
 #include "Statistics.h"
 
@@ -76,6 +77,11 @@ struct Options : public qpid::Options
     uint flowControl;
     bool sequence;
     bool timestamp;
+    std::string groupKey;
+    std::string groupPrefix;
+    uint groupSize;
+    bool groupRandSize;
+    uint groupInterleave;
 
     Options(const std::string& argv0=std::string())
         : qpid::Options("Options"),
@@ -100,7 +106,11 @@ struct Options : public qpid::Options
           sendRate(0),
           flowControl(0),
           sequence(true),
-          timestamp(true)
+          timestamp(true),
+          groupPrefix("GROUP-"),
+          groupSize(10),
+          groupRandSize(false),
+          groupInterleave(1)
     {
         addOptions()
             ("broker,b", qpid::optValue(url, "URL"), "url of broker to connect to")
@@ -111,8 +121,8 @@ struct Options : public qpid::Options
             ("reply-to", qpid::optValue(replyto, "REPLY-TO"), "specify reply-to address")
             ("send-eos", qpid::optValue(sendEos, "N"), "Send N EOS messages to mark end of input")
             ("durable", qpid::optValue(durable, "yes|no"), "Mark messages as durable.")
-	    ("ttl", qpid::optValue(ttl, "msecs"), "Time-to-live for messages, in milliseconds")
-	    ("priority", qpid::optValue(priority, "PRIORITY"), "Priority for messages (higher value implies higher priority)")
+            ("ttl", qpid::optValue(ttl, "msecs"), "Time-to-live for messages, in milliseconds")
+            ("priority", qpid::optValue(priority, "PRIORITY"), "Priority for messages (higher value implies higher priority)")
             ("property,P", qpid::optValue(properties, "NAME=VALUE"), "specify message property")
             ("correlation-id", qpid::optValue(correlationid, "ID"), "correlation-id for message")
             ("user-id", qpid::optValue(userid, "USERID"), "userid for message")
@@ -131,6 +141,11 @@ struct Options : public qpid::Options
             ("flow-control", qpid::optValue(flowControl,"N"), "Do end to end flow control to limit queue depth to 2*N. 0 means no flow control.")
             ("sequence", qpid::optValue(sequence, "yes|no"), "Add a sequence number messages property (required for duplicate/lost message detection)")
             ("timestamp", qpid::optValue(timestamp, "yes|no"), "Add a time stamp messages property (required for latency measurement)")
+            ("group-key", qpid::optValue(groupKey, "KEY"), "Generate groups of messages using message header 'KEY' to hold the group identifier")
+            ("group-prefix", qpid::optValue(groupPrefix, "STRING"), "Generate group identifers with 'STRING' prefix (if group-key specified)")
+            ("group-size", qpid::optValue(groupSize, "N"), "Number of messages per a group (if group-key specified)")
+            ("group-randomize-size", qpid::optValue(groupRandSize), "Randomize the number of messages per group to [1...group-size] (if group-key specified)")
+            ("group-interleave", qpid::optValue(groupInterleave, "N"), "Simultaineously interleave messages from N different groups (if group-key specified)")
             ("help", qpid::optValue(help), "print this usage statement");
         add(log);
     }
@@ -252,6 +267,68 @@ class MapContentGenerator   : public ContentGenerator {
     const Options& opts;
 };
 
+// tag each generated message with a group identifer
+//
+class GroupGenerator {
+public:
+    GroupGenerator(const std::string& key,
+                   const std::string& prefix,
+                   const uint size,
+                   const bool randomize,
+                   const uint interleave)
+        : groupKey(key), groupPrefix(prefix), groupSize(size),
+          randomizeSize(randomize), groupSuffix(0)
+    {
+        if (randomize) srand((unsigned int)qpid::sys::SystemInfo::getProcessId());
+
+        for (uint i = 0; i < 1 || i < interleave; ++i) {
+            newGroup();
+        }
+        current = groups.begin();
+    }
+
+    void setGroupInfo(Message &msg)
+    {
+        if (current == groups.end())
+            current = groups.begin();
+        msg.getProperties()[groupKey] = current->id;
+        // std::cout << "SENDING GROUPID=[" << current->id << "]" << std::endl;
+        if (++(current->count) == current->size) {
+            newGroup();
+            groups.erase(current++);
+        } else
+            ++current;
+    }
+
+  private:
+    const std::string& groupKey;
+    const std::string& groupPrefix;
+    const uint groupSize;
+    const bool randomizeSize;
+
+    uint groupSuffix;
+
+    struct GroupState {
+        std::string id;
+        const uint size;
+        uint count;
+        GroupState( const std::string& i, const uint s )
+            : id(i), size(s), count(0) {}
+    };
+    typedef std::list<GroupState> GroupList;
+    GroupList groups;
+    GroupList::iterator current;
+
+    void newGroup() {
+        std::ostringstream groupId(groupPrefix, ios_base::out|ios_base::ate);
+        groupId << groupSuffix++;
+        uint size = (randomizeSize) ? (rand() % groupSize) + 1 : groupSize;
+        // std::cout << "New group: GROUPID=[" << groupId.str() << "] size=" << size << std::endl;
+        GroupState group( groupId.str(), size );
+        groups.push_back( group );
+    }
+};
+
 int main(int argc, char ** argv)
 {
     Connection connection;
@@ -296,6 +373,14 @@ int main(int argc, char ** argv)
             else
                 contentGen.reset(new FixedContentGenerator(opts.contentString));
 
+            std::auto_ptr<GroupGenerator> groupGen;
+            if (!opts.groupKey.empty())
+                groupGen.reset(new GroupGenerator(opts.groupKey,
+                                                  opts.groupPrefix,
+                                                  opts.groupSize,
+                                                  opts.groupRandSize,
+                                                  opts.groupInterleave));
+
             qpid::sys::AbsTime start = qpid::sys::now();
             int64_t interval = 0;
             if (opts.sendRate) interval = qpid::sys::TIME_SEC/opts.sendRate;
@@ -312,9 +397,6 @@ int main(int argc, char ** argv)
                 ++sent;
                 if (opts.sequence)
                     msg.getProperties()[SN] = sent;
-                if (opts.timestamp)
-                    msg.getProperties()[TS] = int64_t(
-                        qpid::sys::Duration(qpid::sys::EPOCH, qpid::sys::now()));
                 if (opts.flowControl) {
                     if ((sent % opts.flowControl) == 0) {
                         msg.setReplyTo(flowControlAddress);
@@ -323,6 +405,12 @@ int main(int argc, char ** argv)
                     else
                         msg.setReplyTo(Address()); // Clear the reply address.
                 }
+                if (groupGen.get())
+                    groupGen->setGroupInfo(msg);
+
+                if (opts.timestamp)
+                    msg.getProperties()[TS] = int64_t(
+                        qpid::sys::Duration(qpid::sys::EPOCH, qpid::sys::now()));
                 sender.send(msg);
                 reporter.message(msg);
 
diff --git a/qpid/cpp/src/tests/run_msg_group_tests b/qpid/cpp/src/tests/run_msg_group_tests
new file mode 100755
index 0000000000..8423022521
--- /dev/null
+++ b/qpid/cpp/src/tests/run_msg_group_tests
@@ -0,0 +1,66 @@
+#!/bin/bash
+
+#
+# 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.
+#
+#script to run a sequence of message group queue tests via make
+
+#setup path to find qpid-config and msg_group_test progs
+source ./test_env.sh
+
+export PATH=$PWD:$srcdir:$PYTHON_COMMANDS:$PATH
+
+#set port to connect to via env var
+test -s qpidd.port && QPID_PORT=`cat qpidd.port`
+
+#trap cleanup INT TERM QUIT
+
+QUEUE_NAME="group-queue"
+GROUP_KEY="My-Group-Id"
+
+BROKER_URL="${QPID_BROKER:-localhost}:${QPID_PORT:-5672}"
+
+run_test() {
+    $@
+}
+
+##set -x
+
+declare -i i=0
+declare -a tests
+tests=("qpid-config -a $BROKER_URL add queue $QUEUE_NAME --group-header=${GROUP_KEY} --shared-groups"
+    "msg_group_test -b $BROKER_URL -a $QUEUE_NAME --group-key $GROUP_KEY --messages 103 --group-size 13 --receivers 2 --senders 3 --capacity 3 --ack-frequency 7 --randomize-group-size --interleave 3"
+    "msg_group_test -b $BROKER_URL -a $QUEUE_NAME --group-key $GROUP_KEY --messages 103 --group-size 13 --receivers 2 --senders 3 --capacity 7 --ack-frequency 7 --randomize-group-size"
+    "qpid-config -a $BROKER_URL add queue ${QUEUE_NAME}-two --group-header=${GROUP_KEY} --shared-groups"
+    "msg_group_test -b $BROKER_URL -a $QUEUE_NAME --group-key $GROUP_KEY --messages 103 --group-size 13 --receivers 2 --senders 3 --capacity 7 --ack-frequency 3 --randomize-group-size"
+    "msg_group_test -b $BROKER_URL -a ${QUEUE_NAME}-two --group-key $GROUP_KEY --messages 103 --group-size 13 --receivers 2 --senders 3 --capacity 3 --ack-frequency 7 --randomize-group-size --interleave 5"
+    "msg_group_test -b $BROKER_URL -a $QUEUE_NAME --group-key $GROUP_KEY --messages 59  --group-size 5  --receivers 2 --senders 3 --capacity 1 --ack-frequency 3 --randomize-group-size"
+    "qpid-config -a $BROKER_URL del queue ${QUEUE_NAME}-two --force"
+    "msg_group_test -b $BROKER_URL -a $QUEUE_NAME --group-key $GROUP_KEY --messages 59  --group-size 3  --receivers 2 --senders 3 --capacity 1 --ack-frequency 1 --randomize-group-size"
+    "msg_group_test -b $BROKER_URL -a $QUEUE_NAME --group-key $GROUP_KEY --messages 211 --group-size 13 --receivers 2 --senders 3 --capacity 47 --ack-frequency 79 --interleave 53"
+    "qpid-config -a $BROKER_URL del queue $QUEUE_NAME --force")
+
+while [ -n "${tests[i]}" ]; do
+    run_test ${tests[i]}
+    RETCODE=$?
+    if test x$RETCODE != x0; then
+        echo "FAILED message group test.  Failed command: \"${tests[i]}\"";
+        exit 1;
+    fi
+    i+=1
+done
diff --git a/qpid/cpp/src/tests/run_msg_group_tests_soak b/qpid/cpp/src/tests/run_msg_group_tests_soak
new file mode 100755
index 0000000000..5231f74755
--- /dev/null
+++ b/qpid/cpp/src/tests/run_msg_group_tests_soak
@@ -0,0 +1,60 @@
+#!/bin/bash
+
+#
+# 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.
+#
+#script to run a sequence of long-running message group tests via make
+
+#setup path to find qpid-config and msg_group_test test progs
+source ./test_env.sh
+
+export PATH=$PWD:$srcdir:$PYTHON_COMMANDS:$PATH
+
+#set port to connect to via env var
+test -s qpidd.port && QPID_PORT=`cat qpidd.port`
+
+#trap cleanup INT TERM QUIT
+
+QUEUE_NAME="group-queue"
+GROUP_KEY="My-Group-Id"
+
+BROKER_URL="${QPID_BROKER:-localhost}:${QPID_PORT:-5672}"
+
+run_test() {
+    $@
+}
+
+##set -x
+
+declare -i i=0
+declare -a tests
+tests=("qpid-config -a $BROKER_URL add queue $QUEUE_NAME --group-header=${GROUP_KEY} --shared-groups"
+    "msg_group_test -b $BROKER_URL -a $QUEUE_NAME --group-key $GROUP_KEY --messages 10007 --receivers 3 --senders 5 --group-size 211 --randomize-group-size --capacity 47 --ack-frequency 97"
+    "msg_group_test -b $BROKER_URL -a $QUEUE_NAME --group-key $GROUP_KEY --messages 10007 --receivers 3 --senders 5 --group-size 211 --randomize-group-size --capacity 79 --ack-frequency 79"
+    "msg_group_test -b $BROKER_URL -a $QUEUE_NAME --group-key $GROUP_KEY --messages 10007 --receivers 3 --senders 5 --group-size 211 --randomize-group-size --capacity 97 --ack-frequency 47"
+    "qpid-config -a $BROKER_URL del queue $QUEUE_NAME --force")
+
+while [ -n "${tests[i]}" ]; do
+    run_test ${tests[i]}
+    RETCODE=$?
+    if test x$RETCODE != x0; then
+        echo "FAILED message group test.  Failed command: \"${tests[i]}\"";
+        exit 1;
+    fi
+    i+=1
+done