ZOOKEEPER-635. Server supports listening on a specified network address

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

CHANGES.txt

@@ -353,6 +353,8 @@ NEW FEATURES:
 
   ZOOKEEPER-678. Gui browser application to view and edit the contents of a
   zookeeper instance (Colin Goodheart-Smithe via phunt)
+
+  ZOOKEEPER-635. Server supports listening on a specified network address (phunt via breed)
   
 Release 3.2.0 - 2009-06-30
 

docs/zookeeperAdmin.html

@@ -1085,6 +1085,7 @@ server.3=zoo3:2888:3888</span>
               and the configuration variable name is different from the system
               property. Yes - it's not consistent, and it's annoying.)</p>
 </dd>
+
           
 <dt>
 <term>maxClientCnxns</term>
@@ -1098,9 +1099,25 @@ server.3=zoo3:2888:3888</span>
               descriptor exhaustion. The default is 10. Setting this to 0 
               entirely removes the limit on concurrent connections.</p>
 </dd>
+
+           
+<dt>
+<term>clientPortBindAddress</term>
+</dt>
+<dd>
+<p>
+<strong>New in 3.3:</strong> the
+               address (ipv4, ipv6 or hostname) to listen for client
+               connections; that is, the address that clients attempt
+               to connect to. This is optional, by default we bind in
+               such a way that any connection to the <strong>clientPort</strong> for any
+               address/interface/nic on the server will be
+               accepted.</p>
+</dd>
+
         
 </dl>
-<a name="N10342"></a><a name="sc_clusterOptions"></a>
+<a name="N1034E"></a><a name="sc_clusterOptions"></a>
 <h4>Cluster Options</h4>
 <p>The options in this section are designed for use with an ensemble
         of servers -- that is, when deploying clusters of servers.</p>
@@ -1238,7 +1255,7 @@ server.3=zoo3:2888:3888</span>
         
 </dl>
 <p></p>
-<a name="N103C2"></a><a name="sc_authOptions"></a>
+<a name="N103CE"></a><a name="sc_authOptions"></a>
 <h4>Authentication &amp; Authorization Options</h4>
 <p>The options in this section allow control over
         authentication/authorization performed by the service.</p>
@@ -1272,7 +1289,7 @@ server.3=zoo3:2888:3888</span>
 </dd>
         
 </dl>
-<a name="N103E5"></a><a name="Unsafe+Options"></a>
+<a name="N103F1"></a><a name="Unsafe+Options"></a>
 <h4>Unsafe Options</h4>
 <p>The following options can be useful, but be careful when you use
         them. The risk of each is explained along with the explanation of what
@@ -1317,7 +1334,7 @@ server.3=zoo3:2888:3888</span>
 </dd>
         
 </dl>
-<a name="N10417"></a><a name="sc_zkCommands"></a>
+<a name="N10423"></a><a name="sc_zkCommands"></a>
 <h3 class="h4">ZooKeeper Commands: The Four Letter Words</h3>
 <p>ZooKeeper responds to a small set of commands. Each command is
       composed of four letters. You issue the commands to ZooKeeper via telnet
@@ -1438,7 +1455,7 @@ server.3=zoo3:2888:3888</span>
 <pre class="code">$ echo ruok | nc 127.0.0.1 5111
 imok
 </pre>
-<a name="N1047F"></a><a name="sc_dataFileManagement"></a>
+<a name="N1048B"></a><a name="sc_dataFileManagement"></a>
 <h3 class="h4">Data File Management</h3>
 <p>ZooKeeper stores its data in a data directory and its transaction
       log in a transaction log directory. By default these two directories are
@@ -1446,7 +1463,7 @@ imok
       transaction log files in a separate directory than the data files.
       Throughput increases and latency decreases when transaction logs reside
       on a dedicated log devices.</p>
-<a name="N10488"></a><a name="The+Data+Directory"></a>
+<a name="N10494"></a><a name="The+Data+Directory"></a>
 <h4>The Data Directory</h4>
 <p>This directory has two files in it:</p>
 <ul>
@@ -1492,14 +1509,14 @@ imok
         idempotent nature of its updates. By replaying the transaction log
         against fuzzy snapshots ZooKeeper gets the state of the system at the
         end of the log.</p>
-<a name="N104C4"></a><a name="The+Log+Directory"></a>
+<a name="N104D0"></a><a name="The+Log+Directory"></a>
 <h4>The Log Directory</h4>
 <p>The Log Directory contains the ZooKeeper transaction logs.
         Before any update takes place, ZooKeeper ensures that the transaction
         that represents the update is written to non-volatile storage. A new
         log file is started each time a snapshot is begun. The log file's
         suffix is the first zxid written to that log.</p>
-<a name="N104CE"></a><a name="sc_filemanagement"></a>
+<a name="N104DA"></a><a name="sc_filemanagement"></a>
 <h4>File Management</h4>
 <p>The format of snapshot and log files does not change between
         standalone ZooKeeper servers and different configurations of
@@ -1519,7 +1536,7 @@ imok
         this document for more details on setting a retention policy
         and maintenance of ZooKeeper storage.
         </p>
-<a name="N104E3"></a><a name="sc_commonProblems"></a>
+<a name="N104EF"></a><a name="sc_commonProblems"></a>
 <h3 class="h4">Things to Avoid</h3>
 <p>Here are some common problems you can avoid by configuring
       ZooKeeper correctly:</p>
@@ -1573,7 +1590,7 @@ imok
 </dd>
       
 </dl>
