ZOOKEEPER-1810. Add version to FLE notifications for trunk Germán Blanco via michim)

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

CHANGES.txt

@@ -683,6 +683,9 @@ BUGFIXES:
   ZOOKEEPER-1835. dynamic configuration file renaming fails on Windows
   (Bruno Freudensprung via rakeshr)
 
+  ZOOKEEPER-1810. Add version to FLE notifications for trunk Germán Blanco via
+  michim)
+
 IMPROVEMENTS:
 
   ZOOKEEPER-1170. Fix compiler (eclipse) warnings: unused imports,

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

@@ -70,7 +70,7 @@ public class QuorumCnxManager {
     // stale notifications to peers
     static final int SEND_CAPACITY = 1;
 
-    static final int PACKETMAXSIZE = 1024 * 1024; 
+    static final int PACKETMAXSIZE = 1024 * 512;
     /*
      * Maximum number of attempts to connect to a peer
      */

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

@@ -810,7 +810,9 @@ public class QuorumPeer extends ZooKeeperThread implements QuorumStats.Provider
             QuorumCnxManager.Listener listener = qcm.listener;
             if(listener != null){
                 listener.start();
-                le = new FastLeaderElection(this, qcm);
+                FastLeaderElection fle = new FastLeaderElection(this, qcm);
+                fle.start();
+                le = fle;
             } else {
                 LOG.error("Null listener when initializing cnx manager");
             }

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

@@ -23,7 +23,9 @@ import org.apache.zookeeper.server.quorum.QuorumPeer.ServerState;
 
 public class Vote {
     
-    public Vote(long id, long zxid) {
+    public Vote(long id,
+                    long zxid) {
+        this.version = 0x0;
         this.id = id;
         this.zxid = zxid;
         this.electionEpoch = -1;
@@ -31,7 +33,10 @@ public class Vote {
         this.state = ServerState.LOOKING;
     }
     
-    public Vote(long id, long zxid, long peerEpoch) {
+    public Vote(long id,
+                    long zxid,
+                    long peerEpoch) {
+        this.version = 0x0;
         this.id = id;
         this.zxid = zxid;
         this.electionEpoch = -1;
@@ -39,7 +44,11 @@ public class Vote {
         this.state = ServerState.LOOKING;
     }
 
-    public Vote(long id, long zxid, long electionEpoch, long peerEpoch) {
+    public Vote(long id,
+                    long zxid,
+                    long electionEpoch,
+                    long peerEpoch) {
+        this.version = 0x0;
         this.id = id;
         this.zxid = zxid;
         this.electionEpoch = electionEpoch;
@@ -47,7 +56,13 @@ public class Vote {
         this.state = ServerState.LOOKING;
     }
     
-    public Vote(long id, long zxid, long electionEpoch, long peerEpoch, ServerState state) {
+    public Vote(int version,
+                    long id,
+                    long zxid,
+                    long electionEpoch,
+                    long peerEpoch,
+                    ServerState state) {
+        this.version = version;
         this.id = id;
         this.zxid = zxid;
         this.electionEpoch = electionEpoch;
@@ -55,6 +70,21 @@ public class Vote {
         this.peerEpoch = peerEpoch;
     }
     
+    public Vote(long id,
+                    long zxid,
+                    long electionEpoch,
+                    long peerEpoch,
+                    ServerState state) {
+        this.id = id;
+        this.zxid = zxid;
+        this.electionEpoch = electionEpoch;
+        this.state = state;
+        this.peerEpoch = peerEpoch;
+        this.version = 0x0;
+    }
+
+    final private int version;
+
     final private long id;
     
     final private long zxid;
@@ -63,6 +93,10 @@ public class Vote {
     
     final private long peerEpoch;
     
+    public int getVersion() {
+        return version;
+    }
+
     public long getId() {
         return id;
     }
@@ -91,7 +125,10 @@ public class Vote {
             return false;
         }
         Vote other = (Vote) o;
-        return (id == other.id && zxid == other.zxid && electionEpoch == other.electionEpoch && peerEpoch == other.peerEpoch);
+        return (id == other.id
+                    && zxid == other.zxid
+                    && electionEpoch == other.electionEpoch
+                    && peerEpoch == other.peerEpoch);
 
     }
 

src/java/test/org/apache/zookeeper/test/FLEBackwardElectionRoundTest.java → src/java/test/org/apache/zookeeper/server/quorum/FLEBackwardElectionRoundTest.java

@@ -15,7 +15,7 @@
  * limitations under the License.
  */
 
-package org.apache.zookeeper.test;
+package org.apache.zookeeper.server.quorum;
 
 import java.io.File;
 import java.io.IOException;
@@ -32,16 +32,15 @@ import org.apache.zookeeper.server.quorum.QuorumPeer;
 import org.apache.zookeeper.server.quorum.Vote;
 import org.apache.zookeeper.server.quorum.QuorumPeer.QuorumServer;
 import org.apache.zookeeper.server.quorum.QuorumPeer.ServerState;
+import org.apache.zookeeper.test.ClientBase;
 import org.junit.After;
 import org.junit.Assert;
 import org.junit.Before;
 import org.junit.Test;
 
-import org.apache.zookeeper.test.FLETestUtils.LEThread;
-
 public class FLEBackwardElectionRoundTest extends ZKTestCase {
     protected static final Logger LOG = LoggerFactory.getLogger(FLELostMessageTest.class);
-    
+
     int count;
     HashMap<Long,QuorumServer> peers;
     File tmpdir[];
@@ -67,94 +66,85 @@ public class FLEBackwardElectionRoundTest extends ZKTestCase {
             }
         }
     }
-    
+
     /**
      * This test is checking the following case. A server S is
-     * currently LOOKING and it receives notifications from 
+     * currently LOOKING and it receives notifications from
      * a quorum indicating they are following S. The election
-     * round E of S is higher than the election round E' in the 
+     * round E of S is higher than the election round E' in the
      * notification messages, so S becomes the leader and sets
      * its epoch back to E'. In the meanwhile, one or more
      * followers turn to LOOKING and elect S in election round E.
      * Having leader and followers with different election rounds
      * might prevent other servers from electing a leader because
-     * they can't get a consistent set of notifications from a 
-     * quorum. 
-     * 
+     * they can't get a consistent set of notifications from a
+     * quorum.
+     *
      * {@link https://issues.apache.org/jira/browse/ZOOKEEPER-1514}
-     *    
-     * 
+     *
+     *
      * @throws Exception
      */
-    
+
     @Test
     public void testBackwardElectionRound() throws Exception {
-        LOG.info("TestLE: " + getTestName()+ ", " + count);
+        LOG.info("TestLE: {}, {}", getTestName(), count);
         for(int i = 0; i < count; i++) {
             int clientport = PortAssignment.unique();
             peers.put(Long.valueOf(i),
-                new QuorumServer(i,
-                    new InetSocketAddress("127.0.0.1", clientport),
-                    new InetSocketAddress(
-                        "127.0.0.1", PortAssignment.unique())));
+                    new QuorumServer(i,
+                            new InetSocketAddress(clientport),
+                            new InetSocketAddress(PortAssignment.unique())));
             tmpdir[i] = ClientBase.createTmpDir();
             port[i] = clientport;
         }
 
+        ByteBuffer initialMsg = FLETestUtils.createMsg(ServerState.FOLLOWING.ordinal(), 0, 0, 1);
+
         /*
          * Start server 0
          */
-
         QuorumPeer peer = new QuorumPeer(peers, tmpdir[0], tmpdir[0], port[0], 3, 0, 1000, 2, 2);
         peer.startLeaderElection();
         FLETestUtils.LEThread thread = new FLETestUtils.LEThread(peer, 0);
-        thread.start();  
-        
-        
+        thread.start();
+
         /*
          * Start mock server 1
          */
         QuorumPeer mockPeer = new QuorumPeer(peers, tmpdir[1], tmpdir[1], port[1], 3, 1, 1000, 2, 2);
         cnxManagers[0] = new QuorumCnxManager(mockPeer);
-        QuorumCnxManager.Listener listener = cnxManagers[0].listener;
-        listener.start();
+        cnxManagers[0].listener.start();
+
+        cnxManagers[0].toSend(0l, initialMsg);
 
-        cnxManagers[0].toSend(0l, FLETestUtils.createMsg(ServerState.FOLLOWING.ordinal(), 0, 0, 1));
-        
         /*
          * Start mock server 2
          */
         mockPeer = new QuorumPeer(peers, tmpdir[2], tmpdir[2], port[2], 3, 2, 1000, 2, 2);
         cnxManagers[1] = new QuorumCnxManager(mockPeer);
-        listener = cnxManagers[1].listener;
-        listener.start();
+        cnxManagers[1].listener.start();
+
+        cnxManagers[1].toSend(0l, initialMsg);
 
-        cnxManagers[1].toSend(0l, FLETestUtils.createMsg(ServerState.FOLLOWING.ordinal(), 0, 0, 1));
-        
         /*
          * Run another instance of leader election.
          */
         thread.join(5000);
-        Assert.assertTrue("State is not leading. Current state:"
-                + peer.getPeerState(),
-                peer.getPeerState() == ServerState.LEADING);
         thread = new FLETestUtils.LEThread(peer, 0);
         thread.start();
-        
+
         /*
          * Send the same messages, this time should not make 0 the leader.
          */
-        cnxManagers[0].toSend(0l, FLETestUtils.createMsg(ServerState.FOLLOWING.ordinal(), 0, 0, 1));
-        cnxManagers[1].toSend(0l, FLETestUtils.createMsg(ServerState.FOLLOWING.ordinal(), 0, 0, 1));
-        
-        
+        cnxManagers[0].toSend(0l, initialMsg);
+        cnxManagers[1].toSend(0l, initialMsg);
+
         thread.join(5000);
-        Assert.assertTrue("State is not looking. Current state:"
-                + peer.getPeerState(),
-                peer.getPeerState() == ServerState.LOOKING);
+
         if (!thread.isAlive()) {
             Assert.fail("Should not have joined");
         }
-        
+
     }
 }

src/java/test/org/apache/zookeeper/test/FLELostMessageTest.java → src/java/test/org/apache/zookeeper/server/quorum/FLELostMessageTest.java

@@ -16,23 +16,23 @@
  * limitations under the License.
  */
 
-package org.apache.zookeeper.test;
+package org.apache.zookeeper.server.quorum;
 
 import java.io.File;
 import java.io.IOException;
 import java.net.InetSocketAddress;
-import java.nio.ByteBuffer;
 import java.util.HashMap;
 
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.apache.zookeeper.PortAssignment;
 import org.apache.zookeeper.ZKTestCase;
+import org.apache.zookeeper.server.quorum.FastLeaderElection;
 import org.apache.zookeeper.server.quorum.QuorumCnxManager;
 import org.apache.zookeeper.server.quorum.QuorumPeer;
-import org.apache.zookeeper.server.quorum.Vote;
 import org.apache.zookeeper.server.quorum.QuorumPeer.QuorumServer;
 import org.apache.zookeeper.server.quorum.QuorumPeer.ServerState;
+import org.apache.zookeeper.test.ClientBase;
 import org.junit.After;
 import org.junit.Assert;
 import org.junit.Before;
@@ -41,7 +41,6 @@ import org.junit.Test;
 public class FLELostMessageTest extends ZKTestCase {
     protected static final Logger LOG = LoggerFactory.getLogger(FLELostMessageTest.class);
 
-
     int count;
     HashMap<Long,QuorumServer> peers;
     File tmpdir[];
@@ -62,18 +61,16 @@ public class FLELostMessageTest extends ZKTestCase {
     public void tearDown() throws Exception {
         cnxManager.halt();
     }
-    
+
     @Test
     public void testLostMessage() throws Exception {
-
-        LOG.info("TestLE: " + getTestName()+ ", " + count);
+        LOG.info("TestLE: {}, {}", getTestName(), count);
         for(int i = 0; i < count; i++) {
             int clientport = PortAssignment.unique();
             peers.put(Long.valueOf(i),
-                new QuorumServer(i,
-                    new InetSocketAddress("127.0.0.1", clientport),
-                    new InetSocketAddress(
-                        "127.0.0.1", PortAssignment.unique())));
+                    new QuorumServer(i,
+                            new InetSocketAddress(clientport),
+                            new InetSocketAddress(PortAssignment.unique())));
             tmpdir[i] = ClientBase.createTmpDir();
             port[i] = clientport;
         }
@@ -81,7 +78,6 @@ public class FLELostMessageTest extends ZKTestCase {
         /*
          * Start server 0
          */
-
         QuorumPeer peer = new QuorumPeer(peers, tmpdir[1], tmpdir[1], port[1], 3, 1, 1000, 2, 2);
         peer.startLeaderElection();
         FLETestUtils.LEThread thread = new FLETestUtils.LEThread(peer, 1);
@@ -92,26 +88,18 @@ public class FLELostMessageTest extends ZKTestCase {
          */
         mockServer();
         thread.join(5000);
-        Assert.assertTrue("State is not leading. Current state:"
-                + peer.getPeerState(),
-                peer.getPeerState() == ServerState.LEADING);
         if (thread.isAlive()) {
             Assert.fail("Threads didn't join");
         }
     }
 
     void mockServer() throws InterruptedException, IOException {
-        /*
-         * Create an instance of the connection manager
-         */
         QuorumPeer peer = new QuorumPeer(peers, tmpdir[0], tmpdir[0], port[0], 3, 0, 1000, 2, 2);
         cnxManager = new QuorumCnxManager(peer);
-        QuorumCnxManager.Listener listener = cnxManager.listener;
-        listener.start();
-
+        cnxManager.listener.start();
 
-        cnxManager.toSend(1l, FLETestUtils.createMsg(ServerState.LOOKING.ordinal(), 0, 0, 1));
+        cnxManager.toSend(1l, FLETestUtils.createMsg(ServerState.LOOKING.ordinal(), 0, 0, 0));
         cnxManager.recvQueue.take();
-        cnxManager.toSend(1L, FLETestUtils.createMsg(ServerState.FOLLOWING.ordinal(), 1, 0, 1));
+        cnxManager.toSend(1L, FLETestUtils.createMsg(ServerState.FOLLOWING.ordinal(), 1, 0, 0));
     }
 }

src/java/test/org/apache/zookeeper/test/FLETestUtils.java → src/java/test/org/apache/zookeeper/server/quorum/FLETestUtils.java

@@ -15,10 +15,11 @@
  * limitations under the License.
  */
 
-package org.apache.zookeeper.test;
+package org.apache.zookeeper.server.quorum;
 
 import java.nio.ByteBuffer;
 
+import org.apache.zookeeper.server.quorum.FastLeaderElection;
 import org.apache.zookeeper.server.quorum.QuorumPeer;
 import org.apache.zookeeper.server.quorum.Vote;
 
@@ -30,8 +31,7 @@ import org.apache.zookeeper.server.quorum.QuorumPeer.ServerState;
 
 public class FLETestUtils {
     protected static final Logger LOG = LoggerFactory.getLogger(FLETestUtils.class);
-    
-    
+
     /*
      * Thread to run an instance of leader election for 
      * a given quorum peer.
@@ -43,18 +43,18 @@ public class FLETestUtils {
         LEThread(QuorumPeer peer, int i) {
             this.i = i;
             this.peer = peer;
-            LOG.info("Constructor: " + getName());
+            LOG.info("Constructor: {}", getName());
 
         }
 
-        public void run(){
-            try{
+        public void run() {
+            try {
                 Vote v = null;
                 peer.setPeerState(ServerState.LOOKING);
-                LOG.info("Going to call leader election: " + i);
+                LOG.info("Going to call leader election: {}", i);
                 v = peer.getElectionAlg().lookForLeader();
 
-                if (v == null){
+                if (v == null) {
                     Assert.fail("Thread " + i + " got a null vote");
                 }
 
@@ -64,34 +64,21 @@ public class FLETestUtils {
                  */
                 peer.setCurrentVote(v);
 
-                LOG.info("Finished election: " + i + ", " + v.getId());
-                LOG.info("QuorumPeer state: " + peer.getPeerState());
+                LOG.info("Finished election: {}, {}", i, v.getId());
+
+                Assert.assertTrue("State is not leading.", peer.getPeerState() == ServerState.LEADING);
             } catch (Exception e) {
                 e.printStackTrace();
             }
             LOG.info("Joining");
         }
     }
-    
+
     /*
      * Creates a leader election notification message.
      */
-    
     static ByteBuffer createMsg(int state, long leader, long zxid, long epoch){
-        byte requestBytes[] = new byte[28];
-        ByteBuffer requestBuffer = ByteBuffer.wrap(requestBytes);
-
-        /*
-         * Building notification packet to send
-         */
-
-        requestBuffer.clear();
-        requestBuffer.putInt(state);
-        requestBuffer.putLong(leader);
-        requestBuffer.putLong(zxid);
-        requestBuffer.putLong(epoch);
-
-        return requestBuffer;
+        return FastLeaderElection.buildMsg(state, leader, zxid, 1, epoch);
     }
 
-}
\ No newline at end of file
+}

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

@@ -152,8 +152,6 @@ public class FLENewEpochTest extends ZKTestCase {
       @Test
       public void testLENewEpoch() throws Exception {
 
-          FastLeaderElection le[] = new FastLeaderElection[count];
-
           LOG.info("TestLE: " + getTestName()+ ", " + count);
           for(int i = 0; i < count; i++) {
               peers.put(Long.valueOf(i),
@@ -166,7 +164,7 @@ public class FLENewEpochTest extends ZKTestCase {
               port[i] = PortAssignment.unique();
           }
 
-          for(int i = 1; i < le.length; i++) {
+          for(int i = 1; i < count; i++) {
               QuorumPeer peer = new QuorumPeer(peers, tmpdir[i], tmpdir[i], port[i], 3, i, 1000, 2, 2);
               peer.startLeaderElection();
               LEThread thread = new LEThread(peer, i);

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

@@ -78,6 +78,7 @@ public class FLEPredicateTest extends ZKTestCase {
                                         PortAssignment.unique(), 3, 0, 1000, 2, 2);
         
             MockFLE mock = new MockFLE(peer);
+            mock.start();
             
             /*
              * Lower epoch must return false

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

@@ -297,7 +297,6 @@ public class FLETest extends ZKTestCase {
      * @throws Exception
      */
     private void runElection(int rounds) throws Exception {
-        FastLeaderElection le[] = new FastLeaderElection[count];
         ConcurrentHashMap<Long, HashSet<Integer> > quora = 
             new ConcurrentHashMap<Long, HashSet<Integer> >();
 
@@ -322,7 +321,7 @@ public class FLETest extends ZKTestCase {
         /*
          * Start one LEThread for each peer we want to run.
          */
-        for(int i = 0; i < le.length; i++) {
+        for(int i = 0; i < count; i++) {
             QuorumPeer peer = new QuorumPeer(peers, tmpdir[i], tmpdir[i],
                     port[i], 3, i, 1000, 2, 2);
             peer.startLeaderElection();

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

@@ -140,8 +140,6 @@ public class FLEZeroWeightTest extends ZKTestCase {
 
     @Test
     public void testZeroWeightQuorum() throws Exception {
-        FastLeaderElection le[] = new FastLeaderElection[count];
-
         LOG.info("TestZeroWeightQuorum: " + getTestName()+ ", " + count);
         for(int i = 0; i < count; i++) {
             InetSocketAddress addr1 = new InetSocketAddress("127.0.0.1",PortAssignment.unique());
@@ -153,7 +151,7 @@ public class FLEZeroWeightTest extends ZKTestCase {
             tmpdir[i] = ClientBase.createTmpDir();
         }
 
-        for(int i = 0; i < le.length; i++) {
+        for(int i = 0; i < count; i++) {
             QuorumHierarchical hq = new QuorumHierarchical(qp);
             QuorumPeer peer = new QuorumPeer(peers, tmpdir[i], tmpdir[i], port[i], 3, i, 1000, 2, 2, hq);
             peer.startLeaderElection();

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

@@ -37,6 +37,7 @@ import org.apache.zookeeper.PortAssignment;
 import org.apache.zookeeper.ZKTestCase;
 import org.apache.zookeeper.server.ServerCnxnFactory;
 import org.apache.zookeeper.server.quorum.Election;
+import org.apache.zookeeper.server.quorum.FLELostMessageTest;
 import org.apache.zookeeper.server.quorum.LeaderElection;
 import org.apache.zookeeper.server.quorum.QuorumPeer;
 import org.apache.zookeeper.server.quorum.Vote;