ZOOKEEPER-1214. QuorumPeer should unregister only its previsously registered MBeans instead of use MBeanRegistry.unregisterAll() method. (César Álvarez Núñez via michim)

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

CHANGES.txt

@@ -657,6 +657,10 @@ BUGFIXES:
   ZOOKEEPER-1791. ZooKeeper package includes unnecessary jars that are part of
   the package. (mahadev via michim)
 
+  ZOOKEEPER-1214. QuorumPeer should unregister only its previsously registered
+  MBeans instead of use MBeanRegistry.unregisterAll() method.
+  (César Álvarez Núñez via michim)
+
 IMPROVEMENTS:
 
   ZOOKEEPER-1170. Fix compiler (eclipse) warnings: unused imports,

src/java/main/org/apache/zookeeper/jmx/MBeanRegistry.java

@@ -19,7 +19,10 @@
 package org.apache.zookeeper.jmx;
 
 import java.lang.management.ManagementFactory;
+import java.util.Collection;
 import java.util.Map;
+import java.util.Set;
+import java.util.HashSet;
 import java.util.concurrent.ConcurrentHashMap;
 
 import javax.management.JMException;
@@ -42,12 +45,11 @@ public class MBeanRegistry {
     
     private static MBeanRegistry instance = new MBeanRegistry(); 
     
+    private final Object LOCK = new Object();
+    
     private Map<ZKMBeanInfo, String> mapBean2Path =
         new ConcurrentHashMap<ZKMBeanInfo, String>();
     
-    private Map<String, ZKMBeanInfo> mapName2Bean =
-        new ConcurrentHashMap<String, ZKMBeanInfo>();
-
     private MBeanServer mBeanServer;
 
     public static MBeanRegistry getInstance() {
@@ -93,9 +95,10 @@ public class MBeanRegistry {
             return;
         ObjectName oname = makeObjectName(path, bean);
         try {
-            mBeanServer.registerMBean(bean, oname);
-            mapBean2Path.put(bean, path);
-            mapName2Bean.put(bean.getName(), bean);
+            synchronized (LOCK) {
+                mBeanServer.registerMBean(bean, oname);
+                mapBean2Path.put(bean, path);
+            }
         } catch (JMException e) {
             LOG.warn("Failed to register MBean " + bean.getName());
             throw e;
@@ -107,19 +110,28 @@ public class MBeanRegistry {
      * @param path
      * @param bean
      */
-    private void unregister(String path,ZKMBeanInfo bean) throws JMException {
+    private void unregister(String path,ZKMBeanInfo bean) throws JMException  {
         if(path==null)
             return;
         if (!bean.isHidden()) {
-            try {
-                mBeanServer.unregisterMBean(makeObjectName(path, bean));
-            } catch (JMException e) {
-                LOG.warn("Failed to unregister MBean " + bean.getName());
-                throw e;
+            final ObjectName objName = makeObjectName(path, bean);
+            if (LOG.isInfoEnabled()) {
+                LOG.info("Unregister MBean [{}]", objName);
+            }
+            synchronized (LOCK) {
+               mBeanServer.unregisterMBean(objName);
             }
         }        
     }
     
+    /**
+     * @return a {@link Collection} with the {@link ZKMBeanInfo} instances not
+     *         unregistered. Mainly for testing purposes.
+     */
+    public Set<ZKMBeanInfo> getRegisteredBeans() {
+        return new HashSet<ZKMBeanInfo>(mapBean2Path.keySet());
+    }
+
     /**
      * Unregister MBean.
      * @param bean
@@ -127,29 +139,16 @@ public class MBeanRegistry {
     public void unregister(ZKMBeanInfo bean) {
         if(bean==null)
             return;
-        String path=mapBean2Path.get(bean);
+        String path = mapBean2Path.remove(bean);
         try {
             unregister(path,bean);
         } catch (JMException e) {
-            LOG.warn("Error during unregister", e);
+            LOG.warn("Error during unregister of [{}]", bean.getName(), e);
+        } catch (Throwable t) {
+            LOG.error("Unexpected exception during unregister of [{}]. It should be reviewed and fixed.", bean.getName(), t);
         }
-        mapBean2Path.remove(bean);
-        mapName2Bean.remove(bean.getName());
-    }
-    /**
-     * Unregister all currently registered MBeans
-     */
-    public void unregisterAll() {
-        for(Map.Entry<ZKMBeanInfo,String> e: mapBean2Path.entrySet()) {
-            try {
-                unregister(e.getValue(), e.getKey());
-            } catch (JMException e1) {
-                LOG.warn("Error during unregister", e1);
-            }
-        }
-        mapBean2Path.clear();
-        mapName2Bean.clear();
     }
+
     /**
      * Generate a filesystem-like path.
      * @param prefix path prefix

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

@@ -93,10 +93,11 @@ import org.slf4j.LoggerFactory;
 public class QuorumPeer extends ZooKeeperThread implements QuorumStats.Provider {
     private static final Logger LOG = LoggerFactory.getLogger(QuorumPeer.class);
 
-    QuorumBean jmxQuorumBean;
+    private QuorumBean jmxQuorumBean;
     LocalPeerBean jmxLocalPeerBean;
+    private Set<RemotePeerBean> jmxRemotePeerBean;
     LeaderElectionBean jmxLeaderElectionBean;
-    QuorumCnxManager qcm = null;
+    private QuorumCnxManager qcm;
 
     /* ZKDatabase is a top level member of quorumpeer
      * which will be used in all the zookeeperservers
@@ -584,6 +585,7 @@ public class QuorumPeer extends ZooKeeperThread implements QuorumStats.Provider
     public QuorumPeer() {
         super("QuorumPeer");
         quorumStats = new QuorumStats(this);
+        jmxRemotePeerBean = new HashSet<RemotePeerBean>();
     }
 
 
@@ -871,6 +873,7 @@ public class QuorumPeer extends ZooKeeperThread implements QuorumStats.Provider
                     }
                 } else {
                     p = new RemotePeerBean(s);
+                    jmxRemotePeerBean.add((RemotePeerBean) p);
                     try {
                         MBeanRegistry.getInstance().register(p, jmxQuorumBean);
                     } catch (Exception e) {
@@ -998,13 +1001,17 @@ public class QuorumPeer extends ZooKeeperThread implements QuorumStats.Provider
             }
         } finally {
             LOG.warn("QuorumPeer main thread exited");
-            try {
-                MBeanRegistry.getInstance().unregisterAll();
-            } catch (Exception e) {
-                LOG.warn("Failed to unregister with JMX", e);
+            MBeanRegistry instance = MBeanRegistry.getInstance();
+            instance.unregister(jmxQuorumBean);
+            instance.unregister(jmxLocalPeerBean);
+
+            for (RemotePeerBean remotePeerBean : jmxRemotePeerBean) {
+                instance.unregister(remotePeerBean);
             }
+
             jmxQuorumBean = null;
             jmxLocalPeerBean = null;
+            jmxRemotePeerBean = null;
         }
     }
 

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

@@ -306,4 +306,21 @@ public class QuorumUtil {
         shutdownAll();
         JMXEnv.tearDown();
     }
+
+    public int getLeaderServer() {
+        int index = 0;
+        for (int i = 1; i <= ALL; i++) {
+            if (getPeer(i).peer.leader != null) {
+                index = i;
+                break;
+            }
+        }
+
+        Assert.assertTrue("Leader server not found.", index > 0);
+        return index;
+    }
+
+    public String getConnectionStringForServer(final int index) {
+        return "127.0.0.1:" + getPeer(index).clientPort;
+    }
 }

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

@@ -0,0 +1,94 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or morecontributor 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.test;
+
+import java.io.IOException;
+import java.util.Set;
+
+import org.apache.zookeeper.ZKTestCase;
+import org.apache.zookeeper.jmx.MBeanRegistry;
+import org.apache.zookeeper.jmx.ZKMBeanInfo;
+import org.apache.zookeeper.server.quorum.QuorumPeer;
+import org.junit.Assert;
+import org.junit.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * This class is intented to ensure the correct functionality of
+ * {@link QuorumUtil} helper.
+ */
+public class QuorumUtilTest extends ZKTestCase {
+    private static final Logger LOG = LoggerFactory
+            .getLogger(QuorumUtilTest.class);
+
+    /**
+     * <p>
+     * This test ensures that all JXM beans associated to a {@link QuorumPeer}
+     * are unregistered when shuted down ({@link QuorumUtil#shutdown(int)}). It
+     * allows a successfull restarting of several zookeeper servers (
+     * {@link QuorumPeer}) running on the same JVM.
+     * <p>
+     * See ZOOKEEPER-1214 for details.
+     */
+    @Test
+    public void validateAllMXBeanAreUnregistered() throws IOException {
+        QuorumUtil qU = new QuorumUtil(1);
+        LOG.info(">-->> Starting up all servers...");
+        qU.startAll();
+        LOG.info(">-->> Servers up and running...");
+
+        int leaderIndex = qU.getLeaderServer();
+        int firstFollowerIndex = 0;
+        int secondFollowerIndex = 0;
+
+        switch (leaderIndex) {
+        case 1:
+            firstFollowerIndex = 2;
+            secondFollowerIndex = 3;
+            break;
+        case 2:
+            firstFollowerIndex = 1;
+            secondFollowerIndex = 3;
+            break;
+        case 3:
+            firstFollowerIndex = 1;
+            secondFollowerIndex = 2;
+            break;
+
+        default:
+            Assert.fail("Unexpected leaderIndex value: " + leaderIndex);
+            break;
+        }
+
+        LOG.info(">-->> Shuting down server [{}]", firstFollowerIndex);
+        qU.shutdown(firstFollowerIndex);
+        LOG.info(">-->> Shuting down server [{}]", secondFollowerIndex);
+        qU.shutdown(secondFollowerIndex);
+        LOG.info(">-->> Restarting server [{}]", firstFollowerIndex);
+        qU.restart(firstFollowerIndex);
+        LOG.info(">-->> Restarting server [{}]", secondFollowerIndex);
+        qU.restart(secondFollowerIndex);
+
+        qU.shutdownAll();
+        Set<ZKMBeanInfo> pending = MBeanRegistry.getInstance()
+                .getRegisteredBeans();
+        Assert.assertTrue("The following beans should have been unregistered: "
+                + pending, pending.isEmpty());
+    }
+}