ZOOKEEPER-547. Sanity check in QuorumCnxn Manager and quorum communication port.

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

CHANGES.txt

@@ -112,6 +112,9 @@ BUGFIXES:
   ZOOKEEPER-567. javadoc for getchildren2 needs to mention "new in 3.3.0"
   (phunt via breed)
 
+  ZOOKEEPER-547. Sanity check in QuorumCnxn Manager and quorum communication port.
+  (mahadev via breed)
+
 IMPROVEMENTS:
   ZOOKEEPER-473. cleanup junit tests to eliminate false positives due to
   "socket reuse" and failure to close client (phunt via mahadev)

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

@@ -60,7 +60,7 @@ public class QuorumCnxManager {
      */
 
     static final int CAPACITY = 100;
-
+    static final int PACKETMAXSIZE = 1024 * 1024; 
     /*
      * Maximum number of attempts to connect to a peer
      */
@@ -615,11 +615,13 @@ public class QuorumCnxManager {
                     }
                     msgLength.position(0);
                     int length = msgLength.getInt();
-
                     /**
                      * Allocates a new ByteBuffer to receive the message
                      */
                     if (length > 0) {
+                        if (length > PACKETMAXSIZE) {
+                            throw new IOException("Invalid packet of length " + length);
+                        }
                         byte[] msgArray = new byte[length];
                         ByteBuffer message = ByteBuffer.wrap(msgArray);
                         int numbytes = 0;

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

@@ -22,10 +22,16 @@ import static org.apache.zookeeper.test.ClientBase.CONNECTION_TIMEOUT;
 
 import java.io.ByteArrayOutputStream;
 import java.io.File;
+import java.io.FileInputStream;
+import java.io.FileOutputStream;
 import java.io.FileWriter;
 import java.io.IOException;
+import java.io.InputStreamReader;
 import java.io.LineNumberReader;
 import java.io.StringReader;
+import java.net.InetSocketAddress;
+import java.nio.ByteBuffer;
+import java.nio.channels.SocketChannel;
 import java.util.regex.Pattern;
 
 import junit.framework.TestCase;
@@ -41,6 +47,7 @@ import org.apache.zookeeper.Watcher;
 import org.apache.zookeeper.ZooKeeper;
 import org.apache.zookeeper.ZooDefs.Ids;
 import org.apache.zookeeper.test.ClientBase;
+import org.apache.zookeeper.test.ClientTest;
 import org.junit.Test;
 
 
@@ -228,6 +235,60 @@ public class QuorumPeerMainTest extends TestCase implements Watcher {
         assertTrue("complains about host", found);
     }
 
+    /**
+     * verify if bad packets are being handled properly 
+     * at the quorum port
+     * @throws Exception
+     */
+    public void testBadPackets() throws Exception {
+        LOG.info("STARTING " + getName());
+        ClientBase.setupTestEnv();
+        final int CLIENT_PORT_QP1 = PortAssignment.unique();
+        final int CLIENT_PORT_QP2 = PortAssignment.unique();
+        int electionPort1 = PortAssignment.unique();
+        int electionPort2 = PortAssignment.unique();
+        String quorumCfgSection =
+            "server.1=localhost:" + PortAssignment.unique()
+            + ":" + electionPort1
+            + "\nserver.2=localhost:" + PortAssignment.unique()
+            + ":" +  electionPort2;
+        
+        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));
+            
+        byte[] b = new byte[4];
+        int length = 1024*1024*1024;
+        ByteBuffer buff = ByteBuffer.wrap(b);
+        buff.putInt(length);
+        buff.position(0);
+        SocketChannel s = SocketChannel.open(new InetSocketAddress("localhost", electionPort1));
+        s.write(buff);
+        s.close();
+        buff.position(0);
+        s = SocketChannel.open(new InetSocketAddress("localhost", electionPort2));
+        s.write(buff);
+        s.close();
+        
+        ZooKeeper zk = new ZooKeeper("localhost:" + CLIENT_PORT_QP1,
+                ClientBase.CONNECTION_TIMEOUT, this);
+
+        zk.create("/foo_q1", "foobar1".getBytes(), Ids.OPEN_ACL_UNSAFE,
+                CreateMode.PERSISTENT);
+        assertEquals(new String(zk.getData("/foo_q1", null, null)), "foobar1");
+        zk.close();
+        q1.shutdown();
+        q2.shutdown();
+    }
+
 
     /**
      * Verify handling of quorum defaults
@@ -243,7 +304,8 @@ public class QuorumPeerMainTest extends TestCase implements Watcher {
             Logger.getRootLogger().getAppender("CONSOLE").getLayout();
         ByteArrayOutputStream os = new ByteArrayOutputStream();
         WriterAppender appender = new WriterAppender(layout, os);
-        appender.setThreshold(Level.WARN);
+        appender.setImmediateFlush(true);
+        appender.setThreshold(Level.INFO);
         Logger zlogger = Logger.getLogger("org.apache.zookeeper");
         zlogger.addAppender(appender);
 
@@ -282,7 +344,7 @@ public class QuorumPeerMainTest extends TestCase implements Watcher {
         } finally {
             zlogger.removeAppender(appender);
         }
-
+        os.close();
         LineNumberReader r = new LineNumberReader(new StringReader(os.toString()));
         String line;
         boolean found = false;