ZOOKEEPER-499. electionAlg should default to FLE (3) - regression (phunt via mahadev)

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

CHANGES.txt

@@ -55,7 +55,10 @@ BUGFIXES:
   ZOOKEEPER-490. the java docs for session creation are misleading/incomplete
   (phunt)
 
-  ZOOKEEPER-501. CnxManagerTest failed on hudson. (flavio via mahadev) 
+  ZOOKEEPER-501. CnxManagerTest failed on hudson. (flavio via mahadev)
+  
+  ZOOKEEPER-499. electionAlg should default to FLE (3) - regression (phunt via
+  mahadev) 
 
 IMPROVEMENTS:
   ZOOKEEPER-473. cleanup junit tests to eliminate false positives due to

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

@@ -19,15 +19,13 @@
 package org.apache.zookeeper.server.quorum;
 
 import java.io.IOException;
-import java.net.InetAddress;
 import java.net.InetSocketAddress;
 import java.net.Socket;
-import java.net.UnknownHostException;
 import java.nio.ByteBuffer;
 import java.nio.channels.ServerSocketChannel;
 import java.nio.channels.SocketChannel;
+import java.nio.channels.UnresolvedAddressException;
 import java.util.Enumeration;
-import java.util.Random;
 import java.util.concurrent.ArrayBlockingQueue;
 import java.util.concurrent.ConcurrentHashMap;
 
