Revert experimental cluster code, too close to 0.8 release.

Reverts revisions:
r1023966 "Introduce broker::Cluster interface."
r1024275 "Fix compile error: outline set/getCluster fucntions on Broker."
r1027210 "New cluster: core framework and initial implementation of enqueue logic."

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

6 files changed, 3 insertions, 506 deletions

diff --git a/qpid/cpp/src/tests/BrokerClusterCalls.cpp b/qpid/cpp/src/tests/BrokerClusterCalls.cpp
deleted file mode 100644
index f659702387..0000000000
--- a/qpid/cpp/src/tests/BrokerClusterCalls.cpp
+++ /dev/null
@@ -1,435 +0,0 @@
-/*
- *
- * Copyright (c) 2006 The Apache Software Foundation
- *
- * Licensed 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.
- *
- */
-
-///@file
-// Tests using a dummy broker::Cluster implementation to verify the expected
-// Cluster functions are called for various actions on the broker.
-//
-
-#include "unit_test.h"
-#include "test_tools.h"
-#include "qpid/broker/Cluster.h"
-#include "qpid/broker/Queue.h"
-#include "qpid/client/Connection.h"
-#include "qpid/client/Session.h"
-#include "qpid/messaging/Connection.h"
-#include "qpid/messaging/Session.h"
-#include "qpid/messaging/Sender.h"
-#include "qpid/messaging/Receiver.h"
-#include "qpid/messaging/Message.h"
-#include "qpid/messaging/Duration.h"
-#include "BrokerFixture.h"
-#include <boost/assign.hpp>
-#include <boost/format.hpp>
-
-using namespace std;
-using namespace boost;
-using namespace boost::assign;
-using namespace qpid::messaging;
-using boost::format;
-using boost::intrusive_ptr;
-
-namespace qpid {
-namespace tests {
-
-class DummyCluster : public broker::Cluster
-{
-  private:
-    /** Flag used to ignore events other than enqueues while routing,
-     * e.g. acquires and accepts generated in a ring queue to replace an element..
-     * In real impl would be a thread-local variable.
-     */
-    bool isRouting;
-    
-    void recordQm(const string& op, const broker::QueuedMessage& qm) {
-        history += (format("%s(%s, %d, %s)") % op % qm.queue->getName()
-                    % qm.position % qm.payload->getFrames().getContent()).str();
-    }
-    void recordMsg(const string& op, broker::Queue& q, intrusive_ptr<broker::Message> msg) {
-        history += (format("%s(%s, %s)") % op % q.getName() % msg->getFrames().getContent()).str();
-    }
-    void recordStr(const string& op, const string& name) {
-        history += (format("%s(%s)") % op % name).str();
-    }
-  public:
-    // Messages
-
-    virtual void routing(const boost::intrusive_ptr<broker::Message>& m) {
-        isRouting = true;
-        history += (format("routing(%s)") % m->getFrames().getContent()).str();
-    }
-
-    virtual bool enqueue(broker::Queue& q, const intrusive_ptr<broker::Message>&msg) {
-        recordMsg("enqueue", q, msg);
-        return true;
-    }
-
-    virtual void routed(const boost::intrusive_ptr<broker::Message>& m) {
-        history += (format("routed(%s)") % m->getFrames().getContent()).str();
-        isRouting = false;
-    }
-    virtual void acquire(const broker::QueuedMessage& qm) {
-        if (!isRouting) recordQm("acquire", qm);
-    }
-    virtual void accept(const broker::QueuedMessage& qm) {
-        if (!isRouting) recordQm("accept", qm);
-    }
-    virtual void reject(const broker::QueuedMessage& qm) {
-        if (!isRouting) recordQm("reject", qm);
-    }
-    virtual void rejected(const broker::QueuedMessage& qm) {
-        if (!isRouting) recordQm("rejected", qm);
-    }
-    virtual void release(const broker::QueuedMessage& qm) {
-        if (!isRouting) recordQm("release", qm);
-    }
-    virtual void drop(const broker::QueuedMessage& qm) {
-        if (!isRouting) recordQm("dequeue", qm);
-    }
-
-    // Consumers
-
-    virtual void consume(const broker::Queue& q, size_t n) {
-        history += (format("consume(%s, %d)") % q.getName() % n).str();
-    }
-    virtual void cancel(const broker::Queue& q, size_t n) {
-        history += (format("cancel(%s, %d)") % q.getName() % n).str();
-    }
-
-    // Wiring
-
-    virtual void create(const broker::Queue& q) { recordStr("createq", q.getName()); }
-    virtual void destroy(const broker::Queue& q) { recordStr("destroyq", q.getName()); }
-    virtual void create(const broker::Exchange& ex) { recordStr("createex", ex.getName()); }
-    virtual void destroy(const broker::Exchange& ex) { recordStr("destroyex", ex.getName()); }
-    virtual void bind(const broker::Queue& q, const broker::Exchange& ex, const std::string& key, const framing::FieldTable& /*args*/) {
-        history += (format("bind(%s, %s, %s)") % q.getName() % ex.getName() % key).str();
-    }
-    vector<string> history;
-};
-
-QPID_AUTO_TEST_SUITE(BrokerClusterCallsTestSuite)
-
-// Broker fixture with DummyCluster set up and some new API client bits.
-struct DummyClusterFixture: public BrokerFixture {
-    Connection c;
-    Session s;
-    DummyCluster*dc;
-    DummyClusterFixture() {
-        broker->setCluster(auto_ptr<broker::Cluster>(new DummyCluster));
-        dc = &static_cast<DummyCluster&>(broker->getCluster());
-        c = Connection("localhost:"+lexical_cast<string>(getPort()));
-        c.open();
-        s = c.createSession();
-    }
-    ~DummyClusterFixture() {
-        c.close();
-    }
-};
-
-QPID_AUTO_TEST_CASE(testSimplePubSub) {
-    DummyClusterFixture f;
-    vector<string>& h = f.dc->history;
-
-    // Queue creation
-    Sender sender = f.s.createSender("q;{create:always,delete:always}");
-    size_t i = 0;
-    BOOST_CHECK_EQUAL(h.at(i++), "createq(q)"); // Note: at() does bounds checking.
-    BOOST_CHECK_EQUAL(h.size(), i);
-
-    // Consumer
-    Receiver receiver = f.s.createReceiver("q");
-    f.s.sync();
-    BOOST_CHECK_EQUAL(h.at(i++), "consume(q, 1)");
-    BOOST_CHECK_EQUAL(h.size(), i);
-
-    // Send message
-    sender.send(Message("a"));
-    f.s.sync();
-    BOOST_CHECK_EQUAL(h.at(i++), "routing(a)");
-    BOOST_CHECK_EQUAL(h.at(i++), "enqueue(q, a)");
-    BOOST_CHECK_EQUAL(h.at(i++), "routed(a)");
-    // Don't check size here as it is uncertain whether acquire has happened yet.
-
-    // Acquire message
-    Message m = receiver.fetch(Duration::SECOND);
-    BOOST_CHECK_EQUAL(h.at(i++), "acquire(q, 1, a)");
-    BOOST_CHECK_EQUAL(h.size(), i);
-
-    // Acknowledge message
-    f.s.acknowledge(true);
-    f.s.sync();
-    BOOST_CHECK_EQUAL(h.at(i++), "accept(q, 1, a)");
-    BOOST_CHECK_EQUAL(h.at(i++), "dequeue(q, 1, a)");
-    BOOST_CHECK_EQUAL(h.size(), i);
-
-    // Close a consumer
-    receiver.close();
-    BOOST_CHECK_EQUAL(h.at(i++), "cancel(q, 0)");
-    BOOST_CHECK_EQUAL(h.size(), i);
-
-    // Destroy the queue
-    f.c.close();
-    BOOST_CHECK_EQUAL(h.at(i++), "destroyq(q)");
-    BOOST_CHECK_EQUAL(h.size(), i);
-}
-
-QPID_AUTO_TEST_CASE(testReleaseReject) {
-    DummyClusterFixture f;
-    vector<string>& h = f.dc->history;
-
-    Sender sender = f.s.createSender("q;{create:always,delete:always,node:{x-declare:{alternate-exchange:amq.fanout}}}");
-    sender.send(Message("a"));
-    Receiver receiver = f.s.createReceiver("q");
-    Receiver altReceiver = f.s.createReceiver("amq.fanout;{link:{name:altq}}");
-    Message m = receiver.fetch(Duration::SECOND);
-    h.clear();
-
-    // Explicit release
-    f.s.release(m);
-    f.s.sync();
-    size_t i = 0;
-    BOOST_CHECK_EQUAL(h.at(i++), "release(q, 1, a)");
-    BOOST_CHECK_EQUAL(h.size(), i);
-
-    // Implicit release on closing connection.
-    Connection c("localhost:"+lexical_cast<string>(f.getPort()));
-    c.open();
-    Session s = c.createSession();
-    Receiver r = s.createReceiver("q");
-    m = r.fetch(Duration::SECOND);
-    h.clear();
-    i = 0;
-    c.close();
-    BOOST_CHECK_EQUAL(h.at(i++), "cancel(q, 1)");
-    BOOST_CHECK_EQUAL(h.at(i++), "release(q, 1, a)");
-    BOOST_CHECK_EQUAL(h.size(), i);
-
-    // Reject message, goes to alternate exchange.
-    m = receiver.fetch(Duration::SECOND);
-    h.clear();
-    i = 0;
-    f.s.reject(m);
-    BOOST_CHECK_EQUAL(h.at(i++), "reject(q, 1, a)"); 
-    BOOST_CHECK_EQUAL(h.at(i++), "routing(a)"); // Routing to alt exchange
-    BOOST_CHECK_EQUAL(h.at(i++), "enqueue(amq.fanout_altq, a)");
-    BOOST_CHECK_EQUAL(h.at(i++), "routed(a)");
-    BOOST_CHECK_EQUAL(h.at(i++), "dequeue(q, 1, a)");
-    BOOST_CHECK_EQUAL(h.at(i++), "rejected(q, 1, a)"); 
-    BOOST_CHECK_EQUAL(h.size(), i);
-    m = altReceiver.fetch(Duration::SECOND);
-    BOOST_CHECK_EQUAL(m.getContent(), "a");
-
-    // Timed out message
-    h.clear();
-    i = 0;
-    m = Message("t");
-    m.setTtl(Duration(1));                // Timeout 1ms
-    sender.send(m);
-    usleep(2000);               // Sleep 2ms
-    bool received = receiver.fetch(m, Duration::IMMEDIATE);
-    BOOST_CHECK(!received);                     // Timed out
-    BOOST_CHECK_EQUAL(h.at(i++), "routing(t)");
-    BOOST_CHECK_EQUAL(h.at(i++), "enqueue(q, t)");
-    BOOST_CHECK_EQUAL(h.at(i++), "routed(t)");
-    BOOST_CHECK_EQUAL(h.at(i++), "dequeue(q, 2, t)");
-    BOOST_CHECK_EQUAL(h.size(), i);
-
-    // Message replaced on LVQ
-    sender = f.s.createSender("lvq;{create:always,delete:always,node:{x-declare:{arguments:{qpid.last_value_queue:1}}}}");
-    m = Message("a");
-    m.getProperties()["qpid.LVQ_key"] = "foo";
-    sender.send(m);
-    f.s.sync();
-    BOOST_CHECK_EQUAL(h.at(i++), "createq(lvq)");
-    BOOST_CHECK_EQUAL(h.at(i++), "routing(a)");
-    BOOST_CHECK_EQUAL(h.at(i++), "enqueue(lvq, a)");
-    BOOST_CHECK_EQUAL(h.at(i++), "routed(a)");
-    BOOST_CHECK_EQUAL(h.size(), i);
-
-    m = Message("b");
-    m.getProperties()["qpid.LVQ_key"] = "foo";
-    sender.send(m);
-    f.s.sync();
-    BOOST_CHECK_EQUAL(h.at(i++), "routing(b)");
-    BOOST_CHECK_EQUAL(h.at(i++), "enqueue(lvq, b)");
-    BOOST_CHECK_EQUAL(h.at(i++), "routed(b)");
-    BOOST_CHECK_EQUAL(h.size(), i);
-
-    receiver = f.s.createReceiver("lvq");
-    BOOST_CHECK_EQUAL(receiver.fetch(Duration::SECOND).getContent(), "b");
-    f.s.acknowledge(true);
-    BOOST_CHECK_EQUAL(h.at(i++), "consume(lvq, 1)");
-    BOOST_CHECK_EQUAL(h.at(i++), "acquire(lvq, 1, b)");
-    BOOST_CHECK_EQUAL(h.at(i++), "accept(lvq, 1, b)");
-    BOOST_CHECK_EQUAL(h.at(i++), "dequeue(lvq, 1, b)");
-    BOOST_CHECK_EQUAL(h.size(), i);
-}
-
-QPID_AUTO_TEST_CASE(testFanout) {
-    DummyClusterFixture f;
-    vector<string>& h = f.dc->history;
-
-    Receiver r1 = f.s.createReceiver("amq.fanout;{link:{name:r1}}");
-    Receiver r2 = f.s.createReceiver("amq.fanout;{link:{name:r2}}");
-    Sender sender = f.s.createSender("amq.fanout");
-    r1.setCapacity(0);          // Don't receive immediately.
-    r2.setCapacity(0);
-    h.clear();
-    size_t i = 0;
-
-    // Send message
-    sender.send(Message("a"));
-    f.s.sync();
-    BOOST_CHECK_EQUAL(h.at(i++), "routing(a)");
-    BOOST_CHECK_EQUAL(0u, h.at(i++).find("enqueue(amq.fanout_r"));
-    BOOST_CHECK_EQUAL(0u, h.at(i++).find("enqueue(amq.fanout_r"));
-    BOOST_CHECK(h.at(i-1) != h.at(i-2));
-    BOOST_CHECK_EQUAL(h.at(i++), "routed(a)");
-    BOOST_CHECK_EQUAL(h.size(), i);
-
-    // Receive messages
-    Message m1 = r1.fetch(Duration::SECOND);
-    f.s.acknowledge(m1, true);
-    Message m2 = r2.fetch(Duration::SECOND);
-    f.s.acknowledge(m2, true);
-
-    BOOST_CHECK_EQUAL(h.at(i++), "acquire(amq.fanout_r1, 1, a)");
-    BOOST_CHECK_EQUAL(h.at(i++), "accept(amq.fanout_r1, 1, a)");
-    BOOST_CHECK_EQUAL(h.at(i++), "dequeue(amq.fanout_r1, 1, a)");
-    BOOST_CHECK_EQUAL(h.at(i++), "acquire(amq.fanout_r2, 1, a)");
-    BOOST_CHECK_EQUAL(h.at(i++), "accept(amq.fanout_r2, 1, a)");
-    BOOST_CHECK_EQUAL(h.at(i++), "dequeue(amq.fanout_r2, 1, a)");
-    BOOST_CHECK_EQUAL(h.size(), i);
-}
-
-QPID_AUTO_TEST_CASE(testRingQueue) {
-    DummyClusterFixture f;
-    vector<string>& h = f.dc->history;
-
-    // FIXME aconway 2010-10-15: QPID-2908 ring queue address string is not working,
-    // so we can't do this:
-    // Sender sender = f.s.createSender("ring;{create:always,node:{x-declare:{arguments:{qpid.max_size:3,qpid.policy_type:ring}}}}");
-    // Must use old API to declare ring queue:
-    qpid::client::Connection c;
-    f.open(c);
-    qpid::client::Session s = c.newSession();
-    qpid::framing::FieldTable args;
-    args.setInt("qpid.max_size", 3);
-    args.setString("qpid.policy_type","ring");
-    s.queueDeclare(qpid::client::arg::queue="ring", qpid::client::arg::arguments=args);
-    c.close();
-    Sender sender = f.s.createSender("ring");
-    
-    size_t i = 0;
-    // Send message
-    sender.send(Message("a"));
-    sender.send(Message("b"));
-    sender.send(Message("c"));
-    sender.send(Message("d"));
-    f.s.sync();
-
-    BOOST_CHECK_EQUAL(h.at(i++), "createq(ring)");
-
-    BOOST_CHECK_EQUAL(h.at(i++), "routing(a)");
-    BOOST_CHECK_EQUAL(h.at(i++), "enqueue(ring, a)");
-    BOOST_CHECK_EQUAL(h.at(i++), "routed(a)");
-
-    BOOST_CHECK_EQUAL(h.at(i++), "routing(b)");
-    BOOST_CHECK_EQUAL(h.at(i++), "enqueue(ring, b)");
-    BOOST_CHECK_EQUAL(h.at(i++), "routed(b)");
-
-    BOOST_CHECK_EQUAL(h.at(i++), "routing(c)");
-    BOOST_CHECK_EQUAL(h.at(i++), "enqueue(ring, c)");
-    BOOST_CHECK_EQUAL(h.at(i++), "routed(c)");
-
-    BOOST_CHECK_EQUAL(h.at(i++), "routing(d)");
-    BOOST_CHECK_EQUAL(h.at(i++), "enqueue(ring, d)");
-    BOOST_CHECK_EQUAL(h.at(i++), "routed(d)");
-
-    Receiver receiver = f.s.createReceiver("ring");
-    BOOST_CHECK_EQUAL(receiver.fetch().getContent(), "b");
-    BOOST_CHECK_EQUAL(receiver.fetch().getContent(), "c");
-    BOOST_CHECK_EQUAL(receiver.fetch().getContent(), "d");
-    f.s.acknowledge(true);
-
-    BOOST_CHECK_EQUAL(h.at(i++), "consume(ring, 1)");
-    BOOST_CHECK_EQUAL(h.at(i++), "acquire(ring, 2, b)");
-    BOOST_CHECK_EQUAL(h.at(i++), "acquire(ring, 3, c)");
-    BOOST_CHECK_EQUAL(h.at(i++), "acquire(ring, 4, d)");
-    BOOST_CHECK_EQUAL(h.at(i++), "accept(ring, 2, b)");
-    BOOST_CHECK_EQUAL(h.at(i++), "dequeue(ring, 2, b)");
-    BOOST_CHECK_EQUAL(h.at(i++), "accept(ring, 3, c)");
-    BOOST_CHECK_EQUAL(h.at(i++), "dequeue(ring, 3, c)");
-    BOOST_CHECK_EQUAL(h.at(i++), "accept(ring, 4, d)");
-    BOOST_CHECK_EQUAL(h.at(i++), "dequeue(ring, 4, d)");
-
-    BOOST_CHECK_EQUAL(h.size(), i);
-}
-
-QPID_AUTO_TEST_CASE(testTransactions) {
-    DummyClusterFixture f;
-    vector<string>& h = f.dc->history;
-    Session ts = f.c.createTransactionalSession();
-    Sender sender = ts.createSender("q;{create:always,delete:always}");
-    size_t i = 0;
-    BOOST_CHECK_EQUAL(h.at(i++), "createq(q)"); // Note: at() does bounds checking.
-    BOOST_CHECK_EQUAL(h.size(), i);
-
-    sender.send(Message("a"));
-    sender.send(Message("b"));
-    ts.sync();
-    BOOST_CHECK_EQUAL(h.at(i++), "routing(a)");
-    BOOST_CHECK_EQUAL(h.at(i++), "routed(a)");
-    BOOST_CHECK_EQUAL(h.at(i++), "routing(b)");
-    BOOST_CHECK_EQUAL(h.at(i++), "routed(b)");
-    BOOST_CHECK_EQUAL(h.size(), i); // Not replicated till commit
-    ts.commit();
-    // FIXME aconway 2010-10-18: As things stand the cluster is not
-    // compatible with transactions
-    // - enqueues occur after routing is complete
-    // - no call to Cluster::enqueue, should be in Queue::process?
-    // - no transaction context associated with messages in the Cluster interface.
-    // - no call to Cluster::accept in Queue::dequeueCommitted
-    // BOOST_CHECK_EQUAL(h.at(i++), "enqueue(q, a)");
-    // BOOST_CHECK_EQUAL(h.at(i++), "enqueue(q, b)");
-    BOOST_CHECK_EQUAL(h.size(), i);
-
-
-    Receiver receiver = ts.createReceiver("q");
-    BOOST_CHECK_EQUAL(receiver.fetch().getContent(), "a");
-    BOOST_CHECK_EQUAL(receiver.fetch().getContent(), "b");
-    ts.acknowledge();
-    ts.sync();
-    BOOST_CHECK_EQUAL(h.at(i++), "consume(q, 1)");
-    BOOST_CHECK_EQUAL(h.at(i++), "acquire(q, 1, a)");
-    BOOST_CHECK_EQUAL(h.at(i++), "acquire(q, 2, b)");
-    BOOST_CHECK_EQUAL(h.size(), i);
-    ts.commit();
-    ts.sync();
-    // BOOST_CHECK_EQUAL(h.at(i++), "accept(q, 1, a)");
-    BOOST_CHECK_EQUAL(h.at(i++), "dequeue(q, 1, a)");
-    // BOOST_CHECK_EQUAL(h.at(i++), "accept(q, 2, b)");
-    BOOST_CHECK_EQUAL(h.at(i++), "dequeue(q, 2, b)");
-    BOOST_CHECK_EQUAL(h.size(), i);
-}
-
-QPID_AUTO_TEST_SUITE_END()
-
-}} // namespace qpid::tests
-
diff --git a/qpid/cpp/src/tests/Makefile.am b/qpid/cpp/src/tests/Makefile.am
index 2a7430b8ca..241ee0fbb1 100644
--- a/qpid/cpp/src/tests/Makefile.am
+++ b/qpid/cpp/src/tests/Makefile.am
@@ -123,8 +123,7 @@ unit_test_SOURCES= unit_test.cpp unit_test.h \
 	Variant.cpp \
 	Address.cpp \
 	ClientMessage.cpp \
