ZOOKEEPER-1872: QuorumPeer is not shutdown in few cases

(Rakesh R. via rgs)

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

CHANGES.txt

@@ -196,6 +196,9 @@ BUGFIXES:
   ZOOKEEPER-2142: JMX ObjectName is incorrect for observers (Edward Ribeiro
   via michim)
 
+  ZOOKEEPER-1872: QuorumPeer is not shutdown in few cases
+  (Rakesh R. via rgs)
+
 IMPROVEMENTS:
   ZOOKEEPER-1660 Documentation for Dynamic Reconfiguration (Reed Wanderman-Milne via shralex)  
 

src/java/test/org/apache/zookeeper/server/quorum/QuorumPeerMainTest.java

@@ -44,6 +44,7 @@ import org.apache.zookeeper.ZooKeeper.States;
 import org.apache.zookeeper.common.Time;
 import org.apache.zookeeper.server.quorum.Leader.Proposal;
 import org.apache.zookeeper.test.ClientBase;
+import org.junit.After;
 import org.junit.Assert;
 import org.junit.Test;
 
@@ -52,6 +53,23 @@ import org.junit.Test;
  *
  */
 public class QuorumPeerMainTest extends QuorumPeerTestBase {
+
+    private Servers servers;
+    private int numServers = 0;
+
+    @After
+    public void tearDown() throws Exception {
+        if (servers == null || servers.mt == null) {
+            LOG.info("No servers to shutdown!");
+            return;
+        }
+        for (int i = 0; i < numServers; i++) {
+            if (i < servers.mt.length) {
+                servers.mt[i].shutdown();
+            }
+        }
+    }
+
     /**
      * Verify the ability to start a cluster.
      */
@@ -226,14 +244,14 @@ public class QuorumPeerMainTest extends QuorumPeerTestBase {
      */
     @Test
     public void testHighestZxidJoinLate() throws Exception {
-        int numServers = 3;
-        Servers svrs = LaunchServers(numServers);
+        numServers = 3;
+        servers = LaunchServers(numServers);
         String path = "/hzxidtest";
         int leader = -1;
 
         // find the leader
         for (int i = 0; i < numServers; i++) {
-            if (svrs.mt[i].main.quorumPeer.leader != null) {
+            if (servers.mt[i].main.quorumPeer.leader != null) {
                 leader = i;
             }
         }
@@ -248,47 +266,47 @@ public class QuorumPeerMainTest extends QuorumPeerTestBase {
         byte[] output;
 
         // Create a couple of nodes
-        svrs.zk[leader].create(path + leader, input, Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
-        svrs.zk[leader].create(path + nonleader, input, Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
+        servers.zk[leader].create(path + leader, input, Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
+        servers.zk[leader].create(path + nonleader, input, Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
 
         // make sure the updates indeed committed. If it is not
         // the following statement will throw.
-        output = svrs.zk[leader].getData(path + nonleader, false, null);
+        output = servers.zk[leader].getData(path + nonleader, false, null);
 
         // Shutdown every one else but the leader
         for (int i = 0; i < numServers; i++) {
             if (i != leader) {
-                svrs.mt[i].shutdown();
+                servers.mt[i].shutdown();
             }
         }
 
         input[0] = 2;
 
         // Update the node on the leader
-        svrs.zk[leader].setData(path + leader, input, -1, null, null);
+        servers.zk[leader].setData(path + leader, input, -1, null, null);
 
         // wait some time to let this get written to disk
         Thread.sleep(500);
 
         // shut the leader down
-        svrs.mt[leader].shutdown();
+        servers.mt[leader].shutdown();
 
         System.gc();
 
-        waitForAll(svrs.zk, States.CONNECTING);
+        waitForAll(servers.zk, States.CONNECTING);
 
         // Start everyone but the leader
         for (int i = 0; i < numServers; i++) {
             if (i != leader) {
-                svrs.mt[i].start();
+                servers.mt[i].start();
             }
         }
 
         // wait to connect to one of these
-        waitForOne(svrs.zk[nonleader], States.CONNECTED);
+        waitForOne(servers.zk[nonleader], States.CONNECTED);
 
         // validate that the old value is there and not the new one
-        output = svrs.zk[nonleader].getData(path + leader, false, null);
+        output = servers.zk[nonleader].getData(path + leader, false, null);
 
         Assert.assertEquals(
                 "Expecting old value 1 since 2 isn't committed yet",
@@ -296,22 +314,22 @@ public class QuorumPeerMainTest extends QuorumPeerTestBase {
 
         // Do some other update, so we bump the maxCommttedZxid
         // by setting the value to 2
-        svrs.zk[nonleader].setData(path + nonleader, input, -1);
+        servers.zk[nonleader].setData(path + nonleader, input, -1);
 
         // start the old leader
-        svrs.mt[leader].start();
+        servers.mt[leader].start();
 
         // connect to it
-        waitForOne(svrs.zk[leader], States.CONNECTED);
+        waitForOne(servers.zk[leader], States.CONNECTED);
 
         // make sure it doesn't have the new value that it alone had logged
-        output = svrs.zk[leader].getData(path + leader, false, null);
+        output = servers.zk[leader].getData(path + leader, false, null);
         Assert.assertEquals(
                 "Validating that the deposed leader has rolled back that change it had written",
                 output[0], 1);
 
         // make sure the leader has the subsequent changes that were made while it was offline
-        output = svrs.zk[leader].getData(path + nonleader, false, null);
+        output = servers.zk[leader].getData(path + nonleader, false, null);
         Assert.assertEquals(
                 "Validating that the deposed leader caught up on changes it missed",
                 output[0], 2);

src/java/test/org/apache/zookeeper/test/CnxManagerTest.java

@@ -368,41 +368,43 @@ public class CnxManagerTest extends ZKTestCase {
     @Test
     public void testWorkerThreads() throws Exception {
         ArrayList<QuorumPeer> peerList = new ArrayList<QuorumPeer>();
-
-        for (int sid = 0; sid < 3; sid++) {
-            QuorumPeer peer = new QuorumPeer(peers, peerTmpdir[sid], peerTmpdir[sid],
-                                             peerClientPort[sid], 3, sid, 1000, 2, 2);
-            LOG.info("Starting peer " + peer.getId());
-            peer.start();
-            peerList.add(sid, peer);
-        }
-        String failure = verifyThreadCount(peerList, 4);
-        if (failure != null) {
-            Assert.fail(failure);
-        }
-        for (int myid = 0; myid < 3; myid++) {
-            for (int i = 0; i < 5; i++) {
-                // halt one of the listeners and verify count
-                QuorumPeer peer = peerList.get(myid);
-                LOG.info("Round " + i + ", halting peer " + peer.getId());
-                peer.shutdown();
-                peerList.remove(myid);
-                failure = verifyThreadCount(peerList, 2);
-                if (failure != null) {
-                    Assert.fail(failure);
-                }
-
-                // Restart halted node and verify count
-                peer = new QuorumPeer(peers, peerTmpdir[myid], peerTmpdir[myid],
-                                        peerClientPort[myid], 3, myid, 1000, 2, 2);
-                LOG.info("Round " + i + ", restarting peer " + peer.getId());
+        try {
+            for (int sid = 0; sid < 3; sid++) {
+                QuorumPeer peer = new QuorumPeer(peers, peerTmpdir[sid],
+                        peerTmpdir[sid], peerClientPort[sid], 3, sid, 1000, 2,
+                        2);
+                LOG.info("Starting peer {}", peer.getId());
                 peer.start();
-                peerList.add(myid, peer);
-                failure = verifyThreadCount(peerList, 4);
-                if (failure != null) {
-                    Assert.fail(failure);
+                peerList.add(sid, peer);
+            }
+            String failure = verifyThreadCount(peerList, 4);
+            Assert.assertNull(failure, failure);
+            for (int myid = 0; myid < 3; myid++) {
+                for (int i = 0; i < 5; i++) {
+                    // halt one of the listeners and verify count
+                    QuorumPeer peer = peerList.get(myid);
+                    LOG.info("Round {}, halting peer ",
+                            new Object[] { i, peer.getId() });
+                    peer.shutdown();
+                    peerList.remove(myid);
+                    failure = verifyThreadCount(peerList, 2);
+                    Assert.assertNull(failure, failure);
+                    // Restart halted node and verify count
+                    peer = new QuorumPeer(peers, peerTmpdir[myid],
+                            peerTmpdir[myid], peerClientPort[myid], 3, myid,
+                            1000, 2, 2);
+                    LOG.info("Round {}, restarting peer ",
+                            new Object[] { i, peer.getId() });
+                    peer.start();
+                    peerList.add(myid, peer);
+                    failure = verifyThreadCount(peerList, 4);
+                    Assert.assertNull(failure, failure);
                 }
             }
+        } finally {
+            for (QuorumPeer quorumPeer : peerList) {
+                quorumPeer.shutdown();
+            }
         }
     }
 

src/java/test/org/apache/zookeeper/test/JMXEnv.java

@@ -60,14 +60,18 @@ public class JMXEnv {
     
     public static void tearDown() {
         try {
-            cc.close();
+            if (cc != null) {
+                cc.close();
+            }
         } catch (IOException e) {
             LOG.warn("Unexpected, ignoring", e);
             
         }
         cc = null;
         try {
-            cs.stop();
+            if (cs != null) {
+                cs.stop();
+            }
         } catch (IOException e) {
             LOG.warn("Unexpected, ignoring", e);
             

src/java/test/org/apache/zookeeper/test/LeaderSessionTrackerTest.java

@@ -85,12 +85,10 @@ public class LeaderSessionTrackerTest extends ZKTestCase implements Watcher {
      * is not in closing state
      */
     public void testCreateEphemeral(boolean localSessionEnabled) throws Exception {
-        QuorumUtil qu = new QuorumUtil(1);
         if (localSessionEnabled) {
             qu.enableLocalSession(true);
         }
         qu.startAll();
-
         QuorumPeer leader = qu.getLeaderQuorumPeer();
 
         ZooKeeper zk = new ZooKeeper(qu.getConnectString(leader),
@@ -131,7 +129,6 @@ public class LeaderSessionTrackerTest extends ZKTestCase implements Watcher {
      */
     @Test
     public void testCreatePersistent() throws Exception {
-        QuorumUtil qu = new QuorumUtil(1);
         qu.enableLocalSession(true);
         qu.startAll();
 

src/java/test/org/apache/zookeeper/test/QuorumTest.java

@@ -53,6 +53,7 @@ public class QuorumTest extends ZKTestCase {
 
     private final QuorumBase qb = new QuorumBase();
     private final ClientTest ct = new ClientTest();
+    private QuorumUtil qu;
 
     @Before
     public void setUp() throws Exception {
@@ -65,6 +66,9 @@ public class QuorumTest extends ZKTestCase {
     public void tearDown() throws Exception {
         ct.tearDownAll();
         qb.tearDown();
+        if (qu != null) {
+            qu.tearDown();
+        }
     }
 
     @Test
@@ -296,7 +300,7 @@ public class QuorumTest extends ZKTestCase {
      * */
     @Test
     public void testFollowersStartAfterLeader() throws Exception {
-        QuorumUtil qu = new QuorumUtil(1);
+        qu = new QuorumUtil(1);
         CountdownWatcher watcher = new CountdownWatcher();
         qu.startQuorum();
 
@@ -346,11 +350,10 @@ public class QuorumTest extends ZKTestCase {
      * 
      */
     @Test
-    public void testNoLogBeforeLeaderEstablishment () 
-    throws IOException, InterruptedException, KeeperException{
+    public void testNoLogBeforeLeaderEstablishment () throws Exception {
         final Semaphore sem = new Semaphore(0);
 
-        QuorumUtil qu = new QuorumUtil(2, 10);
+        qu = new QuorumUtil(2, 10);
         qu.startQuorum();
 
         int index = 1;
@@ -411,6 +414,7 @@ public class QuorumTest extends ZKTestCase {
         Assert.assertTrue("Zxid: " + qu.getPeer(index).peer.getActiveServer().getZxid() + 
                 "Current epoch: " + epochF, epochF == epochL);
 
+        zk.close();
     }
 
     // skip superhammer and clientcleanup as they are too expensive for quorum
@@ -426,7 +430,7 @@ public class QuorumTest extends ZKTestCase {
      */
     @Test
     public void testMultiToFollower() throws Exception {
-        QuorumUtil qu = new QuorumUtil(1);
+        qu = new QuorumUtil(1);
         CountdownWatcher watcher = new CountdownWatcher();
         qu.startQuorum();