-<a name="N10507"></a><a name="sc_bestPractices"></a>
+<a name="N10513"></a><a name="sc_bestPractices"></a>
 <h3 class="h4">Best Practices</h3>
 <p>For best results, take note of the following list of good
       Zookeeper practices:</p>

src/docs/src/documentation/content/xdocs/zookeeperAdmin.xml

@@ -719,6 +719,7 @@ server.3=zoo3:2888:3888</computeroutput></para>
               property. Yes - it's not consistent, and it's annoying.)</para>
             </listitem>
           </varlistentry>
+
           <varlistentry>
             <term>maxClientCnxns</term>
             <listitem>
@@ -732,6 +733,22 @@ server.3=zoo3:2888:3888</computeroutput></para>
               entirely removes the limit on concurrent connections.</para>
             </listitem>
            </varlistentry>
+
+           <varlistentry>
+             <term>clientPortBindAddress</term>
+
+             <listitem>
+               <para><emphasis role="bold">New in 3.3:</emphasis> the
+               address (ipv4, ipv6 or hostname) to listen for client
+               connections; that is, the address that clients attempt
+               to connect to. This is optional, by default we bind in
+               such a way that any connection to the <emphasis
+               role="bold">clientPort</emphasis> for any
+               address/interface/nic on the server will be
+               accepted.</para>
+             </listitem>
+           </varlistentry>
+
         </variablelist>
       </section>
 

src/java/main/org/apache/zookeeper/ClientCnxn.java

