ZOOKEEPER-131. Fix Old leader election can elect a dead leader over and over again. (breed via mahadev)

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

CHANGES.txt

@@ -61,3 +61,6 @@ Backward compatibile changes:
 
  ZOOKEEPER-130. update build.xml to support apache release process. 
  (phunt via mahadev)
+
+ ZOOKEEPER-131. Fix Old leader election can elect a dead leader over and over
+ again. (breed via mahadev)

src/java/main/org/apache/zookeeper/server/quorum/LeaderElection.java

@@ -26,6 +26,8 @@ import java.net.SocketException;
 import java.nio.ByteBuffer;
 import java.util.Collection;
 import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Iterator;
 import java.util.Random;
 import java.util.Map.Entry;
 
@@ -54,7 +56,7 @@ public class LeaderElection implements Election  {
         public int winningCount;
     }
 
-    private ElectionResult countVotes(HashMap<InetSocketAddress, Vote> votes) {
+    private ElectionResult countVotes(HashMap<InetSocketAddress, Vote> votes, HashSet<Long> heardFrom) {
         ElectionResult result = new ElectionResult();
         // Initialize with null vote
         result.vote = new Vote(Long.MIN_VALUE, Long.MIN_VALUE);
@@ -62,7 +64,13 @@ public class LeaderElection implements Election  {
         Collection<Vote> votesCast = votes.values();
         // First make the views consistent. Sometimes peers will have
         // different zxids for a server depending on timing.
-        for (Vote v : votesCast) {
+        for (Iterator<Vote> i = votesCast.iterator(); i.hasNext();) {
+            Vote v = i.next();
+            if (!heardFrom.contains(v.id)) {
+                // Discard votes for machines that we didn't hear from
+                i.remove();
+                continue;
+            }
             for (Vote w : votesCast) {
                 if (v.id == w.id) {
                     if (v.zxid < w.zxid) {
@@ -71,6 +79,7 @@ public class LeaderElection implements Election  {
                 }
             }
         }
+        
         HashMap<Vote, Integer> countTable = new HashMap<Vote, Integer>();
         // Now do the tally
         for (Vote v : votesCast) {
@@ -127,6 +136,7 @@ public class LeaderElection implements Election  {
             requestBuffer.clear();
             requestBuffer.putInt(xid);
             requestPacket.setLength(4);
+            HashSet<Long> heardFrom = new HashSet<Long>();
             for (QuorumServer server : self.quorumPeers) {
                 requestPacket.setSocketAddress(server.addr);
                 try {
@@ -145,7 +155,8 @@ public class LeaderElection implements Election  {
                                 + " got " + recvedXid);
                         continue;
                     }
-                    responseBuffer.getLong();
+                    long peerId = responseBuffer.getLong();
+                    heardFrom.add(peerId);
                     //if(server.id != peerId){
                         Vote vote = new Vote(responseBuffer.getLong(),
                             responseBuffer.getLong());
@@ -154,12 +165,13 @@ public class LeaderElection implements Election  {
                         votes.put(addr, vote);
                     //}
                 } catch (IOException e) {
+                    LOG.error("Error in looking for leader", e);
                     // Errors are okay, since hosts may be
                     // down
                     // ZooKeeperServer.logException(e);
                 }
             }
-            ElectionResult result = countVotes(votes);
+            ElectionResult result = countVotes(votes, heardFrom);
             if (result.winner.id >= 0) {
                 self.setCurrentVote(result.vote);
                 if (result.winningCount > (self.quorumPeers.size() / 2)) {

src/java/main/org/apache/zookeeper/server/quorum/QuorumPeer.java

@@ -160,13 +160,15 @@ public class QuorumPeer extends Thread implements QuorumStats.Provider {
             super("ResponderThread");
         }
 
+        volatile boolean running = true;
+        
         @Override
         public void run() {
             try {
                 byte b[] = new byte[36];
                 ByteBuffer responseBuffer = ByteBuffer.wrap(b);
                 DatagramPacket packet = new DatagramPacket(b, b.length);
-                while (true) {
+                while (running) {
                     udpSocket.receive(packet);
                     if (packet.getLength() != 4) {
                         LOG.warn("Got more than just an xid! Len = "
@@ -273,7 +275,17 @@ public class QuorumPeer extends Thread implements QuorumStats.Provider {
 
     @Override
     public synchronized void start() {
-        
+        startLeaderElection();
+        super.start();
+    }
+
+    ResponderThread responder;
+    
+    public void stopLeaderElection() {
+        responder.running = false;
+        responder.interrupt();
+    }
+    public void startLeaderElection() {
         currentVote = new Vote(myid, getLastLoggedZxid());
         for (QuorumServer p : quorumPeers) {
             if (p.id == myid) {
@@ -287,13 +299,14 @@ public class QuorumPeer extends Thread implements QuorumStats.Provider {
         if (electionType == 0) {
             try {
                 udpSocket = new DatagramSocket(myQuorumAddr.getPort());
-                new ResponderThread().start();
+                responder = new ResponderThread();
+                responder.start();
             } catch (SocketException e) {
                 throw new RuntimeException(e);
             }
         }
         this.electionAlg = createElectionAlgorithm(electionType);
-        super.start();
+       
     }
     
     /**

src/java/main/org/apache/zookeeper/server/quorum/Vote.java

@@ -44,4 +44,7 @@ public class Vote {
         return (int) (id & zxid);
     }
 
+    public String toString() {
+        return "(" + id + ", " + Long.toHexString(zxid) + ")";
+    }
 }

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

@@ -0,0 +1,131 @@
+/**
+ * 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.
+ */
+
+package org.apache.zookeeper.test;
+
+import java.io.File;
+import java.net.InetSocketAddress;
+import java.util.ArrayList;
+import java.util.Random;
+
+import org.apache.log4j.Logger;
+import org.apache.zookeeper.server.quorum.LeaderElection;
+import org.apache.zookeeper.server.quorum.QuorumPeer;
+import org.apache.zookeeper.server.quorum.QuorumStats;
+import org.apache.zookeeper.server.quorum.Vote;
+import org.apache.zookeeper.server.quorum.QuorumPeer.QuorumServer;
+
+import junit.framework.TestCase;
+
+public class LETest extends TestCase {
+    volatile Vote votes[];
+    volatile boolean leaderDies;
+    volatile long leader = -1; 
+    Random rand = new Random();
+    class LEThread extends Thread {
+        LeaderElection le;
+        int i;
+        QuorumPeer peer;
+        LEThread(LeaderElection le, QuorumPeer peer, int i) {
+            this.le = le;
+            this.i = i;
+            this.peer = peer;
+        }
+        public void run() {
+            try {
+                Vote v = null;
+                while(true) {
+                    v = le.lookForLeader();
+                    votes[i] = v;
+                    if (v.id == i) {
+                        synchronized(LETest.this) {
+                            if (leaderDies) {
+                                leaderDies = false;
+                                peer.stopLeaderElection();
+                                System.out.println("Leader " + i + " dying");
+                                leader = -2;
+                            } else {
+                                leader = i; 
+                            }
+                            LETest.this.notifyAll();
+                        }
+                        break;
+                    }
+                    synchronized(LETest.this) {
+                        if (leader == -1) {
+                            LETest.this.wait();
+                        }
+                        if (leader == v.id) {
+                            break;
+                        }
+                    }
+                    Thread.sleep(rand.nextInt(1000));
+                    peer.setCurrentVote(new Vote(peer.getId(), 0));
+                }
+                System.out.println("Thread " + i + " votes " + v);
+            } catch (InterruptedException e) {
+                e.printStackTrace();
+            }
+        }
+    }
+    public void testLE() throws Exception {
+        int count = 30;
+        int baseport= 33003;
+        ArrayList<QuorumServer> peers = new ArrayList<QuorumServer>(count);
+        ArrayList<LEThread> threads = new ArrayList<LEThread>(count);
+        File tmpdir[] = new File[count];
+        int port[] = new int[count];
+        votes = new Vote[count];
+        QuorumStats.registerAsConcrete();
+        for(int i = 0; i < count; i++) {
+            peers.add(new QuorumServer(i, new InetSocketAddress(baseport+100+i)));
+            tmpdir[i] = File.createTempFile("letest", "test");
+            port[i] = baseport+i;    
+        }
+        LeaderElection le[] = new LeaderElection[count];
+        leaderDies = true;
+        boolean allowOneBadLeader = leaderDies;
+        for(int i = 0; i < le.length; i++) {
+            QuorumPeer peer = new QuorumPeer(peers, tmpdir[i], tmpdir[i], port[i], 0, 0, i, 2, 2, 2);
+            peer.startLeaderElection();
+            le[i] = new LeaderElection(peer);
+            LEThread thread = new LEThread(le[i], peer, i);
+            thread.start();
+            threads.add(thread);
+        }
+        for(int i = 0; i < threads.size(); i++) {
+            threads.get(i).join(15000);
+            if (threads.get(i).isAlive()) {
+                fail("Threads didn't join");
+            }
+        }
+        long id = votes[0].id;
+        for(int i = 1; i < votes.length; i++) {
+            if (votes[i] == null) {
+                fail("Thread " + i + " had a null vote");
+            }
+            if (votes[i].id != id) {
+                if (allowOneBadLeader && votes[i].id == i) {
+                    allowOneBadLeader = false;
+                } else {
+                    fail("Thread " + i + " got " + votes[i].id + " expected " + id);
+                }
+            }
+        }
+    }
+}