ZOOKEEPER-2442: Socket leak in QuorumCnxManager connectOne

rgs1 · rgs1 · commit 3c37184e83a3 · 2016-06-08T15:43:15.000Z
(Michael Han via rgs) git-svn-id: https://svn.apache.org/repos/asf/zookeeper/trunk@1747408 13f79535-47bb-0310-9956-ffa450edef68
diff --git a/CHANGES.txt b/CHANGES.txt
@@ -308,6 +308,9 @@ BUGFIXES:
   ZOOKEEPER-2405: getTGT() in Login.java mishandles confidential
   information (Michael Han via phunt)
 
+  ZOOKEEPER-2442: Socket leak in QuorumCnxManager connectOne
+  (Michael Han via rgs)
+
 IMPROVEMENTS:
   ZOOKEEPER-2024 Major throughput improvement with mixed workloads (Kfir Lev-Ari via shralex)
 
diff --git a/src/java/main/org/apache/zookeeper/server/quorum/QuorumCnxManager.java b/src/java/main/org/apache/zookeeper/server/quorum/QuorumCnxManager.java
@@ -237,9 +237,7 @@ public QuorumCnxManager(QuorumPeer self) {
      * @param sid
      */
     public void testInitiateConnection(long sid) throws Exception {
-        if (LOG.isDebugEnabled()) {
-            LOG.debug("Opening channel to server " + sid);
-        }
+        LOG.debug("Opening channel to server " + sid);
         Socket sock = new Socket();
         setSockOpts(sock);
         sock.connect(self.getVotingView().get(sid).electionAddr, cnxTO);
@@ -434,17 +432,14 @@ synchronized private boolean connectOne(long sid, InetSocketAddress electionAddr
             LOG.debug("There is a connection already for server " + sid);
             return true;
         }
-        try {
 
-             if (LOG.isDebugEnabled()) {
-                 LOG.debug("Opening channel to server " + sid);
-             }
-             Socket sock = new Socket();
+        Socket sock = null;
+        try {
+             LOG.debug("Opening channel to server " + sid);
+             sock = new Socket();
              setSockOpts(sock);
              sock.connect(electionAddr, cnxTO);
-             if (LOG.isDebugEnabled()) {
-                 LOG.debug("Connected to server " + sid);
-             }
+             LOG.debug("Connected to server " + sid);
              initiateConnection(sock, sid);
              return true;
          } catch (UnresolvedAddressException e) {
@@ -454,11 +449,13 @@ synchronized private boolean connectOne(long sid, InetSocketAddress electionAddr
              // detail.
              LOG.warn("Cannot open channel to " + sid
                      + " at election address " + electionAddr, e);
+             closeSocket(sock);
              throw e;
          } catch (IOException e) {
              LOG.warn("Cannot open channel to " + sid
                      + " at election address " + electionAddr,
                      e);
+             closeSocket(sock);
              return false;
          }
    
@@ -574,6 +571,10 @@ private void setSockOpts(Socket sock) throws SocketException {
      *            Reference to socket
      */
     private void closeSocket(Socket sock) {
+        if (sock == null) {
+            return;
+        }
+
         try {
             sock.close();
         } catch (IOException ie) {
@@ -614,7 +615,7 @@ public Listener() {
         public void run() {
             int numRetries = 0;
             InetSocketAddress addr;
-
+            Socket client = null;
             while((!shutdown) && (numRetries < 3)){
                 try {
                     ss = new ServerSocket();
@@ -632,7 +633,7 @@ public void run() {
                     setName(addr.toString());
                     ss.bind(addr);
                     while (!shutdown) {
-                        Socket client = ss.accept();
+                        client = ss.accept();
                         setSockOpts(client);
                         LOG.info("Received connection request "
                                 + client.getRemoteSocketAddress());
@@ -654,6 +655,7 @@ public void run() {
                         LOG.error("Interrupted while sleeping. " +
                             "Ignoring exception", ie);
                     }
+                    closeSocket(client);
                 }
             }
             LOG.info("Leaving listener");
@@ -739,9 +741,7 @@ synchronized RecvWorker getRecvWorker(){
         }
                 
         synchronized boolean finish() {
-            if (LOG.isDebugEnabled()) {
-                LOG.debug("Calling finish for " + sid);
-            }
+            LOG.debug("Calling finish for " + sid);
             
             if(!running){
                 /*
@@ -752,16 +752,14 @@ synchronized boolean finish() {
             
             running = false;
             closeSocket(sock);
-            // channel = null;
 
             this.interrupt();
             if (recvWorker != null) {
                 recvWorker.finish();
             }
 
-            if (LOG.isDebugEnabled()) {
-                LOG.debug("Removing entry from senderWorkerMap sid=" + sid);
-            }
+            LOG.debug("Removing entry from senderWorkerMap sid=" + sid);
+
             senderWorkerMap.remove(sid, this);
             threadCnt.decrementAndGet();
             return running;
@@ -919,9 +917,7 @@ public void run() {
             } finally {
                 LOG.warn("Interrupting SendWorker");
                 sw.finish();
-                if (sock != null) {
-                    closeSocket(sock);
-                }
+                closeSocket(sock);
             }
         }
     }
