ZOOKEEPER-974: Allow a configurable ZooKeeper server socket listen ba…

…cklog

Introduces configuration property "clientPortBacklog". Amendment of
original patch from "Hoonmin Kim".

Author: Josh Elser <elserj@apache.org>

Reviewers: andor@apache.org

Closes #784 from joshelser/974-master and squashes the following commits:

fe5f66a5a [Josh Elser] Add some documentation for the new property
be61d1c5f [Josh Elser] Feedback from Andor
3ed36f824 [Josh Elser] Update 'conf' 4lw for clientPortListenBacklog
fd249939c [Josh Elser] ZOOKEEPER-974: Allow a configurable ZooKeeper server socket listen backlog

zookeeper-docs/src/main/resources/markdown/zookeeperAdmin.md

@@ -171,18 +171,18 @@ ensemble:
 1. Install the Java JDK. You can use the native packaging system
   for your system, or download the JDK from:
   [http://java.sun.com/javase/downloads/index.jsp](http://java.sun.com/javase/downloads/index.jsp)
-  
+
 2. Set the Java heap size. This is very important to avoid
   swapping, which will seriously degrade ZooKeeper performance. To
   determine the correct value, use load tests, and make sure you are
   well below the usage limit that would cause you to swap. Be
   conservative - use a maximum heap size of 3GB for a 4GB
   machine.
-  
+
 3. Install the ZooKeeper Server Package. It can be downloaded
   from:
   [http://zookeeper.apache.org/releases.html](http://zookeeper.apache.org/releases.html)
-  
+
 4. Create a configuration file. This file can be called anything.
   Use the following settings as a starting point:
 
@@ -205,16 +205,16 @@ ensemble:
   *myid*, one for each server, which resides in
   that server's data directory, as specified by the configuration file
   parameter **dataDir**.
-  
+
 5. The myid file
   consists of a single line containing only the text of that machine's
   id. So *myid* of server 1 would contain the text
   "1" and nothing else. The id must be unique within the
   ensemble and should have a value between 1 and 255.
   **IMPORTANT:** if you enable extended features such
-   as TTL Nodes (see below) the id must be between 1 
+   as TTL Nodes (see below) the id must be between 1
    and 254 due to internal limitations.
-  
+
 6. Create an initialization marker file *initialize*
   in the same directory as *myid*. This file indicates
   that an empty data directory is expected. When present, an empty data base
@@ -223,13 +223,13 @@ ensemble:
   populate the data directory until it communicates with an active leader.
   Intended use is to only create this file when bringing up a new
   ensemble.
-  
+
 7. If your configuration file is set up, you can start a
   ZooKeeper server:
-  
+
         $ java -cp zookeeper.jar:lib/slf4j-api-1.7.5.jar:lib/slf4j-log4j12-1.7.5.jar:lib/log4j-1.2.17.jar:conf \\
         org.apache.zookeeper.server.quorum.QuorumPeerMain zoo.cfg
-       
+
   QuorumPeerMain starts a ZooKeeper server,
   [JMX](http://java.sun.com/javase/technologies/core/mntr-mgmt/javamanagement/)
   management beans are also registered which allows
@@ -242,7 +242,7 @@ ensemble:
 8. Test your deployment by connecting to the hosts:
   In Java, you can run the following command to execute
   simple operations:
-  
+
         $ bin/zkCli.sh -server 127.0.0.1:2181
 
 <a name="sc_singleAndDevSetup"></a>
@@ -806,6 +806,15 @@ property, when available, is noted below.
     dropping. This parameter defines the threshold to decrease the dropping
     probability. The default is 0.
 
+ * *clientPortListenBacklog* :
+    **New in 3.4.14, 3.5.5, 3.6.0:**
+    The socket backlog length for the ZooKeeper server socket. This controls
+    the number of requests that will be queued server-side to be processed
+    by the ZooKeeper server. Connections that exceed this length will receive
+    a network timeout (30s) which may cause ZooKeeper session expiry issues.
+    By default, this value is unset (`-1`) which, on Linux, uses a backlog of
+    `50`. This value must be a positive number.
+
 <a name="sc_clusterOptions"></a>
 
 #### Cluster Options
@@ -1411,7 +1420,7 @@ The output contains multiple lines with the following format:
     `TRACE` level first in order to see trace logging
     messages.  The bits of the trace mask correspond to the following
     trace logging categories.
-    
+
     | Trace Mask Bit Values |                     |
     |-----------------------|---------------------|
     | 0b0000000000 | Unused, reserved for future use. |
@@ -1594,7 +1603,7 @@ Running it without any command line parameters or with the `-h,--help` argument,
     -r,--recover   Recovery mode. Re-calculate CRC for broken entries.
     -v,--verbose   Be verbose in recovery mode: print all entries, not just fixed ones.
     -y,--yes       Non-interactive mode: repair all CRC errors without asking
-    
+
 The default behaviour is safe: it dumps the entries of the given
 transaction log file to the screen: (same as using `-d,--dump` parameter)
 
@@ -1694,5 +1703,3 @@ For multi-tenant installations see the [section](zookeeperProgrammers.html#ch_zk
 detailing ZooKeeper "chroot" support, this can be very useful
 when deploying many applications/services interfacing to a
 single ZooKeeper cluster.
-
-

zookeeper-server/src/main/java/org/apache/zookeeper/server/NIOServerCnxnFactory.java

@@ -610,6 +610,7 @@ public class NIOServerCnxnFactory extends ServerCnxnFactory {
         new ConcurrentHashMap<InetAddress, Set<NIOServerCnxn>>( );
 
     protected int maxClientCnxns = 60;
+    int listenBacklog = -1;
 
     int sessionlessCnxnTimeout;
     private ExpiryQueue<NIOServerCnxn> cnxnExpiryQueue;
@@ -637,7 +638,7 @@ public class NIOServerCnxnFactory extends ServerCnxnFactory {
         new HashSet<SelectorThread>();
 
     @Override
-    public void configure(InetSocketAddress addr, int maxcc, boolean secure) throws IOException {
+    public void configure(InetSocketAddress addr, int maxcc, int backlog, boolean secure) throws IOException {
         if (secure) {
             throw new UnsupportedOperationException("SSL isn't supported in NIOServerCnxn");
         }
@@ -679,10 +680,15 @@ public class NIOServerCnxnFactory extends ServerCnxnFactory {
             selectorThreads.add(new SelectorThread(i));
         }
 
+        listenBacklog = backlog;
         this.ss = ServerSocketChannel.open();
         ss.socket().setReuseAddress(true);
         LOG.info("binding to port " + addr);
-        ss.socket().bind(addr);
+        if (listenBacklog == -1) {
+          ss.socket().bind(addr);
+        } else {
+          ss.socket().bind(addr, listenBacklog);
+        }
         ss.configureBlocking(false);
         acceptThread = new AcceptThread(ss, addr, selectorThreads);
     }
@@ -732,6 +738,11 @@ public class NIOServerCnxnFactory extends ServerCnxnFactory {
         maxClientCnxns = max;
     }
 
+    /** {@inheritDoc} */
+    public int getSocketListenBacklog() {
+        return listenBacklog;
+    }
+
     @Override
     public void start() {
         stopped = false;

zookeeper-server/src/main/java/org/apache/zookeeper/server/NettyServerCnxnFactory.java

@@ -82,6 +82,7 @@ public class NettyServerCnxnFactory extends ServerCnxnFactory {
     private final Map<InetAddress, Set<NettyServerCnxn>> ipMap = new HashMap<>();
     private InetSocketAddress localAddress;
     private int maxClientCnxns = 60;
+    int listenBacklog = -1;
     private final ClientX509Util x509Util;
 
     private static final AttributeKey<NettyServerCnxn> CONNECTION_ATTRIBUTE =
@@ -368,13 +369,14 @@ public class NettyServerCnxnFactory extends ServerCnxnFactory {
     }
 
     @Override
-    public void configure(InetSocketAddress addr, int maxClientCnxns, boolean secure)
+    public void configure(InetSocketAddress addr, int maxClientCnxns, int backlog, boolean secure)
             throws IOException
     {
         configureSaslLogin();
         localAddress = addr;
         this.maxClientCnxns = maxClientCnxns;
         this.secure = secure;
+        this.listenBacklog = backlog;
     }
 
     /** {@inheritDoc} */
@@ -387,6 +389,11 @@ public class NettyServerCnxnFactory extends ServerCnxnFactory {
         maxClientCnxns = max;
     }
 
+    /** {@inheritDoc} */
+    public int getSocketListenBacklog() {
+        return listenBacklog;
+    }
+
     @Override
     public int getLocalPort() {
         return localAddress.getPort();
@@ -455,6 +462,9 @@ public class NettyServerCnxnFactory extends ServerCnxnFactory {
     
     @Override
     public void start() {
+        if (listenBacklog != -1) {
+            bootstrap.option(ChannelOption.SO_BACKLOG, listenBacklog);
+        }
         LOG.info("binding to port {}", localAddress);
         parentChannel = bootstrap.bind(localAddress).syncUninterruptibly().channel();
         // Port changes after bind() if the original port was 0, update

zookeeper-server/src/main/java/org/apache/zookeeper/server/ServerCnxnFactory.java

@@ -99,11 +99,15 @@ public abstract class ServerCnxnFactory {
     }
 
     public void configure(InetSocketAddress addr, int maxcc) throws IOException {
-        configure(addr, maxcc, false);
+        configure(addr, maxcc, -1);
     }
 
-    public abstract void configure(InetSocketAddress addr, int maxcc, boolean secure)
-            throws IOException;
+    public void configure(InetSocketAddress addr, int maxcc, int backlog) throws IOException {
+        configure(addr, maxcc, backlog, false);
+    }
+
+    public abstract void configure(InetSocketAddress addr, int maxcc,
+            int backlog, boolean secure) throws IOException;
 
     public abstract void reconfigure(InetSocketAddress addr);
 
@@ -129,6 +133,9 @@ public abstract class ServerCnxnFactory {
     public abstract void startup(ZooKeeperServer zkServer, boolean startServer)
             throws IOException, InterruptedException;
 
+    /** The maximum queue length of the ZooKeeper server's socket */
+    public abstract int getSocketListenBacklog();
+
     public abstract void join() throws InterruptedException;
 
     public abstract void shutdown();
@@ -171,14 +178,26 @@ public abstract class ServerCnxnFactory {
     static public ServerCnxnFactory createFactory(int clientPort,
             int maxClientCnxns) throws IOException
     {
-        return createFactory(new InetSocketAddress(clientPort), maxClientCnxns);
+        return createFactory(new InetSocketAddress(clientPort), maxClientCnxns, -1);
+    }
+
+    static public ServerCnxnFactory createFactory(int clientPort,
+        int maxClientCnxns, int backlog) throws IOException
+    {
+        return createFactory(new InetSocketAddress(clientPort), maxClientCnxns, backlog);
     }
 
     static public ServerCnxnFactory createFactory(InetSocketAddress addr,
             int maxClientCnxns) throws IOException
+    {
+        return createFactory(addr, maxClientCnxns, -1);
+    }
+
+    static public ServerCnxnFactory createFactory(InetSocketAddress addr,
+            int maxClientCnxns, int backlog) throws IOException
     {
         ServerCnxnFactory factory = createFactory();
-        factory.configure(addr, maxClientCnxns);
+        factory.configure(addr, maxClientCnxns, backlog);
         return factory;
     }
 

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

@@ -52,6 +52,8 @@ public class ServerConfig {
     protected int maxSessionTimeout = -1;
     protected String metricsProviderClassName = NullMetricsProvider.class.getName();
     protected Properties metricsProviderConfiguration = new Properties();
+    /** defaults to -1 if not set explicitly */
+    protected int listenBacklog = -1;
 
     /**
      * Parse arguments for server configuration
@@ -105,6 +107,7 @@ public class ServerConfig {
         maxSessionTimeout = config.getMaxSessionTimeout();
         metricsProviderClassName = config.getMetricsProviderClassName();
         metricsProviderConfiguration = config.getMetricsProviderConfiguration();
+        listenBacklog = config.getClientPortListenBacklog();
     }
 
     public InetSocketAddress getClientPortAddress() {
@@ -123,5 +126,6 @@ public class ServerConfig {
     public int getMaxSessionTimeout() { return maxSessionTimeout; }
     public String getMetricsProviderClassName() { return metricsProviderClassName; }
     public Properties getMetricsProviderConfiguration() { return metricsProviderConfiguration; }
-
+    /** Maximum number of pending socket connections to read, -1 if unset */
+    public int getClientPortListenBacklog() { return listenBacklog; }
 }

zookeeper-server/src/main/java/org/apache/zookeeper/server/ZooKeeperServer.java

@@ -103,6 +103,8 @@ public class ZooKeeperServer implements SessionExpirer, ServerStats.Provider {
     protected int minSessionTimeout = -1;
     /** value of -1 indicates unset, use default */
     protected int maxSessionTimeout = -1;
+    /** Socket listen backlog. Value of -1 indicates unset */
+    protected int listenBacklog = -1;
     protected SessionTracker sessionTracker;
     private FileTxnSnapLog txnLogFactory = null;
     private ZKDatabase zkDb;
@@ -190,7 +192,8 @@ public class ZooKeeperServer implements SessionExpirer, ServerStats.Provider {
      * @param dataDir the directory to put the data
      */
     public ZooKeeperServer(FileTxnSnapLog txnLogFactory, int tickTime,
-            int minSessionTimeout, int maxSessionTimeout, ZKDatabase zkDb) {
+            int minSessionTimeout, int maxSessionTimeout, int clientPortListenBacklog,
+            ZKDatabase zkDb) {
         serverStats = new ServerStats(this);
         this.txnLogFactory = txnLogFactory;
         this.txnLogFactory.setServerStats(this.serverStats);
@@ -198,6 +201,8 @@ public class ZooKeeperServer implements SessionExpirer, ServerStats.Provider {
         this.tickTime = tickTime;
         setMinSessionTimeout(minSessionTimeout);
         setMaxSessionTimeout(maxSessionTimeout);
+        this.listenBacklog = clientPortListenBacklog;
+
         listener = new ZooKeeperServerListenerImpl(this);
 
         readResponseCache = new ResponseCache();
@@ -207,6 +212,7 @@ public class ZooKeeperServer implements SessionExpirer, ServerStats.Provider {
         LOG.info("Created server with tickTime " + tickTime
                 + " minSessionTimeout " + getMinSessionTimeout()
                 + " maxSessionTimeout " + getMaxSessionTimeout()
+                + " clientPortListenBacklog " + getClientPortListenBacklog()
                 + " datadir " + txnLogFactory.getDataDir()
                 + " snapdir " + txnLogFactory.getSnapDir());
     }
@@ -219,7 +225,7 @@ public class ZooKeeperServer implements SessionExpirer, ServerStats.Provider {
      */
     public ZooKeeperServer(FileTxnSnapLog txnLogFactory, int tickTime)
             throws IOException {
-        this(txnLogFactory, tickTime, -1, -1, new ZKDatabase(txnLogFactory));
+        this(txnLogFactory, tickTime, -1, -1, -1, new ZKDatabase(txnLogFactory));
     }
 
     public ServerStats serverStats() {
@@ -251,6 +257,8 @@ public class ZooKeeperServer implements SessionExpirer, ServerStats.Provider {
         pwriter.println(getMinSessionTimeout());
         pwriter.print("maxSessionTimeout=");
         pwriter.println(getMaxSessionTimeout());
+        pwriter.print("clientPortListenBacklog=");
+        pwriter.println(getClientPortListenBacklog());
 
         pwriter.print("serverId=");
         pwriter.println(getServerId());
@@ -265,7 +273,8 @@ public class ZooKeeperServer implements SessionExpirer, ServerStats.Provider {
              serverCnxnFactory.getMaxClientCnxnsPerHost(),
              getMinSessionTimeout(),
              getMaxSessionTimeout(),
-             getServerId());
+             getServerId(),
+             getClientPortListenBacklog());
     }
 
     /**
@@ -287,7 +296,7 @@ public class ZooKeeperServer implements SessionExpirer, ServerStats.Provider {
     public ZooKeeperServer(FileTxnSnapLog txnLogFactory)
         throws IOException
     {
-        this(txnLogFactory, DEFAULT_TICK_TIME, -1, -1, new ZKDatabase(txnLogFactory));
+        this(txnLogFactory, DEFAULT_TICK_TIME, -1, -1, -1, new ZKDatabase(txnLogFactory));
     }
 
     /**
@@ -1008,6 +1017,15 @@ public class ZooKeeperServer implements SessionExpirer, ServerStats.Provider {
         LOG.info("maxSessionTimeout set to {}", this.maxSessionTimeout);
     }
 
+    public int getClientPortListenBacklog() {
+        return listenBacklog;
+    }
+
+    public void setClientPortListenBacklog(int backlog) {
+        this.listenBacklog = backlog;
+        LOG.info("clientPortListenBacklog set to " + backlog);
+    }
+
     public int getClientPort() {
         return serverCnxnFactory != null ? serverCnxnFactory.getLocalPort() : -1;
     }

zookeeper-server/src/main/java/org/apache/zookeeper/server/ZooKeeperServerConf.java

@@ -60,6 +60,12 @@ public class ZooKeeperServerConf {
      * The key in the map returned by {@link #toMap()} for the server ID.
      */
     public static final String KEY_SERVER_ID = "server_id";
+    /**
+     * The key in the map returned by {@link #toMap()} for the server socket
+     * listen backlog.
+     */
+    public static final String KEY_CLIENT_PORT_LISTEN_BACKLOG
+            = "client_port_listen_backlog";
 
     private final int clientPort;
     private final String dataDir;
@@ -69,6 +75,7 @@ public class ZooKeeperServerConf {
     private final int minSessionTimeout;
     private final int maxSessionTimeout;
     private final long serverId;
+    private final int clientPortListenBacklog;
 
     /**
      * Creates a new configuration.
@@ -85,7 +92,7 @@ public class ZooKeeperServerConf {
     ZooKeeperServerConf(int clientPort, String dataDir, String dataLogDir,
                         int tickTime, int maxClientCnxnsPerHost,
                         int minSessionTimeout, int maxSessionTimeout,
-                        long serverId) {
+                        long serverId, int clientPortListenBacklog) {
         this.clientPort = clientPort;
         this.dataDir = dataDir;
         this.dataLogDir = dataLogDir;
@@ -94,6 +101,7 @@ public class ZooKeeperServerConf {
         this.minSessionTimeout = minSessionTimeout;
         this.maxSessionTimeout = maxSessionTimeout;
         this.serverId = serverId;
+        this.clientPortListenBacklog = clientPortListenBacklog;
     }
 
     /**
@@ -168,6 +176,13 @@ public class ZooKeeperServerConf {
         return serverId;
     }
 
+    /**
+     * Returns the server socket listen backlog length.
+     */
+    public int getClientPortListenBacklog() {
+        return clientPortListenBacklog;
+    }
+
     /**
      * Converts this configuration to a map. The returned map is mutable, and
      * changes to it do not reflect back into this configuration.
@@ -184,6 +199,7 @@ public class ZooKeeperServerConf {
         conf.put(KEY_MIN_SESSION_TIMEOUT, minSessionTimeout);
         conf.put(KEY_MAX_SESSION_TIMEOUT, maxSessionTimeout);
         conf.put(KEY_SERVER_ID, serverId);
+        conf.put(KEY_CLIENT_PORT_LISTEN_BACKLOG, clientPortListenBacklog);
         return conf;
     }
 }

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

@@ -136,7 +136,8 @@ public class ZooKeeperServerMain {
             // create a file logger url from the command line args
             txnLog = new FileTxnSnapLog(config.dataLogDir, config.dataDir);
             final ZooKeeperServer zkServer = new ZooKeeperServer(txnLog,
-                    config.tickTime, config.minSessionTimeout, config.maxSessionTimeout, null);
+                    config.tickTime, config.minSessionTimeout, config.maxSessionTimeout,
+                    config.listenBacklog, null);
             zkServer.setRootMetricsContext(metricsProvider.getRootContext());
             txnLog.setServerStats(zkServer.serverStats());
 
@@ -154,14 +155,16 @@ public class ZooKeeperServerMain {
             boolean needStartZKServer = true;
             if (config.getClientPortAddress() != null) {
                 cnxnFactory = ServerCnxnFactory.createFactory();
-                cnxnFactory.configure(config.getClientPortAddress(), config.getMaxClientCnxns(), false);
+                cnxnFactory.configure(config.getClientPortAddress(), config.getMaxClientCnxns(),
+                    config.getClientPortListenBacklog(), false);
                 cnxnFactory.startup(zkServer);
                 // zkServer has been started. So we don't need to start it again in secureCnxnFactory.
                 needStartZKServer = false;
             }
             if (config.getSecureClientPortAddress() != null) {
                 secureCnxnFactory = ServerCnxnFactory.createFactory();
-                secureCnxnFactory.configure(config.getSecureClientPortAddress(), config.getMaxClientCnxns(), true);
+                secureCnxnFactory.configure(config.getSecureClientPortAddress(), config.getMaxClientCnxns(),
+                    config.getClientPortListenBacklog(), true);
                 secureCnxnFactory.startup(zkServer, needStartZKServer);
             }
 

zookeeper-server/src/main/java/org/apache/zookeeper/server/quorum/FollowerZooKeeperServer.java

@@ -61,7 +61,8 @@ public class FollowerZooKeeperServer extends LearnerZooKeeperServer {
     FollowerZooKeeperServer(FileTxnSnapLog logFactory,QuorumPeer self,
             ZKDatabase zkDb) throws IOException {
         super(logFactory, self.tickTime, self.minSessionTimeout,
-                self.maxSessionTimeout, zkDb, self);
+                self.maxSessionTimeout, self.clientPortListenBacklog,
+                zkDb, self);
         this.pendingSyncs = new ConcurrentLinkedQueue<Request>();
     }
 

zookeeper-server/src/main/java/org/apache/zookeeper/server/quorum/LeaderZooKeeperServer.java

@@ -55,7 +55,8 @@ public class LeaderZooKeeperServer extends QuorumZooKeeperServer {
      * @throws IOException
      */
     LeaderZooKeeperServer(FileTxnSnapLog logFactory, QuorumPeer self, ZKDatabase zkDb) throws IOException {
-        super(logFactory, self.tickTime, self.minSessionTimeout, self.maxSessionTimeout, zkDb, self);
+        super(logFactory, self.tickTime, self.minSessionTimeout, self.maxSessionTimeout,
+                self.clientPortListenBacklog, zkDb, self);
     }
 
     public Leader getLeader(){

zookeeper-server/src/main/java/org/apache/zookeeper/server/quorum/LearnerZooKeeperServer.java

@@ -42,11 +42,12 @@ public abstract class LearnerZooKeeperServer extends QuorumZooKeeperServer {
     protected SyncRequestProcessor syncProcessor;
 
     public LearnerZooKeeperServer(FileTxnSnapLog logFactory, int tickTime,
-            int minSessionTimeout, int maxSessionTimeout,
+            int minSessionTimeout, int maxSessionTimeout, int listenBacklog,
             ZKDatabase zkDb, QuorumPeer self)
         throws IOException
     {
-        super(logFactory, tickTime, minSessionTimeout, maxSessionTimeout, zkDb, self);
+        super(logFactory, tickTime, minSessionTimeout, maxSessionTimeout,
+            listenBacklog, zkDb, self);
     }
 
     /**

zookeeper-server/src/main/java/org/apache/zookeeper/server/quorum/ObserverZooKeeperServer.java

@@ -52,7 +52,7 @@ public class ObserverZooKeeperServer extends LearnerZooKeeperServer {
         new ConcurrentLinkedQueue<Request>();
 
     ObserverZooKeeperServer(FileTxnSnapLog logFactory, QuorumPeer self, ZKDatabase zkDb) throws IOException {
-        super(logFactory, self.tickTime, self.minSessionTimeout, self.maxSessionTimeout, zkDb, self);
+        super(logFactory, self.tickTime, self.minSessionTimeout, self.maxSessionTimeout, self.clientPortListenBacklog, zkDb, self);
         LOG.info("syncEnabled =" + syncRequestProcessorEnabled);
     }
     

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

@@ -557,6 +557,13 @@ public class QuorumPeer extends ZooKeeperThread implements QuorumStats.Provider
      */
     protected int maxSessionTimeout = -1;
 
+    /**
+     * The ZooKeeper server's socket backlog length. The number of connections
+     * that will be queued to be read before new connections are dropped. A
+     * value of one indicates the default backlog will be used.
+     */
+    protected int clientPortListenBacklog = -1;
+
     /**
      * The number of ticks that the initial synchronization phase can take
      */
@@ -1509,6 +1516,16 @@ public class QuorumPeer extends ZooKeeperThread implements QuorumStats.Provider
         this.maxSessionTimeout = max;
     }
 
+    /** The server socket's listen backlog length */
+    public int getClientPortListenBacklog() {
+        return this.clientPortListenBacklog;
+    }
+
+    /** Sets the server socket's listen backlog length. */
+    public void setClientPortListenBacklog(int backlog) {
+        this.clientPortListenBacklog = backlog;
+    }
+
     /**
      * Get the number of ticks that the initial synchronization phase can take
      */

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

@@ -87,6 +87,8 @@ public class QuorumPeerConfig {
     protected Properties metricsProviderConfiguration = new Properties();
     protected boolean localSessionsEnabled = false;
     protected boolean localSessionsUpgradingEnabled = false;
+    /** defaults to -1 if not set explicitly */
+    protected int clientPortListenBacklog = -1;
 
     protected int initLimit;
     protected int syncLimit;
@@ -266,6 +268,8 @@ public class QuorumPeerConfig {
                 secureClientPortAddress = value.trim();
             } else if (key.equals("observerMasterPort")) {
                 observerMasterPort = Integer.parseInt(value);
+            } else if (key.equals("clientPortListenBacklog")) {
+                clientPortListenBacklog = Integer.parseInt(value);
             } else if (key.equals("tickTime")) {
                 tickTime = Integer.parseInt(value);
             } else if (key.equals("maxClientCnxns")) {
@@ -789,6 +793,7 @@ public class QuorumPeerConfig {
     public boolean shouldUsePortUnification() {
         return shouldUsePortUnification;
     }
+    public int getClientPortListenBacklog() { return clientPortListenBacklog; }
 
     public int getInitLimit() { return initLimit; }
     public int getSyncLimit() { return syncLimit; }

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

@@ -162,14 +162,14 @@ public class QuorumPeerMain {
               cnxnFactory = ServerCnxnFactory.createFactory();
               cnxnFactory.configure(config.getClientPortAddress(),
                       config.getMaxClientCnxns(),
-                      false);
+                      config.getClientPortListenBacklog(), false);
           }
 
           if (config.getSecureClientPortAddress() != null) {
               secureCnxnFactory = ServerCnxnFactory.createFactory();
               secureCnxnFactory.configure(config.getSecureClientPortAddress(),
                       config.getMaxClientCnxns(),
-                      true);
+                      config.getClientPortListenBacklog(), true);
           }
 
           quorumPeer = getQuorumPeer();
@@ -190,6 +190,7 @@ public class QuorumPeerMain {
           quorumPeer.setSyncLimit(config.getSyncLimit());
           quorumPeer.setObserverMasterPort(config.getObserverMasterPort());
           quorumPeer.setConfigFileName(config.getConfigFilename());
+          quorumPeer.setClientPortListenBacklog(config.getClientPortListenBacklog());
           quorumPeer.setZKDatabase(new ZKDatabase(quorumPeer.getTxnFactory()));
           quorumPeer.setQuorumVerifier(config.getQuorumVerifier(), false);
           if (config.getLastSeenQuorumVerifier()!=null) {

zookeeper-server/src/main/java/org/apache/zookeeper/server/quorum/QuorumZooKeeperServer.java

@@ -43,10 +43,10 @@ public abstract class QuorumZooKeeperServer extends ZooKeeperServer {
     protected UpgradeableSessionTracker upgradeableSessionTracker;
 
     protected QuorumZooKeeperServer(FileTxnSnapLog logFactory, int tickTime,
-            int minSessionTimeout, int maxSessionTimeout,
+            int minSessionTimeout, int maxSessionTimeout, int listenBacklog,
             ZKDatabase zkDb, QuorumPeer self)
     {
-        super(logFactory, tickTime, minSessionTimeout, maxSessionTimeout, zkDb);
+        super(logFactory, tickTime, minSessionTimeout, maxSessionTimeout, listenBacklog, zkDb);
         this.self = self;
     }
 

zookeeper-server/src/main/java/org/apache/zookeeper/server/quorum/ReadOnlyZooKeeperServer.java

@@ -46,7 +46,7 @@ public class ReadOnlyZooKeeperServer extends ZooKeeperServer {
     ReadOnlyZooKeeperServer(FileTxnSnapLog logFactory, QuorumPeer self,
                             ZKDatabase zkDb) {
         super(logFactory, self.tickTime, self.minSessionTimeout,
-              self.maxSessionTimeout, zkDb);
+              self.maxSessionTimeout, self.clientPortListenBacklog, zkDb);
         this.self = self;
     }
 

zookeeper-server/src/test/java/org/apache/zookeeper/server/ZooKeeperServerBeanTest.java

@@ -99,7 +99,7 @@ public class ZooKeeperServerBeanTest {
         ServerCnxnFactory cnxnFactory = ServerCnxnFactory.createFactory();
         int secureClientPort = 8443;
         InetSocketAddress address = new InetSocketAddress(secureClientPort);
-        cnxnFactory.configure(address, 5, true);
+        cnxnFactory.configure(address, 5, -1, true);
         zks.setSecureServerCnxnFactory(cnxnFactory);
 
         result = serverBean.getSecureClientPort();
@@ -129,7 +129,7 @@ public class ZooKeeperServerBeanTest {
         ServerCnxnFactory cnxnFactory = ServerCnxnFactory.createFactory();
         int secureClientPort = 8443;
         InetSocketAddress address = new InetSocketAddress(secureClientPort);
-        cnxnFactory.configure(address, 5, true);
+        cnxnFactory.configure(address, 5, -1, true);
         zks.setSecureServerCnxnFactory(cnxnFactory);
 
         result = serverBean.getSecureClientAddress();

zookeeper-server/src/test/java/org/apache/zookeeper/server/ZooKeeperServerConfTest.java

@@ -25,7 +25,7 @@ import static org.junit.Assert.*;
 public class ZooKeeperServerConfTest extends ZKTestCase {
     private ZooKeeperServerConf c;
     @Before public void setUp() {
-        c = new ZooKeeperServerConf(1, "a", "b", 2, 3, 4, 5, 6L);
+        c = new ZooKeeperServerConf(1, "a", "b", 2, 3, 4, 5, 6L, 7);
     }
     @Test public void testGetters() {
         assertEquals(1, c.getClientPort());
@@ -36,10 +36,11 @@ public class ZooKeeperServerConfTest extends ZKTestCase {
         assertEquals(4, c.getMinSessionTimeout());
         assertEquals(5, c.getMaxSessionTimeout());
         assertEquals(6L, c.getServerId());
+        assertEquals(7, c.getClientPortListenBacklog());
     }
     @Test public void testToMap() {
         Map<String, Object> m = c.toMap();
-        assertEquals(8, m.size());
+        assertEquals(9, m.size());
         assertEquals(Integer.valueOf(1), m.get(ZooKeeperServerConf.KEY_CLIENT_PORT));
         assertEquals("a", m.get(ZooKeeperServerConf.KEY_DATA_DIR));
         assertEquals("b", m.get(ZooKeeperServerConf.KEY_DATA_LOG_DIR));
@@ -48,5 +49,6 @@ public class ZooKeeperServerConfTest extends ZKTestCase {
         assertEquals(Integer.valueOf(4), m.get(ZooKeeperServerConf.KEY_MIN_SESSION_TIMEOUT));
         assertEquals(Integer.valueOf(5), m.get(ZooKeeperServerConf.KEY_MAX_SESSION_TIMEOUT));
         assertEquals(Long.valueOf(6L), m.get(ZooKeeperServerConf.KEY_SERVER_ID));
+        assertEquals(Integer.valueOf(7), m.get(ZooKeeperServerConf.KEY_CLIENT_PORT_LISTEN_BACKLOG));
     }
 }

zookeeper-server/src/test/java/org/apache/zookeeper/server/admin/CommandsTest.java

@@ -109,7 +109,8 @@ public class CommandsTest extends ClientBase {
                     new Field("max_client_cnxns", Integer.class),
                     new Field("min_session_timeout", Integer.class),
                     new Field("max_session_timeout", Integer.class),
-                    new Field("server_id", Long.class));
+                    new Field("server_id", Long.class),
+                    new Field("client_port_listen_backlog", Integer.class));
     }
 
     @Test

zookeeper-server/src/test/java/org/apache/zookeeper/server/quorum/LearnerTest.java

@@ -51,7 +51,7 @@ public class LearnerTest extends ZKTestCase {
 
         public SimpleLearnerZooKeeperServer(FileTxnSnapLog ftsl, QuorumPeer self)
                 throws IOException {
-            super(ftsl, 2000, 2000, 2000, new ZKDatabase(ftsl), self);
+            super(ftsl, 2000, 2000, 2000, -1, new ZKDatabase(ftsl), self);
         }
 
         @Override

zookeeper-server/src/test/java/org/apache/zookeeper/server/quorum/LocalPeerBeanTest.java

@@ -55,7 +55,7 @@ public class LocalPeerBeanTest {
         ServerCnxnFactory cnxnFactory = ServerCnxnFactory.createFactory();
         int clientPort = PortAssignment.unique();
         InetSocketAddress address = new InetSocketAddress(clientPort);
-        cnxnFactory.configure(address, 5, false);
+        cnxnFactory.configure(address, 5, -1, false);
         quorumPeer.setCnxnFactory(cnxnFactory);
 
         result = remotePeerBean.getClientAddress();
@@ -72,7 +72,7 @@ public class LocalPeerBeanTest {
         InetAddress clientIP = InetAddress.getLoopbackAddress();
         address = new InetSocketAddress(clientIP, clientPort);
         cnxnFactory = ServerCnxnFactory.createFactory();
-        cnxnFactory.configure(address, 5, false);
+        cnxnFactory.configure(address, 5, -1, false);
         quorumPeer.setCnxnFactory(cnxnFactory);
 
         result = remotePeerBean.getClientAddress();

zookeeper-server/src/test/java/org/apache/zookeeper/server/quorum/ZabUtils.java

@@ -112,6 +112,9 @@ public class ZabUtils {
         public int getMaxClientCnxnsPerHost() {
             return 0;
         }
+        public int getSocketListenBacklog() {
+            return -1;
+        }
         public int getLocalPort() {
             return 0;
         }
@@ -121,8 +124,8 @@ public class ZabUtils {
         public Iterable<ServerCnxn> getConnections() {
             return null;
         }
-        public void configure(InetSocketAddress addr, int maxcc, boolean secure)
-                throws IOException {
+        public void configure(InetSocketAddress addr, int maxcc, int listenBacklog,
+                boolean secure) throws IOException {
         }
 
         public boolean closeSession(long sessionId) {

zookeeper-server/src/test/java/org/apache/zookeeper/test/SSLAuthTest.java

@@ -54,7 +54,7 @@ public class SSLAuthTest extends ClientBase {
         hostPort = host + ":" + port;
 
         serverFactory = ServerCnxnFactory.createFactory();
-        serverFactory.configure(new InetSocketAddress(host, port), maxCnxns, true);
+        serverFactory.configure(new InetSocketAddress(host, port), maxCnxns, -1, true);
 
         super.setUp();
     }