[ZOOKEEPER-3384] Add SO_LINGER socket option to avoid long quorum unavailable during close with full send buffer

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

@@ -59,6 +59,7 @@
 import org.apache.zookeeper.server.quorum.flexible.QuorumVerifier;
 import org.apache.zookeeper.server.util.SerializeUtils;
 import org.apache.zookeeper.server.util.ZxidUtils;
+import org.apache.zookeeper.util.SocketUtils;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -435,7 +436,7 @@ public void run() {
 
                         // start with the initLimit, once the ack is processed
                         // in LearnerHandler switch to the syncLimit
-                        s.setSoTimeout(self.tickTime * self.initLimit);
+                        SocketUtils.setSocketOption(s, self.tickTime * self.initLimit);
                         s.setTcpNoDelay(nodelay);
 
                         BufferedInputStream is = new BufferedInputStream(s.getInputStream());

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

@@ -52,6 +52,7 @@
 import org.apache.zookeeper.server.util.ZxidUtils;
 import org.apache.zookeeper.txn.SetDataTxn;
 import org.apache.zookeeper.txn.TxnHeader;
+import org.apache.zookeeper.util.SocketUtils;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -323,7 +324,7 @@ private Socket createSocket() throws X509Exception, IOException {
         } else {
             sock = new Socket();
         }
-        sock.setSoTimeout(self.tickTime * self.initLimit);
+        SocketUtils.setSocketOption(sock, self.tickTime * self.initLimit);
         return sock;
     }
 

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

@@ -42,6 +42,7 @@
 import org.apache.zookeeper.server.Request;
 import org.apache.zookeeper.server.ZKDatabase;
 import org.apache.zookeeper.server.quorum.auth.QuorumAuthServer;
+import org.apache.zookeeper.util.SocketUtils;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -463,7 +464,7 @@ public void run() {
 
                 // start with the initLimit, once the ack is processed
                 // in LearnerHandler switch to the syncLimit
-                s.setSoTimeout(self.tickTime * self.initLimit);
+                SocketUtils.setSocketOption(s, self.tickTime * self.initLimit);
                 BufferedInputStream is = new BufferedInputStream(s.getInputStream());
                 LearnerHandler lh = new LearnerHandler(s, is, this);
                 lh.start();

zookeeper-server/src/main/java/org/apache/zookeeper/util/SocketUtils.java

@@ -0,0 +1,44 @@
+/**
+ * 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.util;
+
+import java.net.Socket;
+import java.net.SocketException;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Util class used to set the common options on socket.
+ */
+public class SocketUtils {
+    private static final Logger LOG = LoggerFactory.getLogger(SocketUtils.class);
+
+    public static final String SOCKET_SO_LINGER = "zookeeper.quorum.solinger";
+    private static int soLinger = 0;
+
+    static {
+        soLinger = Integer.getInteger(SOCKET_SO_LINGER, 0);
+        LOG.info("{} = {}", SOCKET_SO_LINGER, soLinger);
+    }
+
+    public static void setSocketOption(Socket sock, int soTimeout) throws SocketException {
+        sock.setSoTimeout(soTimeout);
+        sock.setSoLinger(true, soLinger);
+    }
+}