ZOOKEEPER-696. NPE in the hudson logs, seems nioservercnxn closed twice (phunt via mahadev)

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

CHANGES.txt

@@ -264,6 +264,9 @@ BUGFIXES:
   ZOOKEEPER-693. TestObserver stuck in tight notification loop in FLE
   (flavio via phunt)
 
+  ZOOKEEPER-696. NPE in the hudson logs, seems nioservercnxn closed twice 
+  (phunt via mahadev)
+
 IMPROVEMENTS:
   ZOOKEEPER-473. cleanup junit tests to eliminate false positives due to
   "socket reuse" and failure to close client (phunt via mahadev)

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

@@ -41,30 +41,44 @@ public class ConnectionBean implements ConnectionMXBean, ZKMBeanInfo {
     private final CnxnStats stats;
 
     private final ZooKeeperServer zk;
+    
+    private final String remoteIP;
+    private final long sessionId;
 
     public ConnectionBean(ServerCnxn connection,ZooKeeperServer zk){
         this.connection = connection;
         this.stats = (CnxnStats)connection.getStats();
         this.zk = zk;
+        
+        InetSocketAddress sockAddr = connection.getRemoteAddress();
+        if (sockAddr == null) {
+            remoteIP = "Unknown";
+        } else {
+            InetAddress addr = sockAddr.getAddress();
+            if (addr instanceof Inet6Address) {
+                remoteIP = ObjectName.quote(addr.getHostAddress());
+            } else {
+                remoteIP = addr.getHostAddress();
+            }
+        }
+        sessionId = connection.getSessionId();
     }
     
     public String getSessionId() {
-        return "0x" + Long.toHexString(connection.getSessionId());
+        return "0x" + Long.toHexString(sessionId);
     }
 
     public String getSourceIP() {
         InetSocketAddress sockAddr = connection.getRemoteAddress();
+        if (sockAddr == null) {
+            return null;
+        }
         return sockAddr.getAddress().getHostAddress()
             + ":" + sockAddr.getPort();
     }
 
     public String getName() {
-        InetAddress addr = connection.getRemoteAddress().getAddress();
-        String ip = addr.getHostAddress();
-        if (addr instanceof Inet6Address) {
-            ip = ObjectName.quote(ip);
-        }
-        return MBeanRegistry.getInstance().makeFullPath("Connections", ip,
+        return MBeanRegistry.getInstance().makeFullPath("Connections", remoteIP,
                 getSessionId());
     }
     
@@ -74,7 +88,7 @@ public class ConnectionBean implements ConnectionMXBean, ZKMBeanInfo {
     
     public String[] getEphemeralNodes() {
         if(zk.getZKDatabase()  !=null){
-            String[] res= zk.getZKDatabase().getEphemerals(connection.getSessionId())
+            String[] res= zk.getZKDatabase().getEphemerals(sessionId)
                 .toArray(new String[0]);
             Arrays.sort(res);
             return res;
@@ -88,7 +102,7 @@ public class ConnectionBean implements ConnectionMXBean, ZKMBeanInfo {
     
     public void terminateSession() {
         try {
-            zk.closeSession(connection.getSessionId());
+            zk.closeSession(sessionId);
         } catch (Exception e) {
             LOG.warn("Unable to closeSession() for session: 0x" 
                     + getSessionId(), e);

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

@@ -276,26 +276,29 @@ public class NIOServerCnxn implements Watcher, ServerCnxn {
         }
 
         /**
-         * clear all the connections in the selector
+         * Clear all the connections in the selector.
+         * 
+         * You must first close ss (the serversocketchannel) if you wish
+         * to block any new connections from being established.
          *
          */
+        @SuppressWarnings("unchecked")
         synchronized public void clear() {
             selector.wakeup();
-            synchronized (cnxns) {
-                // got to clear all the connections that we have in the selector
-                for (Iterator<NIOServerCnxn> it = cnxns.iterator(); it
-                        .hasNext();) {
-                    NIOServerCnxn cnxn = it.next();
-                    it.remove();
-                    try {
-                        cnxn.close();
-                    } catch (Exception e) {
-                        LOG.warn("Ignoring exception closing cnxn sessionid 0x"
-                                + Long.toHexString(cnxn.sessionId), e);
-                    }
+            HashSet<NIOServerCnxn> cnxns;
+            synchronized (this.cnxns) {
+                cnxns = (HashSet<NIOServerCnxn>)this.cnxns.clone();
+            }
+            // got to clear all the connections that we have in the selector
+            for (NIOServerCnxn cnxn: cnxns) {
+                try {
+                    // don't hold this.cnxns lock as deadlock may occur
+                    cnxn.close();
+                } catch (Exception e) {
+                    LOG.warn("Ignoring exception closing cnxn sessionid 0x"
+                            + Long.toHexString(cnxn.sessionId), e);
                 }
             }
-
         }
 
         public void shutdown() {
@@ -324,21 +327,21 @@ public class NIOServerCnxn implements Watcher, ServerCnxn {
             closeSessionWithoutWakeup(sessionId);
         }
 
-
+        @SuppressWarnings("unchecked")
         private void closeSessionWithoutWakeup(long sessionId) {
-            synchronized (cnxns) {
-                for (Iterator<NIOServerCnxn> it = cnxns.iterator(); it
-                        .hasNext();) {
-                    NIOServerCnxn cnxn = it.next();
-                    if (cnxn.sessionId == sessionId) {
-                        it.remove();
-                        try {
-                            cnxn.close();
-                        } catch (Exception e) {
-                            LOG.warn("exception during session close", e);
-                        }
-                        break;
+            HashSet<NIOServerCnxn> cnxns;
+            synchronized (this.cnxns) {
+                cnxns = (HashSet<NIOServerCnxn>)this.cnxns.clone();
+            }
+
+            for (NIOServerCnxn cnxn : cnxns) {
+                if (cnxn.sessionId == sessionId) {
+                    try {
+                        cnxn.close();
+                    } catch (Exception e) {
+                        LOG.warn("exception during session close", e);
                     }
+                    break;
                 }
             }
         }
@@ -1202,31 +1205,58 @@ public class NIOServerCnxn implements Watcher, ServerCnxn {
     }
 
     /*
-     * (non-Javadoc)
-     *
-     * @see org.apache.zookeeper.server.ServerCnxnIface#close()
+     * Close the cnxn and remove it from the factory cnxns list.
+     * 
+     * This function returns immediately if the cnxn is not on the cnxns list.
      */
     public void close() {
-        // unregister from JMX
-        try {
-            if(jmxConnectionBean != null){
-                MBeanRegistry.getInstance().unregister(jmxConnectionBean);
+        synchronized(factory.cnxns){
+            // if this is not in cnxns then it's already closed
+            if (!factory.cnxns.remove(this)) {
+                return;
             }
-        } catch (Exception e) {
-            LOG.warn("Failed to unregister with JMX", e);
-        }
-        jmxConnectionBean = null;
 
-        synchronized (factory.ipMap)
-        {
-            Set<NIOServerCnxn> s = factory.ipMap.get(sock.socket().getInetAddress());
-            s.remove(this);
-        }
-        synchronized (factory.cnxns) {
-            factory.cnxns.remove(this);
+            synchronized (factory.ipMap) {
+                Set<NIOServerCnxn> s =
+                    factory.ipMap.get(sock.socket().getInetAddress());
+                s.remove(this);
+            }
+
+            // unregister from JMX
+            try {
+                if(jmxConnectionBean != null){
+                    MBeanRegistry.getInstance().unregister(jmxConnectionBean);
+                }
+            } catch (Exception e) {
+                LOG.warn("Failed to unregister with JMX", e);
+            }
+            jmxConnectionBean = null;
+    
+            if (zk != null) {
+                zk.removeCnxn(this);
+            }
+    
+            closeSock();
+    
+            if (sk != null) {
+                try {
+                    // need to cancel this selection key from the selector
+                    sk.cancel();
+                } catch (Exception e) {
+                    if (LOG.isDebugEnabled()) {
+                        LOG.debug("ignoring exception during selectionkey cancel", e);
+                    }
+                }
+            }
         }
-        if (zk != null) {
-            zk.removeCnxn(this);
+    }
+
+    /**
+     * Close resources associated with the sock of this cnxn. 
+     */
+    private void closeSock() {
+        if (sock == null) {
+            return;
         }
 
         LOG.info("Closed socket connection for client "
@@ -1275,18 +1305,8 @@ public class NIOServerCnxn implements Watcher, ServerCnxn {
             }
         }
         sock = null;
-        if (sk != null) {
-            try {
-                // need to cancel this selection key from the selector
-                sk.cancel();
-            } catch (Exception e) {
-                if (LOG.isDebugEnabled()) {
-                    LOG.debug("ignoring exception during selectionkey cancel", e);
-                }
-            }
-        }
     }
-
+    
     private final static byte fourBytes[] = new byte[4];
 
     /*
@@ -1419,7 +1439,10 @@ public class NIOServerCnxn implements Watcher, ServerCnxn {
         return authInfo;
     }
 
-    public InetSocketAddress getRemoteAddress() {
+    public synchronized InetSocketAddress getRemoteAddress() {
+        if (sock == null) {
+            return null;
+        }
         return (InetSocketAddress) sock.socket().getRemoteSocketAddress();
     }
 

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

@@ -396,17 +396,17 @@ public class Leader {
         
         // NIO should not accept conenctions
         self.cnxnFactory.setZooKeeperServer(null);
+        try {
+            ss.close();
+        } catch (IOException e) {
+            LOG.warn("Ignoring unexpected exception during close",e);
+        }
         // clear all the connections
         self.cnxnFactory.clear();
         // shutdown the previous zk
         if (zk != null) {
             zk.shutdown();
         }
-        try {
-            ss.close();
-        } catch (IOException e) {
-            LOG.warn("Ignoring unexpected exception during close",e);
-        }
         synchronized (learners) {
             for (Iterator<LearnerHandler> it = learners.iterator(); it
                     .hasNext();) {

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

@@ -327,7 +327,7 @@ public class SessionTest extends TestCase implements Watcher {
     public void process(WatchedEvent event) {
         LOG.info("Event:" + event.getState() + " " + event.getType() + " " + event.getPath());
         if (event.getState() == KeeperState.SyncConnected
-                && startSignal.getCount() > 0)
+                && startSignal != null && startSignal.getCount() > 0)
         {
             startSignal.countDown();
         }