@@ -316,17 +314,28 @@ public class QuorumCnxManager {
      */
     
     synchronized void connectOne(long sid){
-        
-        if ((senderWorkerMap.get(sid) == null)){ 
+        if (senderWorkerMap.get(sid) == null){ 
+            InetSocketAddress electionAddr =
+                self.quorumPeers.get(sid).electionAddr;
             try {
                 SocketChannel channel;
                 LOG.debug("Opening channel to server "  + sid);
-                channel = SocketChannel
-                        .open(self.quorumPeers.get(sid).electionAddr);
+                channel = SocketChannel.open(electionAddr);
                 channel.socket().setTcpNoDelay(true);
                 initiateConnection(channel, sid);
+            } catch (UnresolvedAddressException e) {
+                // Sun doesn't include the address that causes this
+                // exception to be thrown, also UAE cannot be wrapped cleanly
+                // so we log the exception in order to capture this critical
+                // detail.
+                LOG.warn("Cannot open channel to " + sid
+                        + " at election address " + electionAddr,
+                        e);
+                throw e;
             } catch (IOException e) {
-                LOG.warn("Cannot open channel to " + sid, e);
+                LOG.warn("Cannot open channel to " + sid
+                        + " at election address " + electionAddr,
+                        e);
             }
         } else {
             LOG.error("There is a connection for server " + sid);

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

@@ -47,8 +47,8 @@ public class QuorumPeerConfig {
 
     protected int initLimit;
     protected int syncLimit;
-    protected int electionAlg;
-    protected int electionPort;
+    protected int electionAlg = 3;
+    protected int electionPort = 2182;
     protected int maxClientCnxns = 10;
     protected final HashMap<Long,QuorumServer> servers =
         new HashMap<Long, QuorumServer>();

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

@@ -35,6 +35,7 @@ import org.apache.log4j.Level;
 import org.apache.log4j.Logger;
 import org.apache.log4j.WriterAppender;
 import org.apache.zookeeper.CreateMode;
+import org.apache.zookeeper.PortAssignment;
 import org.apache.zookeeper.WatchedEvent;
 import org.apache.zookeeper.Watcher;
 import org.apache.zookeeper.ZooKeeper;
@@ -118,14 +119,14 @@ public class QuorumPeerMainTest extends TestCase implements Watcher {
         LOG.info("STARTING " + getName());
         ClientBase.setupTestEnv();
 
-        final int CLIENT_PORT_QP1 = 3181;
-        final int CLIENT_PORT_QP2 = CLIENT_PORT_QP1 + 3;
+        final int CLIENT_PORT_QP1 = PortAssignment.unique();
+        final int CLIENT_PORT_QP2 = PortAssignment.unique();
 
         String quorumCfgSection =
-            "server.1=localhost:" + (CLIENT_PORT_QP1 + 1)
-            + ":" + (CLIENT_PORT_QP1 + 2)
-            + "\nserver.2=localhost:" + (CLIENT_PORT_QP2 + 1)
-            + ":" + (CLIENT_PORT_QP2 + 2);
+            "server.1=localhost:" + PortAssignment.unique()
+            + ":" + PortAssignment.unique()
+            + "\nserver.2=localhost:" + PortAssignment.unique()
+            + ":" + PortAssignment.unique();
 
         MainThread q1 = new MainThread(1, CLIENT_PORT_QP1, quorumCfgSection);
         MainThread q2 = new MainThread(2, CLIENT_PORT_QP2, quorumCfgSection);
@@ -181,18 +182,18 @@ public class QuorumPeerMainTest extends TestCase implements Watcher {
         ByteArrayOutputStream os = new ByteArrayOutputStream();
         WriterAppender appender = new WriterAppender(layout, os);
         appender.setThreshold(Level.WARN);
-        Logger.getLogger(org.apache.zookeeper.server.quorum.QuorumPeer.class)
-            .addAppender(appender);
+        Logger qlogger = Logger.getLogger("org.apache.zookeeper.server.quorum");
+        qlogger.addAppender(appender);
 
         try {
-            final int CLIENT_PORT_QP1 = 3181;
-            final int CLIENT_PORT_QP2 = CLIENT_PORT_QP1 + 3;
+            final int CLIENT_PORT_QP1 = PortAssignment.unique();
+            final int CLIENT_PORT_QP2 = PortAssignment.unique();
 
             String quorumCfgSection =
-                "server.1=localhost:" + (CLIENT_PORT_QP1 + 1)
-                + ":" + (CLIENT_PORT_QP1 + 2)
-                + "\nserver.2=fee.fii.foo.fum:" + (CLIENT_PORT_QP2 + 1)
-                + ":" + (CLIENT_PORT_QP2 + 2);
+                "server.1=localhost:" + PortAssignment.unique()
+                + ":" + PortAssignment.unique()
+                + "\nserver.2=fee.fii.foo.fum:" + PortAssignment.unique()
+                + ":" + PortAssignment.unique();
 
             MainThread q1 = new MainThread(1, CLIENT_PORT_QP1, quorumCfgSection);
             q1.start();
@@ -210,15 +211,14 @@ public class QuorumPeerMainTest extends TestCase implements Watcher {
                             ClientBase.CONNECTION_TIMEOUT));
 
         } finally {
-            Logger.getLogger(org.apache.zookeeper.server.quorum.QuorumPeer.class)
-                .removeAppender(appender);
+            qlogger.removeAppender(appender);
         }
 
         LineNumberReader r = new LineNumberReader(new StringReader(os.toString()));
         String line;
         boolean found = false;
         Pattern p =
-            Pattern.compile(".*IllegalArgumentException.*fee.fii.foo.fum.*");
+            Pattern.compile(".*Cannot open channel to .* at election address .*");
         while ((line = r.readLine()) != null) {
             found = p.matcher(line).matches();
             if (found) {
@@ -228,6 +228,75 @@ public class QuorumPeerMainTest extends TestCase implements Watcher {
         assertTrue("complains about host", found);
     }
 
+
+    /**
+     * Verify handling of quorum defaults
+     * * default electionAlg is fast leader election
+     */
+    @Test
+    public void testQuorumDefaults() throws Exception {
+        LOG.info("STARTING " + getName());
+        ClientBase.setupTestEnv();
+
+        // setup the logger to capture all logs
+        Layout layout =
+            Logger.getRootLogger().getAppender("CONSOLE").getLayout();
+        ByteArrayOutputStream os = new ByteArrayOutputStream();
+        WriterAppender appender = new WriterAppender(layout, os);
+        appender.setThreshold(Level.WARN);
+        Logger zlogger = Logger.getLogger("org.apache.zookeeper");
+        zlogger.addAppender(appender);
+
+        try {
+            final int CLIENT_PORT_QP1 = PortAssignment.unique();
+            final int CLIENT_PORT_QP2 = PortAssignment.unique();
+
+            String quorumCfgSection =
+                "server.1=localhost:" + PortAssignment.unique()
+                + ":" + PortAssignment.unique()
+                + "\nserver.2=localhost:" + PortAssignment.unique()
+                + ":" + PortAssignment.unique();
+
+            MainThread q1 = new MainThread(1, CLIENT_PORT_QP1, quorumCfgSection);
+            MainThread q2 = new MainThread(2, CLIENT_PORT_QP2, quorumCfgSection);
+            q1.start();
+            q2.start();
+
+            assertTrue("waiting for server 1 being up",
+                    ClientBase.waitForServerUp("localhost:" + CLIENT_PORT_QP1,
+                            CONNECTION_TIMEOUT));
+            assertTrue("waiting for server 2 being up",
+                    ClientBase.waitForServerUp("localhost:" + CLIENT_PORT_QP2,
+                            CONNECTION_TIMEOUT));
+
+            q1.shutdown();
+            q2.shutdown();
+
+            assertTrue("waiting for server 1 down",
+                    ClientBase.waitForServerDown("localhost:" + CLIENT_PORT_QP1,
+                            ClientBase.CONNECTION_TIMEOUT));
+            assertTrue("waiting for server 2 down",
+                    ClientBase.waitForServerDown("localhost:" + CLIENT_PORT_QP2,
+                            ClientBase.CONNECTION_TIMEOUT));
+
+        } finally {
+            zlogger.removeAppender(appender);
+        }
+
+        LineNumberReader r = new LineNumberReader(new StringReader(os.toString()));
+        String line;
+        boolean found = false;
+        Pattern p =
+            Pattern.compile(".*FastLeaderElection.*");
+        while ((line = r.readLine()) != null) {
+            found = p.matcher(line).matches();
+            if (found) {
+                break;
+            }
+        }
+        assertTrue("fastleaderelection used", found);
+    }
+
     public void process(WatchedEvent event) {
         // ignore for this test
     }