Added cluster option --cluster-size.

--cluster-size=N means that during start-up the cluster waits to have N members
before accepting any clients.


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

3 files changed, 64 insertions, 16 deletions

diff --git a/cpp/src/tests/InitialStatusMap.cpp b/cpp/src/tests/InitialStatusMap.cpp
index 70b077b695..c3587965e5 100644
--- a/cpp/src/tests/InitialStatusMap.cpp
+++ b/cpp/src/tests/InitialStatusMap.cpp
@@ -40,7 +40,7 @@ Status newcomerStatus(const Uuid& id=Uuid()) { return Status(ProtocolVersion(),
 
 QPID_AUTO_TEST_CASE(testFirstInCluster) {
     // Single member is first in cluster.
-    InitialStatusMap map(MemberId(0));
+    InitialStatusMap map(MemberId(0), 1);
     Uuid id(true);
     BOOST_CHECK(!map.isComplete());
     MemberSet members = list_of(MemberId(0));
@@ -56,7 +56,7 @@ QPID_AUTO_TEST_CASE(testFirstInCluster) {
 
 QPID_AUTO_TEST_CASE(testJoinExistingCluster) {
     // Single member 0 joins existing cluster 1,2
-    InitialStatusMap map(MemberId(0));
+    InitialStatusMap map(MemberId(0), 1);
     Uuid id(true);
     MemberSet members = list_of(MemberId(0))(MemberId(1))(MemberId(2));
     map.configChange(members);
@@ -79,7 +79,7 @@ QPID_AUTO_TEST_CASE(testJoinExistingCluster) {
 
 QPID_AUTO_TEST_CASE(testMultipleFirstInCluster) {
     // Multiple members 0,1,2 join at same time.
-    InitialStatusMap map(MemberId(1)); // self is 1
+    InitialStatusMap map(MemberId(1), 1); // self is 1
     Uuid id(true);
     MemberSet members = list_of(MemberId(0))(MemberId(1))(MemberId(2));
     map.configChange(members);
@@ -99,7 +99,7 @@ QPID_AUTO_TEST_CASE(testMultipleFirstInCluster) {
 
 QPID_AUTO_TEST_CASE(testMultipleJoinExisting) {
     // Multiple members 1,2,3 join existing cluster containing 0.
-    InitialStatusMap map(MemberId(2)); // self is 2
+    InitialStatusMap map(MemberId(2), 1); // self is 2
     Uuid id(true);
     MemberSet members = list_of(MemberId(0))(MemberId(1))(MemberId(2))(MemberId(3));
     map.configChange(members);
@@ -119,7 +119,7 @@ QPID_AUTO_TEST_CASE(testMultipleJoinExisting) {
 
 QPID_AUTO_TEST_CASE(testMembersLeave) {
     // Test that map completes if members leave rather than send status.
-    InitialStatusMap map(MemberId(0));
+    InitialStatusMap map(MemberId(0), 1);
     Uuid id(true);
     map.configChange(list_of(MemberId(0))(MemberId(1))(MemberId(2)));
     map.received(MemberId(0), newcomerStatus());
@@ -134,7 +134,7 @@ QPID_AUTO_TEST_CASE(testMembersLeave) {
 
 QPID_AUTO_TEST_CASE(testInteveningConfig) {
     // Multiple config changes arrives before we complete the map.
-    InitialStatusMap map(MemberId(0));
+    InitialStatusMap map(MemberId(0), 1);
     Uuid id(true);
 
     map.configChange(list_of<MemberId>(0)(1));
@@ -159,6 +159,20 @@ QPID_AUTO_TEST_CASE(testInteveningConfig) {
     BOOST_CHECK_EQUAL(map.getClusterId(), id);
 }
 
+QPID_AUTO_TEST_CASE(testInitialSize) {
+    InitialStatusMap map(MemberId(0), 3);
+    map.configChange(list_of<MemberId>(0)(1));
+    map.received(MemberId(0), newcomerStatus());
+    map.received(MemberId(1), newcomerStatus());
+    BOOST_CHECK(!map.isComplete());
+
+    map.configChange(list_of<MemberId>(0)(1)(2));
+    map.received(MemberId(0), newcomerStatus());
+    map.received(MemberId(1), newcomerStatus());
+    map.received(MemberId(2), newcomerStatus());
+    BOOST_CHECK(map.isComplete());
+}
+
 QPID_AUTO_TEST_SUITE_END()
 
 }} // namespace qpid::tests
diff --git a/cpp/src/tests/cluster_tests.py b/cpp/src/tests/cluster_tests.py
index 3ded6c103e..ed39277f77 100755
--- a/cpp/src/tests/cluster_tests.py
+++ b/cpp/src/tests/cluster_tests.py
@@ -26,13 +26,8 @@ from qpid.messaging import Message
 from threading import Thread
 
 
-class ClusterTests(BrokerTest):
-    """Cluster tests with support for testing with a store plugin."""
-
-    def duration(self):
-        d = self.config.defines.get("DURATION")
-        if d: return float(d)*60
-        else: return 3
+class ShortTests(BrokerTest):
+    """Short cluster functionality tests."""
 
     def test_message_replication(self):
         """Test basic cluster message replication."""
@@ -57,6 +52,42 @@ class ClusterTests(BrokerTest):
         self.assertEqual("y", m.content)
         s2.connection.close()
 
+    def test_cluster_size(self):
+        """Verify cluster startup waits for N brokers if --cluster-size=N"""
+        class ConnectThread(Thread):
+            def __init__(self, broker):
+                Thread.__init__(self)
+                self.broker=broker
+                self.connected = False
+                self.error = None
+
+            def run(self):
+                try:
+                    self.broker.connect()
+                    self.connected = True
+                except Exception, e: self.error = RethrownException(e)
+
+        cluster = self.cluster(1, args=["--cluster-size=3"], wait_for_start=False)
+        c = ConnectThread(cluster[0])
+        c.start()
+        time.sleep(.01)
+        assert not c.connected
+        cluster.start(wait_for_start=False)
+        time.sleep(.01)
+        assert not c.connected
+        cluster.start(wait_for_start=False)
+        c.join(1)
+        assert not c.isAlive()       # Join didn't time out
+        assert c.connected
+        if c.error: raise c.error
+
+class LongTests(BrokerTest):
+    """Tests that can run for a long time if -DDURATION=<minutes> is set"""
+    def duration(self):
+        d = self.config.defines.get("DURATION")
+        if d: return float(d)*60
+        else: return 3                  # Default is to be quick
+
     def test_failover(self):
         """Test fail-over during continuous send-receive with errors"""
 
@@ -84,7 +115,8 @@ class ClusterTests(BrokerTest):
         receiver.stop(sender.sent)
         for i in range(i, len(cluster)): cluster[i].kill()
 
-class ClusterStoreTests(BrokerTest):
+
+class StoreTests(BrokerTest):
     """
     Cluster tests that can only be run if there is a store available.
     """
diff --git a/cpp/src/tests/run_cluster_tests b/cpp/src/tests/run_cluster_tests
index 014233d8d3..b090f71852 100755
--- a/cpp/src/tests/run_cluster_tests
+++ b/cpp/src/tests/run_cluster_tests
@@ -45,8 +45,10 @@ rm -rf $OUTDIR
 mkdir -p $OUTDIR
 
 # Ignore tests requiring a store by default. 
-TESTS="-i cluster_tests.ClusterStoreTests.* -I $srcdir/cluster_tests.fail $*"
+CLUSTER_TESTS_IGNORE=${CLUSTER_TESTS_IGNORE:--i cluster_tests.StoreTests.* -I $srcdir/cluster_tests.fail}
+CLUSTER_TESTS=${CLUSTER_TESTS:-$*}
 
-with_ais_group $PYTHON_COMMANDS/qpid-python-test -DOUTDIR=$OUTDIR -m cluster_tests $TESTS || exit 1
+set -x
+with_ais_group $PYTHON_COMMANDS/qpid-python-test -DOUTDIR=$OUTDIR -m cluster_tests $CLUSTER_TESTS_IGNORE $CLUSTER_TESTS || exit 1
 rm -rf $OUTDIR
 #exit 0