-	Qmf2.cpp \
-	BrokerClusterCalls.cpp
+	Qmf2.cpp
 
 if HAVE_XML
 unit_test_SOURCES+= XmlClientSessionTest.cpp
diff --git a/qpid/cpp/src/tests/cluster.mk b/qpid/cpp/src/tests/cluster.mk
index bb0f5d150b..da191e8682 100644
--- a/qpid/cpp/src/tests/cluster.mk
+++ b/qpid/cpp/src/tests/cluster.mk
@@ -77,7 +77,7 @@ cluster_test_SOURCES =				\
 	PartialFailure.cpp			\
 	ClusterFailover.cpp
 
-cluster_test_LDADD=$(lib_client) $(lib_broker) $(lib_messaging) ../cluster.la -lboost_unit_test_framework
+cluster_test_LDADD=$(lib_client) $(lib_broker) ../cluster.la -lboost_unit_test_framework
 
 qpidtest_SCRIPTS += run_cluster_tests cluster_tests.py run_long_cluster_tests long_cluster_tests.py testlib.py cluster_tests.fail
 
diff --git a/qpid/cpp/src/tests/cluster2_tests.py b/qpid/cpp/src/tests/cluster2_tests.py
deleted file mode 100755
index e3a19ae2a0..0000000000
--- a/qpid/cpp/src/tests/cluster2_tests.py
+++ /dev/null
@@ -1,66 +0,0 @@
-#!/usr/bin/env python
-
-# 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.
-#
-
-import os, signal, sys, time, imp, re, subprocess
-from qpid import datatypes, messaging
-from qpid.brokertest import *
-from qpid.harness import Skipped
-from qpid.messaging import Message
-from qpid.messaging.exceptions import Empty
-from threading import Thread, Lock
-from logging import getLogger
-from itertools import chain
-
-log = getLogger("qpid.cluster_tests")
-
-class Cluster2Tests(BrokerTest):
-    """Tests for new cluster code."""
-
-    def test_message_enqueue(self):
-        """Test basic replication of enqueued messages."""
-
-        cluster = self.cluster(2, cluster2=True, args=["--log-enable=trace+:cluster"])
-
-        sn0 = cluster[0].connect().session()
-        r0p = sn0.receiver("p; {mode:browse, create:always, node:{x-bindings:[{exchange:'amq.fanout', queue:p}]}}");
-        r0q = sn0.receiver("q; {mode:browse, create:always, node:{x-bindings:[{exchange:'amq.fanout', queue:q}]}}");
-        s0 = sn0.sender("amq.fanout");
-
-        sn1 = cluster[1].connect().session()
-        r1p = sn1.receiver("p; {mode:browse, create:always, node:{x-bindings:[{exchange:'amq.fanout', queue:p}]}}");
-        r1q = sn1.receiver("q; {mode:browse, create:always, node:{x-bindings:[{exchange:'amq.fanout', queue:q}]}}");
-
-
-        # Send messages on member 0
-        content = ["a","b","c"]
-        for m in content: s0.send(Message(m))
-
-        # Browse on both members.
-        def check(content, receiver):
-            for c in content: self.assertEqual(c, receiver.fetch(1).content)
-            self.assertRaises(Empty, receiver.fetch, 0)
-
-        check(content, r0p)
-        check(content, r0q)
-        check(content, r1p)
-        check(content, r1q)
-
-        sn1.connection.close()
-        sn0.connection.close()
diff --git a/qpid/cpp/src/tests/run_cluster_tests b/qpid/cpp/src/tests/run_cluster_tests
index 3971a39144..e136d3810a 100755
--- a/qpid/cpp/src/tests/run_cluster_tests
+++ b/qpid/cpp/src/tests/run_cluster_tests
@@ -33,5 +33,5 @@ mkdir -p $OUTDIR
 CLUSTER_TESTS_IGNORE=${CLUSTER_TESTS_IGNORE:--i cluster_tests.StoreTests.* -I $srcdir/cluster_tests.fail}
 CLUSTER_TESTS=${CLUSTER_TESTS:-$*}
 
