New cluster member pushes store when joining an active cluster.

Previously a broker with a clean store would not be able to join an
active cluster because the shtudown-id did not match. This commit
ensures that when a broker joins an active cluster, it always pushes
its store regardless of status. Clean/dirty status is only compared
when forming an initial cluster.

This change required splitting initialization into two phases:

PRE_INIT: occurs in the Cluster ctor during early-initialize. This
phase determines whether or not to push the store.

INIT: occurs after Cluster::initialize and does the remaining
initialization chores.


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

2 files changed, 57 insertions, 29 deletions

diff --git a/cpp/src/tests/InitialStatusMap.cpp b/cpp/src/tests/InitialStatusMap.cpp
index 91c95ac517..ecbe2d4161 100644
--- a/cpp/src/tests/InitialStatusMap.cpp
+++ b/cpp/src/tests/InitialStatusMap.cpp
@@ -173,20 +173,6 @@ 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_CASE(testAllCleanNoUpdate) {
     InitialStatusMap map(MemberId(0), 3);
     map.configChange(list_of<MemberId>(0)(1)(2));
@@ -244,8 +230,6 @@ QPID_AUTO_TEST_CASE(testEmptyAlone) {
     BOOST_CHECK(!map.isUpdateNeeded());
 }
 
-// FIXME aconway 2009-11-20: consistency tests for mixed stores,
-
 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 08f1697c7a..4fefe26db3 100755
--- a/cpp/src/tests/cluster_tests.py
+++ b/cpp/src/tests/cluster_tests.py
@@ -29,9 +29,19 @@ from itertools import chain
 
 log = getLogger("qpid.cluster_tests")
 
+# Note: brokers that shut themselves down due to critical error during
+# normal operation will still have an exit code of 0. Brokers that
+# shut down because of an error found during initialize will exit with
+# a non-0 code. Hence the apparently inconsistent use of EXPECT_EXIT_OK
+# and EXPECT_EXIT_FAIL in some of the tests below.
+
+# FIXME aconway 2010-03-11: resolve this - ideally any exit due to an error
+# should give non-0 exit status.
+
 # Import scripts as modules
 qpid_cluster=import_script(checkenv("QPID_CLUSTER_EXEC"))
 
+
 def readfile(filename):
     """Returns te content of file named filename as a string"""
     f = file(filename)
@@ -287,6 +297,11 @@ class StoreTests(BrokerTest):
         m = cluster.start("restartme").get_message("q")
         self.assertEqual("x", m.content)
 
+    def stop_cluster(self,broker):
+        """Clean shut-down of a cluster"""
+        self.assertEqual(0, qpid_cluster.main(
+            ["qpid-cluster", "-kf", broker.host_port()]))
+
     def test_persistent_restart(self):
         """Verify persistent cluster shutdown/restart scenarios"""
         cluster = self.cluster(0, args=self.args() + ["--cluster-size=3"])
@@ -302,7 +317,7 @@ class StoreTests(BrokerTest):
         self.assertEqual(c.get_message("q").content, "2")
         # Shut down the entire cluster cleanly and bring it back up
         a.send_message("q", Message("3", durable=True))
-        self.assertEqual(0, qpid_cluster.main(["qpid-cluster", "-kf", a.host_port()]))
+        self.stop_cluster(a)
         a = cluster.start("a", wait=False)
         b = cluster.start("b", wait=False)
         c = cluster.start("c", wait=True)
@@ -320,7 +335,7 @@ class StoreTests(BrokerTest):
         b.kill()
         self.assertEqual(c.get_message("q").content, "4")
         c.send_message("q", Message("clean", durable=True))
-        self.assertEqual(0, qpid_cluster.main(["qpid-cluster", "-kf", c.host_port()]))
+        self.stop_cluster(c)
         a = cluster.start("a", wait=False)
         b = cluster.start("b", wait=False)
         c = cluster.start("c", wait=True)
@@ -333,7 +348,7 @@ class StoreTests(BrokerTest):
         a.terminate()
         cluster2 = self.cluster(1, args=self.args())
         try:
-            a = cluster2.start("a", expect=EXPECT_EXIT_OK)
+            a = cluster2.start("a", expect=EXPECT_EXIT_FAIL)
             a.ready()
             self.fail("Expected exception")
         except: pass
@@ -343,27 +358,29 @@ class StoreTests(BrokerTest):
         cluster = self.cluster(0, args=self.args()+["--cluster-size=2"])
         a = cluster.start("a", expect=EXPECT_EXIT_OK, wait=False)
         b = cluster.start("b", expect=EXPECT_EXIT_OK, wait=False)
-        self.assertEqual(0, qpid_cluster.main(["qpid_cluster", "-kf", a.host_port()]))
+        self.stop_cluster(a)
         self.assertEqual(a.wait(), 0)
         self.assertEqual(b.wait(), 0)
 
         # Restart with a different member and shut down.
         a = cluster.start("a", expect=EXPECT_EXIT_OK, wait=False)
         c = cluster.start("c", expect=EXPECT_EXIT_OK, wait=False)
-        self.assertEqual(0, qpid_cluster.main(["qpid_cluster", "-kf", a.host_port()]))
+        self.stop_cluster(a)
         self.assertEqual(a.wait(), 0)
         self.assertEqual(c.wait(), 0)
-
         # Mix members from both shutdown events, they should fail
-        a = cluster.start("a", expect=EXPECT_EXIT_OK, wait=False)
-        b = cluster.start("b", expect=EXPECT_EXIT_OK, wait=False)
+        # FIXME aconway 2010-03-11: can't predict the exit status of these
+        # as it depends on the order of delivery of initial-status messages.
+        # See comment at top of this file.
+        a = cluster.start("a", expect=EXPECT_UNKNOWN, wait=False)
+        b = cluster.start("b", expect=EXPECT_UNKNOWN, wait=False)
         self.assertRaises(Exception, lambda: a.ready())
         self.assertRaises(Exception, lambda: b.ready())
 
     def assert_dirty_store(self, broker):
-        self.assertRaises(Exception, lambda: broker.ready())
+        assert retry(lambda: os.path.exists(broker.log)), "Missing log file %s"%broker.log
         msg = re.compile("critical.*no clean store")
-        assert msg.search(readfile(broker.log))
+        assert retry(lambda: msg.search(readfile(broker.log))), "Expected dirty store message in %s"%broker.log
 
     def test_solo_store_clean(self):
         # A single node cluster should always leave a clean store.
@@ -375,7 +392,6 @@ class StoreTests(BrokerTest):
         self.assertEqual(a.get_message("q").content, "x")
 
     def test_last_store_clean(self):
-
         # Verify that only the last node in a cluster to shut down has
         # a clean store. Start with cluster of 3, reduce to 1 then
         # increase again to ensure that a node that was once alone but
@@ -394,13 +410,41 @@ class StoreTests(BrokerTest):
         time.sleep(0.1)   # pause for a to find out hes last.
         a.kill()          # really last
         # b & c should be dirty
-        b = cluster.start("b", wait=False, expect=EXPECT_EXIT_OK)
+        b = cluster.start("b", wait=False, expect=EXPECT_EXIT_FAIL)
         self.assert_dirty_store(b)
-        c = cluster.start("c", wait=False, expect=EXPECT_EXIT_OK)
+        c = cluster.start("c", wait=False, expect=EXPECT_EXIT_FAIL)
         self.assert_dirty_store(c)
         # a should be clean
         a = cluster.start("a")
         self.assertEqual(a.get_message("q").content, "x")
 
+    def test_restart_clean(self):
+        """Verify that we can re-start brokers one by one in a
+        persistent cluster after a clean oshutdown"""
+        cluster = self.cluster(0, self.args())
+        a = cluster.start("a", expect=EXPECT_EXIT_OK)
+        b = cluster.start("b", expect=EXPECT_EXIT_OK)
+        c = cluster.start("c", expect=EXPECT_EXIT_OK)
+        a.send_message("q", Message("x", durable=True))
+        self.stop_cluster(a)
+        a = cluster.start("a")
+        b = cluster.start("b")
+        c = cluster.start("c")
+        self.assertEqual(c.get_message("q").content, "x")
 
+    def test_join_sub_size(self):
+        """Verify that after starting a cluster with cluster-size=N,
+        we can join new members even if size < N-1"""
+        cluster = self.cluster(0, self.args())
+        a = cluster.start("a", wait=False, expect=EXPECT_EXIT_FAIL)
+        b = cluster.start("b", wait=False, expect=EXPECT_EXIT_FAIL)
+        c = cluster.start("c")
+        a.send_message("q", Message("x", durable=True))
+        a.send_message("q", Message("y", durable=True))
+        a.kill()
+        b.kill()
+        a = cluster.start("a")
+        self.assertEqual(c.get_message("q").content, "x")
+        b = cluster.start("b")
+        self.assertEqual(c.get_message("q").content, "y")