ZOOKEEPER-3712: Add setKeepAlive support for NIOServerCnxn.

Author: Fangxi Yin <yinfangxikuaishou.com>

Author: yinfangxi <yinfangxi@kuaishou.com>

Reviewers: Michael Han <hanm@apache.org>, Justin Ling Mao <maoling199210191@sina.com>

Closes #1242 from yfxhust/ZOOKEEPER-3712

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

@@ -1335,7 +1335,23 @@ As an example, this will enable all four letter word commands:
     properly, check your operating system's options regarding TCP
     keepalive for more information.  Defaults to
     **false**.
-    
+
+* *clientTcpKeepAlive* :
+    (Java system property: **zookeeper.clientTcpKeepAlive**)
+    **New in 3.6.1:**
+    Setting this to true sets the TCP keepAlive flag on the
+    client sockets. Some broken network infrastructure may lose
+    the FIN packet that is sent from closing client. These never
+    closed client sockets cause OS resource leak. Enabling this
+    option terminates these zombie sockets by idle check.
+    Enabling this option relies on OS level settings to work
+    properly, check your operating system's options regarding TCP
+    keepalive for more information. Defaults to **false**. Please
+    note the distinction between it and **tcpKeepAlive**. It is
+    applied for the client sockets while **tcpKeepAlive** is for
+    the sockets used by quorum members. Currently this option is
+    only available when default `NIOServerCnxnFactory` is used.
+
 * *electionPortBindRetry* :
     (Java system property only: **zookeeper.electionPortBindRetry**)
     Property set max retry count when Zookeeper server fails to bind 

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

@@ -81,6 +81,11 @@ public class NIOServerCnxn extends ServerCnxn {
      */
     private long sessionId;
 
+    /**
+     * Client socket option for TCP keepalive
+     */
+    private final boolean clientTcpKeepAlive = Boolean.getBoolean("zookeeper.clientTcpKeepAlive");
+
     public NIOServerCnxn(ZooKeeperServer zk, SocketChannel sock, SelectionKey sk, NIOServerCnxnFactory factory, SelectorThread selectorThread) throws IOException {
         super(zk);
         this.sock = sock;
@@ -93,6 +98,7 @@ public class NIOServerCnxn extends ServerCnxn {
         sock.socket().setTcpNoDelay(true);
         /* set socket linger to false, so that socket close does not block */
         sock.socket().setSoLinger(false, -1);
+        sock.socket().setKeepAlive(clientTcpKeepAlive);
         InetAddress addr = ((InetSocketAddress) sock.socket().getRemoteSocketAddress()).getAddress();
         addAuthInfo(new Id("ip", addr.getHostAddress()));
         this.sessionTimeout = factory.sessionlessCnxnTimeout;