-with_ais_group $QPID_PYTHON_TEST -DOUTDIR=$OUTDIR -m cluster_tests -m cluster2_tests $CLUSTER_TESTS_IGNORE $CLUSTER_TESTS || exit 1
+with_ais_group $QPID_PYTHON_TEST -DOUTDIR=$OUTDIR -m cluster_tests $CLUSTER_TESTS_IGNORE $CLUSTER_TESTS || exit 1
 rm -rf $OUTDIR
diff --git a/qpid/cpp/src/tests/test_env.sh.in b/qpid/cpp/src/tests/test_env.sh.in
index 96fe6b64f4..b5c3b0fa3d 100644
--- a/qpid/cpp/src/tests/test_env.sh.in
+++ b/qpid/cpp/src/tests/test_env.sh.in
@@ -63,7 +63,6 @@ export TEST_STORE_LIB=$testmoduledir/test_store.so
 exportmodule() { test -f $moduledir/$2 && eval "export $1=$moduledir/$2"; }
 exportmodule ACL_LIB acl.so
 exportmodule CLUSTER_LIB cluster.so
-exportmodule CLUSTER2_LIB cluster2.so
 exportmodule REPLICATING_LISTENER_LIB replicating_listener.so
 exportmodule REPLICATION_EXCHANGE_LIB replication_exchange.so
 exportmodule SSLCONNECTOR_LIB sslconnector.so