ZOOKEEPER-2366: Reconfiguration of client port causes a socket leak. (fpj via cnauroth)

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

CHANGES.txt

@@ -320,6 +320,9 @@ BUGFIXES:
   ZOOKEEPER-2380: Deadlock between leader shutdown and forwarding ACK to the
   leader (Arshad Mohammad via cnauroth)
 
+  ZOOKEEPER-2366: Reconfiguration of client port causes a socket leak.
+  (fpj via cnauroth)
+
 IMPROVEMENTS:
   ZOOKEEPER-2024 Major throughput improvement with mixed workloads (Kfir Lev-Ari via shralex)
 

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

@@ -688,31 +688,42 @@ public class NIOServerCnxnFactory extends ServerCnxnFactory {
         ss.configureBlocking(false);
         acceptThread = new AcceptThread(ss, addr, selectorThreads);
     }
-   
+
+    private void tryClose(ServerSocketChannel s) {
+        try {
+            s.close();
+        } catch (IOException sse) {
+            LOG.error("Error while closing server socket.", sse);
+        }
+    }
+
     @Override
-    public void reconfigure(InetSocketAddress addr){
+    public void reconfigure(InetSocketAddress addr) {
         ServerSocketChannel oldSS = ss;        
         try {
-           this.ss = ServerSocketChannel.open();
-           ss.socket().setReuseAddress(true);
-           LOG.info("binding to port " + addr);
-           ss.socket().bind(addr);
-           ss.configureBlocking(false);
-           acceptThread.setReconfiguring();
-           oldSS.close();           
-           acceptThread.wakeupSelector();
-           try {
-			  acceptThread.join();
-		   } catch (InterruptedException e) {
-			   LOG.error("Error joining old acceptThread when reconfiguring client port " + e.getMessage());
-		   }
-           acceptThread = new AcceptThread(ss, addr, selectorThreads);
-           acceptThread.start();
+            this.ss = ServerSocketChannel.open();
+            ss.socket().setReuseAddress(true);
+            LOG.info("binding to port " + addr);
+            ss.socket().bind(addr);
+            ss.configureBlocking(false);
+            acceptThread.setReconfiguring();
+            tryClose(oldSS);
+            acceptThread.wakeupSelector();
+            try {
+                acceptThread.join();
+            } catch (InterruptedException e) {
+                LOG.error("Error joining old acceptThread when reconfiguring client port {}",
+                            e.getMessage());
+                Thread.currentThread().interrupt();
+            }
+            acceptThread = new AcceptThread(ss, addr, selectorThreads);
+            acceptThread.start();
         } catch(IOException e) {
-           LOG.error("Error reconfiguring client port to " + addr + " " + e.getMessage());
+            LOG.error("Error reconfiguring client port to {} {}", addr, e.getMessage());
+            tryClose(oldSS);
         }
     }
-    
+
     /** {@inheritDoc} */
     public int getMaxClientCnxnsPerHost() {
         return maxClientCnxns;

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

@@ -488,13 +488,17 @@ public class NettyServerCnxnFactory extends ServerCnxnFactory {
         parentChannel = bootstrap.bind(localAddress);
     }
     
-    public void reconfigure(InetSocketAddress addr) 
-    {  
+    public void reconfigure(InetSocketAddress addr) {
        Channel oldChannel = parentChannel;
-       LOG.info("binding to port " + addr);
-        parentChannel = bootstrap.bind(addr);
-        localAddress = addr;  
-        oldChannel.close();
+       try {
+           LOG.info("binding to port {}", addr);
+           parentChannel = bootstrap.bind(addr);
+           localAddress = addr;
+       } catch (Exception e) {
+           LOG.error("Error while reconfiguring", e);
+       } finally {
+           oldChannel.close();
+       }
     }
     
     @Override

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

@@ -710,7 +710,7 @@ public class Leader {
        // that different operations wait for different sets of acks, and we still want to enforce
        // that they are committed in order. Currently we only permit one outstanding reconfiguration
        // such that the reconfiguration and subsequent outstanding ops proposed while the reconfig is
-       // pending all wait for a quorum of old and new config, so its not possible to get enough acks
+       // pending all wait for a quorum of old and new config, so it's not possible to get enough acks
        // for an operation without getting enough acks for preceding ops. But in the future if multiple
        // concurrent reconfigs are allowed, this can happen.
        if (outstandingProposals.containsKey(zxid - 1)) return false;
@@ -751,7 +751,7 @@ public class Leader {
             QuorumVerifier newQV = p.qvAcksetPairs.get(p.qvAcksetPairs.size()-1).getQuorumVerifier();
        
             self.processReconfig(newQV, designatedLeader, zk.getZxid(), true);
-       
+
             if (designatedLeader != self.getId()) {
                 allowedToCommit = false;
             }
@@ -1261,7 +1261,7 @@ public class Leader {
         QuorumVerifier newQV = self.getLastSeenQuorumVerifier();
         
         Long designatedLeader = getDesignatedLeader(newLeaderProposal, zk.getZxid());                                         
-        
+
         self.processReconfig(newQV, designatedLeader, zk.getZxid(), true);
         if (designatedLeader != self.getId()) {
             allowedToCommit = false;

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

@@ -1728,7 +1728,7 @@ public class QuorumPeer extends ZooKeeperThread implements QuorumStats.Provider
         writeLongToFile(ACCEPTED_EPOCH_FILENAME, e);
     }
    
-    public boolean processReconfig(QuorumVerifier qv, Long suggestedLeaderId, Long zxid, boolean restartLE){
+    public boolean processReconfig(QuorumVerifier qv, Long suggestedLeaderId, Long zxid, boolean restartLE) {
        InetSocketAddress oldClientAddr = getClientAddress();
 
        // update last committed quorum verifier, write the new config to disk
@@ -1756,8 +1756,8 @@ public class QuorumPeer extends ZooKeeperThread implements QuorumStats.Provider
                cnxnFactory.reconfigure(myNewQS.clientAddr);
                updateThreadName();
            }
-           
-            boolean roleChange = updateLearnerType(qv);
+
+           boolean roleChange = updateLearnerType(qv);
            boolean leaderChange = false;
            if (suggestedLeaderId != null) {
                // zxid should be non-null too
@@ -1875,7 +1875,9 @@ public class QuorumPeer extends ZooKeeperThread implements QuorumStats.Provider
     }
 
     private void updateThreadName() {
-        String plain = cnxnFactory != null ? cnxnFactory.getLocalAddress().toString() : "disabled";
+        String plain = cnxnFactory != null ?
+                cnxnFactory.getLocalAddress() != null ?
+                        cnxnFactory.getLocalAddress().toString() : "disabled" : "disabled";
         String secure = secureCnxnFactory != null ? secureCnxnFactory.getLocalAddress().toString() : "disabled";
         setName(String.format("QuorumPeer[myid=%d](plain=%s)(secure=%s)", getId(), plain, secure));
     }

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

@@ -30,6 +30,7 @@ import org.junit.runners.Suite;
         ClientTest.class,
         FourLetterWordsTest.class,
         NullDataTest.class,
+        ReconfigTest.class,
         SessionTest.class,
         WatcherTest.class
 })

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

@@ -30,6 +30,7 @@ import org.junit.runners.Suite;
         ClientTest.class,
         FourLetterWordsTest.class,
         NullDataTest.class,
+        ReconfigTest.class,
         SessionTest.class,
         WatcherTest.class
         })

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

@@ -18,13 +18,18 @@
 
 package org.apache.zookeeper.test;
 
+import static java.net.InetAddress.getLoopbackAddress;
+
 import java.io.IOException;
 import java.net.InetAddress;
+import java.net.InetSocketAddress;
+import java.net.ServerSocket;
 import java.net.UnknownHostException;
 import java.util.ArrayList;
 import java.util.LinkedList;
 import java.util.List;
 
+import org.apache.zookeeper.AsyncCallback.DataCallback;
 import org.apache.zookeeper.CreateMode;
 import org.apache.zookeeper.KeeperException;
 import org.apache.zookeeper.PortAssignment;
@@ -33,18 +38,16 @@ import org.apache.zookeeper.Watcher;
 import org.apache.zookeeper.ZKTestCase;
 import org.apache.zookeeper.ZooDefs;
 import org.apache.zookeeper.ZooKeeper;
-import org.apache.zookeeper.AsyncCallback.DataCallback;
 import org.apache.zookeeper.data.Stat;
 import org.apache.zookeeper.jmx.CommonNames;
 import org.apache.zookeeper.server.quorum.QuorumPeer;
-import org.apache.zookeeper.server.quorum.QuorumStats;
 import org.apache.zookeeper.server.quorum.QuorumPeer.QuorumServer;
 import org.apache.zookeeper.server.quorum.QuorumPeer.ServerState;
 import org.apache.zookeeper.server.quorum.flexible.QuorumHierarchical;
 import org.apache.zookeeper.server.quorum.flexible.QuorumMaj;
 import org.apache.zookeeper.server.quorum.flexible.QuorumVerifier;
-import org.junit.Assert;
 import org.junit.After;
+import org.junit.Assert;
 import org.junit.Test;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -650,6 +653,89 @@ public class ReconfigTest extends ZKTestCase implements DataCallback{
         closeAllHandles(zkArr);
     }
 
+    @Test
+    public void testPortChangeToBlockedPortFollower() throws Exception {
+        testPortChangeToBlockedPort(false);
+    }
+    @Test
+    public void testPortChangeToBlockedPortLeader() throws Exception {
+        testPortChangeToBlockedPort(true);
+    }
+
+    private void testPortChangeToBlockedPort(boolean testLeader) throws Exception {
+        qu = new QuorumUtil(1); // create 3 servers
+        qu.disableJMXTest = true;
+        qu.startAll();
+        ZooKeeper[] zkArr = createHandles(qu);
+
+        List<String> joiningServers = new ArrayList<String>();
+
+        int leaderIndex = getLeaderId(qu);
+        int followerIndex = leaderIndex == 1 ? 2 : 1;
+        int serverIndex = testLeader ? leaderIndex : followerIndex;
+        int reconfigIndex = testLeader ? followerIndex : leaderIndex;
+
+        // modify server's client port
+        int quorumPort = qu.getPeer(serverIndex).peer.getQuorumAddress().getPort();
+        int electionPort = qu.getPeer(serverIndex).peer.getElectionAddress().getPort();
+        int oldClientPort = qu.getPeer(serverIndex).peer.getClientPort();
+        int newClientPort = PortAssignment.unique();
+
+        try(ServerSocket ss = new ServerSocket()) {
+            ss.bind(new InetSocketAddress(getLoopbackAddress(), newClientPort));
+
+            joiningServers.add("server." + serverIndex + "=localhost:" + quorumPort
+                        + ":" + electionPort + ":participant;localhost:" + newClientPort);
+
+            // create a /test znode and check that read/write works before
+            // any reconfig is invoked
+            testNormalOperation(zkArr[followerIndex], zkArr[leaderIndex]);
+
+            // Reconfigure
+            reconfig(zkArr[reconfigIndex], joiningServers, null, null, -1);
+            Thread.sleep(1000);
+
+            // The follower reconfiguration will have failed
+            zkArr[serverIndex].close();
+            zkArr[serverIndex] = new ZooKeeper("127.0.0.1:"
+                    + newClientPort,
+                    ClientBase.CONNECTION_TIMEOUT, new Watcher() {
+                        public void process(WatchedEvent event) {}});
+
+            try {
+                Thread.sleep(1000);
+                zkArr[serverIndex].setData("/test", "teststr".getBytes(), -1);
+                Assert.fail("New client connected to new client port!");
+            } catch (KeeperException.ConnectionLossException e) {
+                // Exception is expected
+            }
+
+            //The old port should be clear at this stage
+
+            try (ServerSocket ss2 = new ServerSocket()) {
+                ss2.bind(new InetSocketAddress(getLoopbackAddress(), oldClientPort));
+            }
+
+            // Move back to the old port
+            joiningServers.clear();
+            joiningServers.add("server." + serverIndex + "=localhost:" + quorumPort
+                    + ":" + electionPort + ":participant;localhost:" + oldClientPort);
+
+            reconfig(zkArr[reconfigIndex], joiningServers, null, null, -1);
+
+            zkArr[serverIndex].close();
+            zkArr[serverIndex] = new ZooKeeper("127.0.0.1:"
+                    + oldClientPort,
+                    ClientBase.CONNECTION_TIMEOUT, new Watcher() {
+                        public void process(WatchedEvent event) {}});
+
+            testNormalOperation(zkArr[followerIndex], zkArr[leaderIndex]);
+            testServerHasConfig(zkArr[serverIndex], joiningServers, null);
+            Assert.assertEquals(oldClientPort, qu.getPeer(serverIndex).peer.getClientPort());
+        }
+        closeAllHandles(zkArr);
+    }
+
     @Test
     public void testUnspecifiedClientAddress() throws Exception {
     	int[] ports = new int[3];