@@ -372,10 +372,13 @@ public class ClientCnxn {
         String hostsList[] = hosts.split(",");
         for (String host : hostsList) {
             int port = 2181;
-            String parts[] = host.split(":");
-            if (parts.length > 1) {
-                port = Integer.parseInt(parts[1]);
-                host = parts[0];
+            int pidx = host.lastIndexOf(':');
+            if (pidx >= 0) {
+                // otherwise : is at the end of the string, ignore
+                if (pidx < host.length() - 1) {
+                    port = Integer.parseInt(host.substring(pidx + 1));
+                }
+                host = host.substring(0, pidx);
             }
             InetAddress addrs[] = InetAddress.getAllByName(host);
             for (InetAddress addr : addrs) {

src/java/main/org/apache/zookeeper/server/NIOServerCnxn.java

@@ -123,8 +123,8 @@ public class NIOServerCnxn implements Watcher, ServerCnxn {
          * @param port
          * @throws IOException
          */
-        public Factory(int port) throws IOException {
-            this(port,0);
+        public Factory(InetSocketAddress addr) throws IOException {
+            this(addr, 0);
         }
 
 
@@ -136,14 +136,14 @@ public class NIOServerCnxn implements Watcher, ServerCnxn {
          * @param maxcc - the number of concurrent connections allowed from a single client.
          * @throws IOException
          */
-        public Factory(int port, int maxcc) throws IOException {
-            super("NIOServerCxn.Factory:" + port);
+        public Factory(InetSocketAddress addr, int maxcc) throws IOException {
+            super("NIOServerCxn.Factory:" + addr);
             setDaemon(true);
             maxClientCnxns = maxcc;
             this.ss = ServerSocketChannel.open();
             ss.socket().setReuseAddress(true);
-            LOG.info("binding to port " + port);
-            ss.socket().bind(new InetSocketAddress(port));
+            LOG.info("binding to port " + addr);
+            ss.socket().bind(addr);
             ss.configureBlocking(false);
             ss.register(selector, SelectionKey.OP_ACCEPT);
         }

src/java/main/org/apache/zookeeper/server/ServerConfig.java

@@ -18,6 +18,7 @@
 
 package org.apache.zookeeper.server;
 
+import java.net.InetSocketAddress;
 import java.util.Arrays;
 
 import org.apache.zookeeper.server.quorum.QuorumPeerConfig;
@@ -30,10 +31,10 @@ import org.apache.zookeeper.server.quorum.QuorumPeerConfig.ConfigException;
  *
  */
 public class ServerConfig {
-    protected int clientPort;
+    protected InetSocketAddress clientPortAddress;
     protected String dataDir;
     protected String dataLogDir;
-    protected int tickTime = ZooKeeperServer.DEFAULT_TICK_TIME;    
+    protected int tickTime = ZooKeeperServer.DEFAULT_TICK_TIME;
     protected int maxClientCnxns;
 
     /**
@@ -48,7 +49,7 @@ public class ServerConfig {
                     + Arrays.toString(args));
         }
 
-        clientPort = Integer.parseInt(args[0]);
+        clientPortAddress = new InetSocketAddress(Integer.parseInt(args[0]));
         dataDir = args[1];
         dataLogDir = dataDir;
         if (args.length == 3) {
@@ -79,14 +80,16 @@ public class ServerConfig {
      * @param config
      */
     public void readFrom(QuorumPeerConfig config) {
-      clientPort = config.getClientPort();
+      clientPortAddress = config.getClientPortAddress();
       dataDir = config.getDataDir();
       dataLogDir = config.getDataLogDir();
       tickTime = config.getTickTime();
       maxClientCnxns = config.getMaxClientCnxns();
     }
 
-    public int getClientPort() { return clientPort; }
+    public InetSocketAddress getClientPortAddress() {
+        return clientPortAddress;
+    }
     public String getDataDir() { return dataDir; }
     public String getDataLogDir() { return dataLogDir; }
     public int getTickTime() { return tickTime; }

src/java/main/org/apache/zookeeper/server/ZooKeeperServerMain.java

@@ -103,7 +103,8 @@ public class ZooKeeperServerMain {
                    File(config.dataLogDir), new File(config.dataDir));
             zkServer.setTxnLogFactory(ftxn);
             zkServer.setTickTime(config.tickTime);
-            cnxnFactory = new NIOServerCnxn.Factory(config.clientPort, config.getMaxClientCnxns());
+            cnxnFactory = new NIOServerCnxn.Factory(
+                    config.getClientPortAddress(), config.getMaxClientCnxns());
             cnxnFactory.startup(zkServer);
             cnxnFactory.join();
             if (zkServer.isRunning()) {

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

@@ -33,14 +33,12 @@ import java.util.Map;
 import org.apache.log4j.Logger;
 import org.apache.zookeeper.jmx.MBeanRegistry;
 import org.apache.zookeeper.jmx.ZKMBeanInfo;
-import org.apache.zookeeper.server.DataTree;
 import org.apache.zookeeper.server.NIOServerCnxn;
 import org.apache.zookeeper.server.ZKDatabase;
 import org.apache.zookeeper.server.ZooKeeperServer;
 import org.apache.zookeeper.server.persistence.FileTxnSnapLog;
-import org.apache.zookeeper.server.persistence.Util;
-import org.apache.zookeeper.server.quorum.flexible.QuorumVerifier;
 import org.apache.zookeeper.server.quorum.flexible.QuorumMaj;
+import org.apache.zookeeper.server.quorum.flexible.QuorumVerifier;
 
 /**
  * This class manages the quorum protocol. There are three states this server
@@ -439,7 +437,8 @@ public class QuorumPeer extends Thread implements QuorumStats.Provider {
     {
         this(quorumPeers, snapDir, logDir, electionAlg,
                 myid,tickTime, initLimit,syncLimit,
-                new NIOServerCnxn.Factory(clientPort),
+                new NIOServerCnxn.Factory(
+                        new InetSocketAddress(clientPort)),
                 new QuorumMaj(countParticipants(quorumPeers)));
     }
     
@@ -455,7 +454,8 @@ public class QuorumPeer extends Thread implements QuorumStats.Provider {
     {
         this(quorumPeers, snapDir, logDir, electionAlg,
                 myid,tickTime, initLimit,syncLimit,
-                new NIOServerCnxn.Factory(clientPort), quorumConfig);
+                new NIOServerCnxn.Factory(new InetSocketAddress(clientPort)),
+                    quorumConfig);
     }
     
     /**
@@ -873,7 +873,7 @@ public class QuorumPeer extends Thread implements QuorumStats.Provider {
         return cnxnFactory.getLocalPort();
     }
 
-    public void setClientPort(int clientPort) {
+    public void setClientPortAddress(InetSocketAddress addr) {
     }
  
     public void setTxnFactory(FileTxnSnapLog factory) {

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

@@ -23,6 +23,7 @@ import java.io.File;
 import java.io.FileInputStream;
 import java.io.FileReader;
 import java.io.IOException;
+import java.net.InetAddress;
 import java.net.InetSocketAddress;
 import java.util.Collections;
 import java.util.HashMap;
@@ -41,7 +42,7 @@ import org.apache.zookeeper.server.quorum.flexible.QuorumHierarchical;
 public class QuorumPeerConfig {
     private static final Logger LOG = Logger.getLogger(QuorumPeerConfig.class);
 
-    protected int clientPort;
+    protected InetSocketAddress clientPortAddress;
     protected String dataDir;
     protected String dataLogDir;
     protected int tickTime = ZooKeeperServer.DEFAULT_TICK_TIME;
@@ -61,7 +62,7 @@ public class QuorumPeerConfig {
     protected HashMap<Long, Long> serverGroup = new HashMap<Long, Long>();
     protected int numGroups = 0;
     protected QuorumVerifier quorumVerifier;
-    
+
     protected LearnerType peerType = LearnerType.PARTICIPANT;
 
     @SuppressWarnings("serial")
@@ -89,7 +90,7 @@ public class QuorumPeerConfig {
                 throw new IllegalArgumentException(configFile.toString()
                         + " file is missing");
             }
-    
+
             Properties cfg = new Properties();
             FileInputStream in = new FileInputStream(configFile);
             try {
@@ -97,7 +98,7 @@ public class QuorumPeerConfig {
             } finally {
                 in.close();
             }
-    
+
             parseProperties(cfg);
         } catch (IOException e) {
             throw new ConfigException("Error processing " + path, e);
@@ -114,6 +115,8 @@ public class QuorumPeerConfig {
      */
     public void parseProperties(Properties zkProp)
     throws IOException, ConfigException {
+        int clientPort = 0;
+        String clientPortAddress = null;
         for (Entry<Object, Object> entry : zkProp.entrySet()) {
             String key = entry.getKey().toString().trim();
             String value = entry.getValue().toString().trim();
@@ -123,6 +126,8 @@ public class QuorumPeerConfig {
                 dataLogDir = value;
             } else if (key.equals("clientPort")) {
                 clientPort = Integer.parseInt(value);
+            } else if (key.equals("clientPortAddress")) {
+                clientPortAddress = value.trim();
             } else if (key.equals("tickTime")) {
                 tickTime = Integer.parseInt(value);
             } else if (key.equals("initLimit")) {
@@ -140,7 +145,7 @@ public class QuorumPeerConfig {
                     peerType = LearnerType.PARTICIPANT;
                 } else
                 {
-                    throw new ConfigException("Unrecognised peertype: " + value);                      
+                    throw new ConfigException("Unrecognised peertype: " + value);
                 }
             } else if (key.startsWith("server.")) {
                 int dot = key.indexOf('.');
@@ -174,14 +179,14 @@ public class QuorumPeerConfig {
                                 electionAddr,type));
                     } else {
                         throw new ConfigException("Unrecognised peertype: " + value);
-                    }                    
+                    }
                 }
             } else if (key.startsWith("group")) {
                 int dot = key.indexOf('.');
                 long gid = Long.parseLong(key.substring(dot + 1));
-                
+
                 numGroups++;
-                
+
                 String parts[] = value.split(":");
                 for(String s : parts){
                     long sid = Long.parseLong(s);
@@ -189,8 +194,8 @@ public class QuorumPeerConfig {
                         throw new ConfigException("Server " + sid + "is in multiple groups");
                     else
                         serverGroup.put(sid, gid);
-                }       
-                
+                }
+
             } else if(key.startsWith("weight")) {
                 int dot = key.indexOf('.');
                 long sid = Long.parseLong(key.substring(dot + 1));
@@ -199,7 +204,7 @@ public class QuorumPeerConfig {
                 System.setProperty("zookeeper." + key, value);
             }
         }
-       
+
         if (dataDir == null) {
             throw new IllegalArgumentException("dataDir is not set");
         }
@@ -214,6 +219,13 @@ public class QuorumPeerConfig {
         if (clientPort == 0) {
             throw new IllegalArgumentException("clientPort is not set");
         }
+        if (clientPortAddress != null) {
+            this.clientPortAddress = new InetSocketAddress(
+                    InetAddress.getByName(clientPortAddress), clientPort);
+        } else {
+            this.clientPortAddress = new InetSocketAddress(clientPort);
+        }
+
         if (tickTime == 0) {
             throw new IllegalArgumentException("tickTime is not set");
         }
@@ -239,35 +251,35 @@ public class QuorumPeerConfig {
             /*
              * Default of quorum config is majority
              */
-            if(serverGroup.size() > 0){  
+            if(serverGroup.size() > 0){
                 if(servers.size() != serverGroup.size())
                     throw new ConfigException("Every server must be in exactly one group");
-            	/*
-            	 * The deafult weight of a server is 1
-            	 */
-            	for(QuorumServer s : servers.values()){
-            		if(!serverWeight.containsKey(s.id))
-            			serverWeight.put(s.id, (long) 1);
-            	}
-            	
-            	/*
-            	 * Set the quorumVerifier to be QuorumHierarchical 
-            	 */
-                quorumVerifier = new QuorumHierarchical(numGroups, 
+                /*
+                 * The deafult weight of a server is 1
+                 */
+                for(QuorumServer s : servers.values()){
+                    if(!serverWeight.containsKey(s.id))
+                        serverWeight.put(s.id, (long) 1);
+                }
+
+                /*
+                 * Set the quorumVerifier to be QuorumHierarchical
+                 */
+                quorumVerifier = new QuorumHierarchical(numGroups,
                         serverWeight, serverGroup);
             } else {
-            	/*
-            	 * The default QuorumVerifier is QuorumMaj
-            	 */
-            	
+                /*
+                 * The default QuorumVerifier is QuorumMaj
+                 */
+
                 LOG.info("Defaulting to majority quorums");
                 quorumVerifier = new QuorumMaj(servers.size());
             }
-            
-            // Now add observers to servers, once the quorums have been 
+
+            // Now add observers to servers, once the quorums have been
             // figured out
             servers.putAll(observers);
-            
+
             File myIdFile = new File(dataDir, "myid");
             if (!myIdFile.exists()) {
                 throw new IllegalArgumentException(myIdFile.toString()
@@ -289,7 +301,7 @@ public class QuorumPeerConfig {
         }
     }
 
-    public int getClientPort() { return clientPort; }
+    public InetSocketAddress getClientPortAddress() { return clientPortAddress; }
     public String getDataDir() { return dataDir; }
     public String getDataLogDir() { return dataLogDir; }
     public int getTickTime() { return tickTime; }
@@ -297,13 +309,13 @@ public class QuorumPeerConfig {
     public int getInitLimit() { return initLimit; }
     public int getSyncLimit() { return syncLimit; }
     public int getElectionAlg() { return electionAlg; }
-    public int getElectionPort() { return electionPort; }    
+    public int getElectionPort() { return electionPort; }
     public int getMaxClientCnxns() { return maxClientCnxns; }
-    
-    public QuorumVerifier getQuorumVerifier() {   
+
+    public QuorumVerifier getQuorumVerifier() {
         return quorumVerifier;
     }
-    
+
     public Map<Long,QuorumServer> getServers() {
         return Collections.unmodifiableMap(servers);
     }

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

@@ -119,10 +119,11 @@ public class QuorumPeerMain {
       LOG.info("Starting quorum peer");
       try {
           NIOServerCnxn.Factory cnxnFactory =
-              new NIOServerCnxn.Factory(config.getClientPort(), config.getMaxClientCnxns());
+              new NIOServerCnxn.Factory(config.getClientPortAddress(),
+                      config.getMaxClientCnxns());
   
           quorumPeer = new QuorumPeer();
-          quorumPeer.setClientPort(config.getClientPort());
+          quorumPeer.setClientPortAddress(config.getClientPortAddress());
           quorumPeer.setTxnFactory(new FileTxnSnapLog(
                       new File(config.getDataLogDir()),
                       new File(config.getDataDir())));

src/java/test/org/apache/zookeeper/server/CRCTest.java

@@ -26,6 +26,7 @@ import java.io.FileInputStream;
 import java.io.IOException;
 import java.io.InputStream;
 import java.io.RandomAccessFile;
+import java.net.InetSocketAddress;
 import java.util.List;
 import java.util.Map;
 import java.util.concurrent.ConcurrentHashMap;
@@ -124,7 +125,8 @@ public class CRCTest extends TestCase implements Watcher{
         ZooKeeperServer zks = new ZooKeeperServer(tmpDir, tmpDir, 3000);
         SyncRequestProcessor.setSnapCount(150);
         final int PORT = Integer.parseInt(HOSTPORT.split(":")[1]);
-        NIOServerCnxn.Factory f = new NIOServerCnxn.Factory(PORT);
+        NIOServerCnxn.Factory f = new NIOServerCnxn.Factory(
+                new InetSocketAddress(PORT));
         f.startup(zks);
         LOG.info("starting up the zookeeper server .. waiting");
         assertTrue("waiting for server being up",

src/java/test/org/apache/zookeeper/server/InvalidSnapshotTest.java

@@ -20,6 +20,7 @@ package org.apache.zookeeper.server;
 
 import java.io.File;
 import java.io.RandomAccessFile;
+import java.net.InetSocketAddress;
 
 import junit.framework.TestCase;
 
@@ -59,7 +60,8 @@ public class InvalidSnapshotTest extends TestCase implements Watcher {
        ZooKeeperServer zks = new ZooKeeperServer(tmpDir, tmpDir, 3000);
        SyncRequestProcessor.setSnapCount(100);
        final int PORT = Integer.parseInt(HOSTPORT.split(":")[1]);
-       NIOServerCnxn.Factory f = new NIOServerCnxn.Factory(PORT);
+       NIOServerCnxn.Factory f = new NIOServerCnxn.Factory(
+               new InetSocketAddress(PORT));
        f.startup(zks);
        assertTrue("waiting for server being up ",
                ClientBase.waitForServerUp(HOSTPORT,CONNECTION_TIMEOUT));
@@ -83,7 +85,7 @@ public class InvalidSnapshotTest extends TestCase implements Watcher {
        // now restart the server and see if it starts
        zks = new ZooKeeperServer(tmpDir, tmpDir, 3000);
        SyncRequestProcessor.setSnapCount(100);
-       f = new NIOServerCnxn.Factory(PORT);
+       f = new NIOServerCnxn.Factory(new InetSocketAddress(PORT));
        f.startup(zks);
        assertTrue("waiting for server being up ",
                ClientBase.waitForServerUp(HOSTPORT,CONNECTION_TIMEOUT));

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

@@ -21,6 +21,7 @@ package org.apache.zookeeper.test;
 import static org.apache.zookeeper.test.ClientBase.CONNECTION_TIMEOUT;
 
 import java.io.File;
+import java.net.InetSocketAddress;
 import java.util.ArrayList;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.TimeUnit;
@@ -62,7 +63,8 @@ public class ACLTest extends TestCase implements Watcher {
         ZooKeeperServer zks = new ZooKeeperServer(tmpDir, tmpDir, 3000);
         SyncRequestProcessor.setSnapCount(1000);
         final int PORT = Integer.parseInt(HOSTPORT.split(":")[1]);
-        NIOServerCnxn.Factory f = new NIOServerCnxn.Factory(PORT);
+        NIOServerCnxn.Factory f = new NIOServerCnxn.Factory(
+                new InetSocketAddress(PORT));
         f.startup(zks);
         LOG.info("starting up the zookeeper server .. waiting");
         assertTrue("waiting for server being up",
@@ -93,7 +95,8 @@ public class ACLTest extends TestCase implements Watcher {
         ZooKeeperServer zks = new ZooKeeperServer(tmpDir, tmpDir, 3000);
         SyncRequestProcessor.setSnapCount(1000);
         final int PORT = Integer.parseInt(HOSTPORT.split(":")[1]);
-        NIOServerCnxn.Factory f = new NIOServerCnxn.Factory(PORT);
+        NIOServerCnxn.Factory f = new NIOServerCnxn.Factory(
+                new InetSocketAddress(PORT));
         f.startup(zks);
         LOG.info("starting up the zookeeper server .. waiting");
         assertTrue("waiting for server being up",
@@ -127,7 +130,7 @@ public class ACLTest extends TestCase implements Watcher {
         startSignal = new CountDownLatch(1);
 
         zks = new ZooKeeperServer(tmpDir, tmpDir, 3000);
-        f = new NIOServerCnxn.Factory(PORT);
+        f = new NIOServerCnxn.Factory(new InetSocketAddress(PORT));
 
         f.startup(zks);
 

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

@@ -25,9 +25,12 @@ import java.io.InputStreamReader;
 import java.io.OutputStream;
 import java.lang.management.ManagementFactory;
 import java.lang.management.OperatingSystemMXBean;
+import java.net.InetSocketAddress;
 import java.net.Socket;
+import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.LinkedList;
+import java.util.List;
 import java.util.Map;
 import java.util.Map.Entry;
 import java.util.concurrent.CountDownLatch;
@@ -178,18 +181,42 @@ public abstract class ClientBase extends TestCase {
         return zk;
     }
 
-    public static String send4LetterWord(String hp, String cmd)
-        throws IOException
-    {
-        String split[] = hp.split(":");
-        String host = split[0];
+    public static class HostPort {
+        String host;
         int port;
-        try {
-            port = Integer.parseInt(split[1]);
-        } catch(RuntimeException e) {
-            throw new RuntimeException("Problem parsing " + hp + e.toString());
+        public HostPort(String host, int port) {
+            this.host = host;
+            this.port = port;
+        }
+    }
+    public static List<HostPort> parseHostPortList(String hplist) {
+        ArrayList<HostPort> alist = new ArrayList<HostPort>();
+        for (String hp: hplist.split(",")) {
+            int idx = hp.lastIndexOf(':');
+            String host = hp.substring(0, idx);
+            int port;
+            try {
+                port = Integer.parseInt(hp.substring(idx + 1));
+            } catch(RuntimeException e) {
+                throw new RuntimeException("Problem parsing " + hp + e.toString());
+            }
+            alist.add(new HostPort(host,port));
         }
+        return alist;
+    }
 
+    /**
+     * Send the 4letterword
+     * @param host the destination host
+     * @param port the destination port
+     * @param cmd the 4letterword
+     * @return
+     * @throws IOException
+     */
+    public static String send4LetterWord(String host, int port, String cmd)
+        throws IOException
+    {
+        LOG.info("connecting to " + host + " " + port);
         Socket sock = new Socket(host, port);
         BufferedReader reader = null;
         try {
@@ -219,8 +246,8 @@ public abstract class ClientBase extends TestCase {
         while (true) {
             try {
                 // if there are multiple hostports, just take the first one
-                hp = hp.split(",")[0];
-                String result = send4LetterWord(hp, "stat");
+                HostPort hpobj = parseHostPortList(hp).get(0);
+                String result = send4LetterWord(hpobj.host, hpobj.port, "stat");
                 if (result.startsWith("Zookeeper version:")) {
                     return true;
                 }
@@ -244,7 +271,8 @@ public abstract class ClientBase extends TestCase {
         long start = System.currentTimeMillis();
         while (true) {
             try {
-                send4LetterWord(hp, "stat");
+                HostPort hpobj = parseHostPortList(hp).get(0);
+                send4LetterWord(hpobj.host, hpobj.port, "stat");
             } catch (IOException e) {
                 return true;
             }
@@ -288,7 +316,8 @@ public abstract class ClientBase extends TestCase {
         return tmpDir;
     }
     private static int getPort(String hostPort) {
-        String portstr = hostPort.split(":")[1];
+        String[] split = hostPort.split(":");
+        String portstr = split[split.length-1];
         String[] pc = portstr.split("/");
         if (pc.length > 1) {
             portstr = pc[0];
@@ -303,10 +332,9 @@ public abstract class ClientBase extends TestCase {
         ZooKeeperServer zks = new ZooKeeperServer(dataDir, dataDir, 3000);
         final int PORT = getPort(hostPort);
         if (factory == null) {
-            factory = new NIOServerCnxn.Factory(PORT,maxCnxns);
+            factory = new NIOServerCnxn.Factory(new InetSocketAddress(PORT),maxCnxns);
         }
         factory.startup(zks);
-
         assertTrue("waiting for server up",
                    ClientBase.waitForServerUp("127.0.0.1:" + PORT,
                                               CONNECTION_TIMEOUT));

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

@@ -0,0 +1,127 @@
+/**
+ * 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 static org.apache.zookeeper.test.ClientBase.CONNECTION_TIMEOUT;
+
+import java.io.File;
+import java.net.InetAddress;
+import java.net.InetSocketAddress;
+import java.net.NetworkInterface;
+import java.util.Enumeration;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+
+import junit.framework.TestCase;
+
+import org.apache.log4j.Logger;
+import org.apache.zookeeper.PortAssignment;
+import org.apache.zookeeper.WatchedEvent;
+import org.apache.zookeeper.Watcher;
+import org.apache.zookeeper.ZooKeeper;
+import org.apache.zookeeper.Watcher.Event.KeeperState;
+import org.apache.zookeeper.server.NIOServerCnxn;
+import org.apache.zookeeper.server.ZooKeeperServer;
+import org.junit.Test;
+
+public class ClientPortBindTest extends TestCase implements Watcher {
+    protected static final Logger LOG = 
+        Logger.getLogger(ClientPortBindTest.class);
+
+    private volatile CountDownLatch startSignal;
+
+    @Override
+    protected void setUp() throws Exception {
+        LOG.info("STARTING " + getName());
+    }
+    @Override
+    protected void tearDown() throws Exception {
+        LOG.info("FINISHED " + getName());
+    }
+
+    @Test
+    /**
+     * Verify that the server binds to the specified address
+     */
+    public void testBindByAddress() throws Exception {
+        String bindAddress = null;
+        Enumeration<NetworkInterface> intfs =
+            NetworkInterface.getNetworkInterfaces();
+        // if we have a loopback and it has an address use it
+        while(intfs.hasMoreElements()) {
+            NetworkInterface i = intfs.nextElement();
+            if (i.isLoopback()) {
+                Enumeration<InetAddress> addrs = i.getInetAddresses();
+                if (addrs.hasMoreElements()) {
+                    bindAddress = addrs.nextElement().getHostAddress();
+                    // handle the ipv6 scope_id - ie remove it
+                    bindAddress = bindAddress.split("%")[0];
+                }
+            }
+        }
+        if (bindAddress == null) {
+            LOG.warn("Unable to determine loop back address, skipping test");
+            return;
+        }
+        final int PORT = PortAssignment.unique();
+
+        LOG.info("Using " + bindAddress + " as the bind address");
+        final String HOSTPORT = bindAddress + ":" + PORT;
+        LOG.info("Using " + HOSTPORT + " as the host/port");
+
+
+        File tmpDir = ClientBase.createTmpDir();
+
+        ClientBase.setupTestEnv();
+        ZooKeeperServer zks = new ZooKeeperServer(tmpDir, tmpDir, 3000);
+
+        NIOServerCnxn.Factory f = new NIOServerCnxn.Factory(
+                new InetSocketAddress(bindAddress, PORT));
+        f.startup(zks);
+        LOG.info("starting up the the server, waiting");
+
+        assertTrue("waiting for server up",
+                   ClientBase.waitForServerUp(HOSTPORT,
+                                   CONNECTION_TIMEOUT));
+
+        startSignal = new CountDownLatch(1);
+        ZooKeeper zk = new ZooKeeper(HOSTPORT, CONNECTION_TIMEOUT, this);
+        try {
+            startSignal.await(CONNECTION_TIMEOUT,
+                    TimeUnit.MILLISECONDS);
+            assertTrue("count == 0", startSignal.getCount() == 0);
+            zk.close();
+        } finally {
+            f.shutdown();
+
+            assertTrue("waiting for server down",
+                       ClientBase.waitForServerDown(HOSTPORT,
+                                                    CONNECTION_TIMEOUT));
+        }
+    }
+
+    public void process(WatchedEvent event) {
+        LOG.info("Event:" + event.getState() + " " + event.getType() + " " + event.getPath());
+        if (event.getState() == KeeperState.SyncConnected
+                && startSignal != null && startSignal.getCount() > 0)
+        {
+            startSignal.countDown();
+        }
+    }
+}

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

@@ -22,6 +22,7 @@ import java.io.IOException;
 
 import org.apache.log4j.Logger;
 import org.apache.zookeeper.TestableZooKeeper;
+import org.apache.zookeeper.test.ClientBase.HostPort;
 import org.junit.Test;
 
 public class FourLetterWordsQuorumTest extends QuorumBase {
@@ -93,13 +94,15 @@ public class FourLetterWordsQuorumTest extends QuorumBase {
     private void verify(String hp, String cmd, String expected)
         throws IOException
     {
-        String resp = send4LetterWord(hp, cmd);
-        LOG.info("cmd " + cmd + " expected " + expected + " got " + resp);
-        if (cmd.equals("dump")) {
-            assertTrue(resp.contains(expected)
-                    || resp.contains("Sessions with Ephemerals"));
-        } else {
-            assertTrue(resp.contains(expected));
+        for(HostPort hpobj: parseHostPortList(hp)) {
+            String resp = send4LetterWord(hpobj.host, hpobj.port, cmd);
+            LOG.info("cmd " + cmd + " expected " + expected + " got " + resp);
+            if (cmd.equals("dump")) {
+                assertTrue(resp.contains(expected)
+                        || resp.contains("Sessions with Ephemerals"));
+            } else {
+                assertTrue(resp.contains(expected));
+            }
         }
     }
 }

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

@@ -88,7 +88,8 @@ public class FourLetterWordsTest extends ClientBase {
     }
 
     private void verify(String cmd, String expected) throws IOException {
-        String resp = send4LetterWord(hostPort, cmd);
+        HostPort hpobj = parseHostPortList(hostPort).get(0);
+        String resp = send4LetterWord(hpobj.host, hpobj.port, cmd);
         LOG.info("cmd " + cmd + " expected " + expected + " got " + resp);
         assertTrue(resp.contains(expected));
     }

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

@@ -21,6 +21,7 @@ package org.apache.zookeeper.test;
 import static org.apache.zookeeper.test.ClientBase.CONNECTION_TIMEOUT;
 
 import java.io.File;
+import java.net.InetSocketAddress;
 import java.util.concurrent.CountDownLatch;
 
 import junit.framework.TestCase;
@@ -76,7 +77,8 @@ public class InvalidSnapshotTest extends TestCase implements Watcher {
         ZooKeeperServer zks = new ZooKeeperServer(snapDir, snapDir, 3000);
         SyncRequestProcessor.setSnapCount(1000);
         final int PORT = Integer.parseInt(HOSTPORT.split(":")[1]);
-        NIOServerCnxn.Factory f = new NIOServerCnxn.Factory(PORT);
+        NIOServerCnxn.Factory f = new NIOServerCnxn.Factory(
+                new InetSocketAddress(PORT));
         f.startup(zks);
         LOG.info("starting up the zookeeper server .. waiting");
         assertTrue("waiting for server being up",

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

@@ -22,6 +22,7 @@ import static org.apache.zookeeper.test.ClientBase.CONNECTION_TIMEOUT;
 
 import java.io.File;
 import java.io.IOException;
+import java.net.InetSocketAddress;
 import java.util.ArrayList;
 
 import junit.framework.TestCase;
@@ -63,7 +64,8 @@ public class OOMTest extends TestCase implements Watcher {
         ZooKeeperServer zks = new ZooKeeperServer(tmpDir, tmpDir, 3000);
 
         final int PORT = PortAssignment.unique();
-        NIOServerCnxn.Factory f = new NIOServerCnxn.Factory(PORT);
+        NIOServerCnxn.Factory f = new NIOServerCnxn.Factory(
+                new InetSocketAddress(PORT));
         f.startup(zks);
         assertTrue("waiting for server up",
                    ClientBase.waitForServerUp("127.0.0.1:" + PORT,

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

@@ -19,6 +19,7 @@
 package org.apache.zookeeper.test;
 
 import java.io.File;
+import java.net.InetSocketAddress;
 import java.util.List;
 
 import junit.framework.TestCase;
@@ -53,7 +54,8 @@ public class PurgeTxnTest extends TestCase implements  Watcher {
         ZooKeeperServer zks = new ZooKeeperServer(tmpDir, tmpDir, 3000);
         SyncRequestProcessor.setSnapCount(100);
         final int PORT = Integer.parseInt(HOSTPORT.split(":")[1]);
-        NIOServerCnxn.Factory f = new NIOServerCnxn.Factory(PORT);
+        NIOServerCnxn.Factory f = new NIOServerCnxn.Factory(
+                new InetSocketAddress(PORT));
         f.startup(zks);
         assertTrue("waiting for server being up ",
                 ClientBase.waitForServerUp(HOSTPORT,CONNECTION_TIMEOUT));

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

@@ -80,11 +80,11 @@ public class QuorumBase extends ClientBase {
         portLE4 = PortAssignment.unique();
         portLE5 = PortAssignment.unique();
         
-        hostPort = "127.0.0.1:" + port1 + ":" + portLE1
-            + ",127.0.0.1:" + port2 + ":" + portLE2
-            + ",127.0.0.1:" + port3 + ":" + portLE3
-            + ",127.0.0.1:" + port4 + ":" + portLE4
-            + ",127.0.0.1:" + port5 + ":" + portLE5;
+        hostPort = "127.0.0.1:" + port1
+            + ",127.0.0.1:" + port2
+            + ",127.0.0.1:" + port3
+            + ",127.0.0.1:" + port4
+            + ",127.0.0.1:" + port5;
         LOG.info("Ports are: " + hostPort);
 
         s1dir = ClientBase.createTmpDir();

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

@@ -21,6 +21,7 @@ package org.apache.zookeeper.test;
 import static org.apache.zookeeper.test.ClientBase.CONNECTION_TIMEOUT;
 
 import java.io.File;
+import java.net.InetSocketAddress;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.TimeUnit;
 
@@ -80,7 +81,8 @@ public class RecoveryTest extends TestCase implements Watcher {
         SyncRequestProcessor.setSnapCount(1000);
         try {
             final int PORT = Integer.parseInt(HOSTPORT.split(":")[1]);
-            NIOServerCnxn.Factory f = new NIOServerCnxn.Factory(PORT);
+            NIOServerCnxn.Factory f = new NIOServerCnxn.Factory(
+                    new InetSocketAddress(PORT));
             f.startup(zks);
             LOG.info("starting up the the server, waiting");
 
@@ -118,7 +120,7 @@ public class RecoveryTest extends TestCase implements Watcher {
                                           CONNECTION_TIMEOUT));
 
             zks = new ZooKeeperServer(tmpDir, tmpDir, 3000);
-            f = new NIOServerCnxn.Factory(PORT);
+            f = new NIOServerCnxn.Factory(new InetSocketAddress(PORT));
 
             startSignal = new CountDownLatch(1);
 
@@ -156,7 +158,7 @@ public class RecoveryTest extends TestCase implements Watcher {
                                           ClientBase.CONNECTION_TIMEOUT));
 
             zks = new ZooKeeperServer(tmpDir, tmpDir, 3000);
-            f = new NIOServerCnxn.Factory(PORT);
+            f = new NIOServerCnxn.Factory(new InetSocketAddress(PORT));
 
             startSignal = new CountDownLatch(1);
 

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

@@ -18,6 +18,8 @@
 
 package org.apache.zookeeper.test;
 
+import java.net.InetSocketAddress;
+
 import junit.framework.TestCase;
 
 import org.apache.zookeeper.CreateMode;
@@ -56,7 +58,8 @@ public class RepeatStartupTest extends TestCase {
         ZooKeeperServer zks = new ZooKeeperServer(qb.s1.getTxnFactory().getSnapDir(),
                 qb.s1.getTxnFactory().getDataDir(), 3000);
         final int PORT = Integer.parseInt(hp.split(":")[1]);
-        NIOServerCnxn.Factory factory = new NIOServerCnxn.Factory(PORT);
+        NIOServerCnxn.Factory factory = new NIOServerCnxn.Factory(
+                new InetSocketAddress(PORT));
 
         factory.startup(zks);
         System.out.println("Comment: starting factory");

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

@@ -22,6 +22,7 @@ import static org.apache.zookeeper.test.ClientBase.CONNECTION_TIMEOUT;
 
 import java.io.File;
 import java.io.IOException;
+import java.net.InetSocketAddress;
 import java.util.LinkedList;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.TimeUnit;
@@ -68,7 +69,7 @@ public class SessionTest extends TestCase implements Watcher {
         ZooKeeperServer zs = new ZooKeeperServer(tmpDir, tmpDir, TICK_TIME);
 
         final int PORT = Integer.parseInt(HOSTPORT.split(":")[1]);
-        serverFactory = new NIOServerCnxn.Factory(PORT);
+        serverFactory = new NIOServerCnxn.Factory(new InetSocketAddress(PORT));
         serverFactory.startup(zs);
 
         assertTrue("waiting for server up",

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

@@ -21,6 +21,7 @@ package org.apache.zookeeper.test;
 import static org.apache.zookeeper.test.ClientBase.CONNECTION_TIMEOUT;
 
 import java.io.File;
+import java.net.InetSocketAddress;
 import java.util.Collections;
 import java.util.List;
 import java.util.concurrent.CountDownLatch;
@@ -69,7 +70,8 @@ public class UpgradeTest extends TestCase implements Watcher {
         ZooKeeperServer zks = new ZooKeeperServer(upgradeDir, upgradeDir, 3000);
         SyncRequestProcessor.setSnapCount(1000);
         final int PORT = Integer.parseInt(HOSTPORT.split(":")[1]);
-        NIOServerCnxn.Factory f = new NIOServerCnxn.Factory(PORT);
+        NIOServerCnxn.Factory f = new NIOServerCnxn.Factory(
+                new InetSocketAddress(PORT));
         f.startup(zks);
         LOG.info("starting up the zookeeper server .. waiting");
         assertTrue("waiting for server being up",