ZOOKEEPER-3123: MetricsProvider Lifecycle in ZooKeeper Server

Manage the lifecycle of a MetricsProvider inside a ZooKeeper server.
- handle configuration
- start and configure the MetricsProvider
- notify shutdown to the MetricsProvider

This is an early preview, because there are some points to discuss:
- We have to throw an IOException in case of failure (in order not to change the current signature of main methods used to start the server)
- The patch only provides the lifecycle, it introduces some dead fields (root metrics context), this is expected as the real instrumentation will be done in a further step, is it okay ?
- Test cases cover only standalone mode, do we need to add a new suite for testing configuration and boot errors on QuorumPeer mode ? (the answer should be YES)
- MetricsProvider configuration is not subject to dynamic 'reconfig'

Configuration to the MetricsProvider is not yet handled, the idea is to let the user configure properties like
metricsProvider.className=o.a.z.metrics.prometheus.PrometheusMetricsProvider
metricsProvider.customParam1=value1
metricsProvider.customParam2=value2

in this case the MetricsProvider will receive {customParam1=value1, customParam2=value2} as parameter in configure()

is it okay ?

Author: Enrico Olivelli <eolivelli@apache.org>

Reviewers: fangmin@apache.org, andor@apache.org

Closes #601 from eolivelli/fix/boot-provider and squashes the following commits:

8964ed17 [Enrico Olivelli] Fix tests, use getters in order to support Mock QuorumPeerConfig
93749f6a [Enrico Olivelli] fix imports
7ad552db [Enrico Olivelli] Add testcases around QuorumPeerMain
22f79eb8 [Enrico Olivelli] clean up
c92450e4 [Enrico Olivelli] implement MetricsProvider configuration, fix some review comments
f4f66ecb [Enrico Olivelli] ZOOKEEPER-3123 MetricsProvider Lifecycle in ZooKeeper Server

src/java/main/org/apache/zookeeper/metrics/impl/MetricsProviderBootstrap.java

@@ -0,0 +1,50 @@
+/**
+ * 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.metrics.impl;
+
+import java.util.Properties;
+import org.apache.zookeeper.metrics.MetricsProvider;
+import org.apache.zookeeper.metrics.MetricsProviderLifeCycleException;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Utility for bootstrap process of MetricsProviders
+ */
+public abstract class MetricsProviderBootstrap {
+
+    private static final Logger LOG = LoggerFactory.getLogger(MetricsProviderBootstrap.class);
+
+    public static MetricsProvider startMetricsProvider(String metricsProviderClassName, Properties configuration)
+            throws MetricsProviderLifeCycleException {
+        try {
+            MetricsProvider metricsProvider = (MetricsProvider) Class.forName(metricsProviderClassName,
+                    true, Thread.currentThread().getContextClassLoader()).newInstance();
+            metricsProvider.configure(configuration);
+            metricsProvider.start();
+            return metricsProvider;
+        } catch (ClassNotFoundException | IllegalAccessException | InstantiationException error) {
+            LOG.error("Cannot boot MetricsProvider {}", metricsProviderClassName, error);
+            throw new MetricsProviderLifeCycleException("Cannot boot MetricsProvider " + metricsProviderClassName,
+                    error);
+        } catch (MetricsProviderLifeCycleException error) {
+            LOG.error("Cannot boot MetricsProvider {}", metricsProviderClassName, error);
+            throw error;
+        }
+    }
+}

src/java/main/org/apache/zookeeper/metrics/impl/NullMetricsProvider.java

@@ -0,0 +1,100 @@
+/**
+ * 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.metrics.impl;
+
+import java.util.Properties;
+import org.apache.zookeeper.metrics.Counter;
+import org.apache.zookeeper.metrics.Gauge;
+import org.apache.zookeeper.metrics.MetricsContext;
+import org.apache.zookeeper.metrics.MetricsProvider;
+import org.apache.zookeeper.metrics.MetricsProviderLifeCycleException;
+import org.apache.zookeeper.metrics.Summary;
+
+/**
+ * This is a dummy MetricsProvider which does nothing.
+ */
+public class NullMetricsProvider implements MetricsProvider {
+
+    @Override
+    public void configure(Properties configuration) throws MetricsProviderLifeCycleException {
+    }
+
+    @Override
+    public void start() throws MetricsProviderLifeCycleException {
+    }
+
+    @Override
+    public MetricsContext getRootContext() {
+        return NullMetricsContext.INSTANCE;
+    }
+
+    @Override
+    public void stop() {
+    }
+
+    public static final class NullMetricsContext implements MetricsContext {
+
+        public static final NullMetricsContext INSTANCE = new NullMetricsContext();
+
+        @Override
+        public MetricsContext getContext(String name) {
+            return INSTANCE;
+        }
+
+        @Override
+        public Counter getCounter(String name) {
+            return NullCounter.INSTANCE;
+        }
+
+        @Override
+        public boolean registerGauge(String name, Gauge gauge) {
+            return true;
+        }
+
+        @Override
+        public Summary getSummary(String name) {
+            return NullSummary.INSTANCE;
+        }
+
+    }
+
+    private static final class NullCounter implements Counter {
+
+        private static final NullCounter INSTANCE = new NullCounter();
+
+        @Override
+        public void inc(long delta) {
+        }
+
+        @Override
+        public long get() {
+            return 0;
+        }
+
+    }
+
+    private static final class NullSummary implements Summary {
+
+        private static final NullSummary INSTANCE = new NullSummary();
+
+        @Override
+        public void registerValue(long value) {
+        }
+
+    }
+}

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

@@ -21,8 +21,10 @@ package org.apache.zookeeper.server;
 import java.io.File;
 import java.net.InetSocketAddress;
 import java.util.Arrays;
+import java.util.Properties;
 
 import org.apache.yetus.audience.InterfaceAudience;
+import org.apache.zookeeper.metrics.impl.NullMetricsProvider;
 import org.apache.zookeeper.server.quorum.QuorumPeerConfig;
 import org.apache.zookeeper.server.quorum.QuorumPeerConfig.ConfigException;
 
@@ -48,6 +50,8 @@ public class ServerConfig {
     protected int minSessionTimeout = -1;
     /** defaults to -1 if not set explicitly */
     protected int maxSessionTimeout = -1;
+    protected String metricsProviderClassName = NullMetricsProvider.class.getName();
+    protected Properties metricsProviderConfiguration = new Properties();
 
     /**
      * Parse arguments for server configuration
@@ -99,6 +103,8 @@ public class ServerConfig {
         maxClientCnxns = config.getMaxClientCnxns();
         minSessionTimeout = config.getMinSessionTimeout();
         maxSessionTimeout = config.getMaxSessionTimeout();
+        metricsProviderClassName = config.getMetricsProviderClassName();
+        metricsProviderConfiguration = config.getMetricsProviderConfiguration();
     }
 
     public InetSocketAddress getClientPortAddress() {
@@ -115,4 +121,7 @@ public class ServerConfig {
     public int getMinSessionTimeout() { return minSessionTimeout; }
     /** maximum session timeout in milliseconds, -1 if unset */
     public int getMaxSessionTimeout() { return maxSessionTimeout; }
+    public String getMetricsProviderClassName() { return metricsProviderClassName; }
+    public Properties getMetricsProviderConfiguration() { return metricsProviderConfiguration; }
+
 }

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

@@ -52,6 +52,8 @@ import org.apache.zookeeper.data.ACL;
 import org.apache.zookeeper.data.Id;
 import org.apache.zookeeper.data.StatPersisted;
 import org.apache.zookeeper.jmx.MBeanRegistry;
+import org.apache.zookeeper.metrics.MetricsContext;
+import org.apache.zookeeper.metrics.impl.NullMetricsProvider;
 import org.apache.zookeeper.proto.AuthPacket;
 import org.apache.zookeeper.proto.ConnectRequest;
 import org.apache.zookeeper.proto.ConnectResponse;
@@ -127,6 +129,7 @@ public class ZooKeeperServer implements SessionExpirer, ServerStats.Provider {
 
     private final ServerStats serverStats;
     private final ZooKeeperServerListener listener;
+    private MetricsContext rootMetricsContext = NullMetricsProvider.NullMetricsContext.INSTANCE;
     private ZooKeeperServerShutdownHandler zkShutdownHandler;
     private volatile int createSessionTrackerServerId = 1;
 
@@ -877,6 +880,14 @@ public class ZooKeeperServer implements SessionExpirer, ServerStats.Provider {
         secureServerCnxnFactory = factory;
     }
 
+    public MetricsContext getRootMetricsContext() {
+        return rootMetricsContext;
+    }
+
+    public void setRootMetricsContext(MetricsContext rootMetricsContext) {
+        this.rootMetricsContext = rootMetricsContext;
+    }
+
     /**
      * return the last proceesed id from the
      * datatree

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

@@ -19,6 +19,7 @@
 package org.apache.zookeeper.server;
 
 import java.io.IOException;
+import java.util.Properties;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.TimeUnit;
 
@@ -26,6 +27,9 @@ import javax.management.JMException;
 
 import org.apache.yetus.audience.InterfaceAudience;
 import org.apache.zookeeper.jmx.ManagedUtil;
+import org.apache.zookeeper.metrics.MetricsProvider;
+import org.apache.zookeeper.metrics.MetricsProviderLifeCycleException;
+import org.apache.zookeeper.metrics.impl.MetricsProviderBootstrap;
 import org.apache.zookeeper.server.admin.AdminServer;
 import org.apache.zookeeper.server.admin.AdminServer.AdminServerException;
 import org.apache.zookeeper.server.admin.AdminServerFactory;
@@ -50,7 +54,7 @@ public class ZooKeeperServerMain {
     private ServerCnxnFactory cnxnFactory;
     private ServerCnxnFactory secureCnxnFactory;
     private ContainerManager containerManager;
-
+    private MetricsProvider metricsProvider;
     private AdminServer adminServer;
 
     /*
@@ -117,6 +121,15 @@ public class ZooKeeperServerMain {
         LOG.info("Starting server");
         FileTxnSnapLog txnLog = null;
         try {
+            try {
+                metricsProvider = MetricsProviderBootstrap
+                        .startMetricsProvider(config.getMetricsProviderClassName(),
+                                              config.getMetricsProviderConfiguration());
+            } catch (MetricsProviderLifeCycleException error) {
+                throw new IOException("Cannot boot MetricsProvider "+config.getMetricsProviderClassName(),
+                    error);
+            }
+
             // Note that this thread isn't going to be doing anything else,
             // so rather than spawning another thread, we will just call
             // run() in this thread.
@@ -124,6 +137,7 @@ public class ZooKeeperServerMain {
             txnLog = new FileTxnSnapLog(config.dataLogDir, config.dataDir);
             final ZooKeeperServer zkServer = new ZooKeeperServer(txnLog,
                     config.tickTime, config.minSessionTimeout, config.maxSessionTimeout, null);
+            zkServer.setRootMetricsContext(metricsProvider.getRootContext());
             txnLog.setServerStats(zkServer.serverStats());
 
             // Registers shutdown handler which will be used to know the
@@ -179,6 +193,13 @@ public class ZooKeeperServerMain {
             if (txnLog != null) {
                 txnLog.close();
             }
+            if (metricsProvider != null) {
+                try {
+                    metricsProvider.stop();
+                } catch (Throwable error) {
+                    LOG.warn("Error while stopping metrics", error);
+                }
+            }
         }
     }
 

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

@@ -74,6 +74,8 @@ import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 import static org.apache.zookeeper.common.NetUtils.formatInetAddr;
+import org.apache.zookeeper.metrics.MetricsContext;
+import org.apache.zookeeper.metrics.impl.NullMetricsProvider;
 
 /**
  * This class manages the quorum protocol. There are three states this server
@@ -773,6 +775,8 @@ public class QuorumPeer extends ZooKeeperThread implements QuorumStats.Provider
 
     AdminServer adminServer;
 
+    private MetricsContext rootMetricsContext = NullMetricsProvider.NullMetricsContext.INSTANCE;
+
     public static QuorumPeer testingQuorumPeer() throws SaslException {
         return new QuorumPeer();
     }
@@ -1679,6 +1683,10 @@ public class QuorumPeer extends ZooKeeperThread implements QuorumStats.Provider
         this.secureCnxnFactory = secureCnxnFactory;
     }
 
+    public void setRootMetricsContext(MetricsContext rootMetricsContext) {
+        this.rootMetricsContext = rootMetricsContext;
+    }
+
     private void startServerCnxnFactory() {
         if (cnxnFactory != null) {
             cnxnFactory.start();

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

@@ -54,9 +54,9 @@ 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.apache.zookeeper.server.util.VerifyingFileFactory;
-import org.apache.zookeeper.server.util.ConfigUtils;
 
 import static org.apache.zookeeper.common.NetUtils.formatInetAddr;
+import org.apache.zookeeper.metrics.impl.NullMetricsProvider;
 
 @InterfaceAudience.Public
 public class QuorumPeerConfig {
@@ -79,6 +79,8 @@ public class QuorumPeerConfig {
     protected int minSessionTimeout = -1;
     /** defaults to -1 if not set explicitly */
     protected int maxSessionTimeout = -1;
+    protected String metricsProviderClassName = NullMetricsProvider.class.getName();
+    protected Properties metricsProviderConfiguration = new Properties();
     protected boolean localSessionsEnabled = false;
     protected boolean localSessionsUpgradingEnabled = false;
 
@@ -325,6 +327,11 @@ public class QuorumPeerConfig {
                 quorumServicePrincipal = value;
             } else if (key.equals("quorum.cnxn.threads.size")) {
                 quorumCnxnThreadsSize = Integer.parseInt(value);
+            } else if (key.equals("metricsProvider.className")) {
+                metricsProviderClassName = value;
+            } else if (key.startsWith("metricsProvider.")) {
+                String keyForMetricsProvider = key.substring(16);
+                metricsProviderConfiguration.put(keyForMetricsProvider, value);
             } else {
                 System.setProperty("zookeeper." + key, value);
             }
@@ -409,7 +416,14 @@ public class QuorumPeerConfig {
         if (minSessionTimeout > maxSessionTimeout) {
             throw new IllegalArgumentException(
                     "minSessionTimeout must not be larger than maxSessionTimeout");
-        }          
+        }
+
+        LOG.info("metricsProvider.className is {}", metricsProviderClassName);
+        try {
+            Class.forName(metricsProviderClassName, false, Thread.currentThread().getContextClassLoader());
+        } catch (ClassNotFoundException error) {
+            throw new IllegalArgumentException("metrics provider class was not found", error);
+        }
 
         // backward compatibility - dynamic configuration in the same file as
         // static configuration params see writeDynamicConfig()
@@ -735,6 +749,8 @@ public class QuorumPeerConfig {
     public int getMaxClientCnxns() { return maxClientCnxns; }
     public int getMinSessionTimeout() { return minSessionTimeout; }
     public int getMaxSessionTimeout() { return maxSessionTimeout; }
+    public String getMetricsProviderClassName() { return metricsProviderClassName; }
+    public Properties getMetricsProviderConfiguration() { return metricsProviderConfiguration; }
     public boolean areLocalSessionsEnabled() { return localSessionsEnabled; }
     public boolean isLocalSessionsUpgradingEnabled() {
         return localSessionsUpgradingEnabled;

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

@@ -18,6 +18,7 @@
 package org.apache.zookeeper.server.quorum;
 
 import java.io.IOException;
+import java.util.Properties;
 
 import javax.management.JMException;
 import javax.security.sasl.SaslException;
@@ -26,6 +27,9 @@ import org.apache.yetus.audience.InterfaceAudience;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.apache.zookeeper.jmx.ManagedUtil;
+import org.apache.zookeeper.metrics.MetricsProvider;
+import org.apache.zookeeper.metrics.MetricsProviderLifeCycleException;
+import org.apache.zookeeper.metrics.impl.MetricsProviderBootstrap;
 import org.apache.zookeeper.server.ExitCode;
 import org.apache.zookeeper.server.ServerCnxnFactory;
 import org.apache.zookeeper.server.ZKDatabase;
@@ -140,7 +144,17 @@ public class QuorumPeerMain {
       }
 
       LOG.info("Starting quorum peer");
+      MetricsProvider metricsProvider;
       try {
+        metricsProvider = MetricsProviderBootstrap
+                      .startMetricsProvider(config.getMetricsProviderClassName(),
+                                            config.getMetricsProviderConfiguration());
+      } catch (MetricsProviderLifeCycleException error) {
+        throw new IOException("Cannot boot MetricsProvider " + config.getMetricsProviderClassName(),
+                      error);
+      }
+      try {
+
           ServerCnxnFactory cnxnFactory = null;
           ServerCnxnFactory secureCnxnFactory = null;
 
@@ -159,6 +173,7 @@ public class QuorumPeerMain {
           }
 
           quorumPeer = getQuorumPeer();
+          quorumPeer.setRootMetricsContext(metricsProvider.getRootContext());
           quorumPeer.setTxnFactory(new FileTxnSnapLog(
                       config.getDataLogDir(),
                       config.getDataDir()));
@@ -203,6 +218,14 @@ public class QuorumPeerMain {
       } catch (InterruptedException e) {
           // warn, but generally this is ok
           LOG.warn("Quorum Peer interrupted", e);
+      } finally {
+          if (metricsProvider != null) {
+              try {
+                  metricsProvider.stop();
+              } catch (Throwable error) {
+                  LOG.warn("Error while stopping metrics", error);
+              }
+          }
       }
     }
 

src/java/test/org/apache/zookeeper/metrics/BaseTestMetricsProvider.java

@@ -0,0 +1,137 @@
+/**
+ * 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.metrics;
+
+import java.util.Properties;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
+import org.apache.zookeeper.metrics.impl.NullMetricsProvider;
+
+/**
+ * Simple MetricsProvider for tests.
+ */
+public abstract class BaseTestMetricsProvider implements MetricsProvider {
+
+    @Override
+    public void configure(Properties prprts) throws MetricsProviderLifeCycleException {
+    }
+
+    @Override
+    public void start() throws MetricsProviderLifeCycleException {
+    }
+
+    @Override
+    public MetricsContext getRootContext() {
+        return NullMetricsProvider.NullMetricsContext.INSTANCE;
+    }
+
+    @Override
+    public void stop() {
+    }
+
+    public static final class MetricsProviderCapturingLifecycle extends BaseTestMetricsProvider {
+
+        public static final AtomicBoolean configureCalled = new AtomicBoolean();
+        public static final AtomicBoolean startCalled = new AtomicBoolean();
+        public static final AtomicBoolean stopCalled = new AtomicBoolean();
+        public static final AtomicBoolean getRootContextCalled = new AtomicBoolean();
+
+        public static void reset() {
+            configureCalled.set(false);
+            startCalled.set(false);
+            stopCalled.set(false);
+            getRootContextCalled.set(false);
+        }
+        
+        @Override
+        public void configure(Properties prprts) throws MetricsProviderLifeCycleException {
+            if (!configureCalled.compareAndSet(false, true)) {
+                // called twice
+                throw new IllegalStateException();
+            }
+        }
+
+        @Override
+        public void start() throws MetricsProviderLifeCycleException {
+            if (!startCalled.compareAndSet(false, true)) {
+                // called twice
+                throw new IllegalStateException();
+            }
+        }
+
+        @Override
+        public MetricsContext getRootContext() {
+            if (!getRootContextCalled.compareAndSet(false, true)) {
+                // called twice
+                throw new IllegalStateException();
+            }
+            return NullMetricsProvider.NullMetricsContext.INSTANCE;
+        }
+
+        @Override
+        public void stop() {
+            if (!stopCalled.compareAndSet(false, true)) {
+                // called twice
+                throw new IllegalStateException();
+            }
+        }
+
+    }
+
+    public static final class MetricsProviderWithErrorInStart extends BaseTestMetricsProvider {
+
+        @Override
+        public void start() throws MetricsProviderLifeCycleException {
+            throw new MetricsProviderLifeCycleException();
+        }
+
+    }
+
+    public static final class MetricsProviderWithErrorInConfigure extends BaseTestMetricsProvider {
+
+        @Override
+        public void configure(Properties prprts) throws MetricsProviderLifeCycleException {
+            throw new MetricsProviderLifeCycleException();
+        }
+
+    }
+
+    public static final class MetricsProviderWithConfiguration extends BaseTestMetricsProvider {
+
+        public static final AtomicInteger httpPort = new AtomicInteger();
+
+        @Override
+        public void configure(Properties prprts) throws MetricsProviderLifeCycleException {
+            httpPort.set(Integer.parseInt(prprts.getProperty("httpPort")));
+        }
+
+    }
+
+    public static final class MetricsProviderWithErrorInStop extends BaseTestMetricsProvider {
+
+        public static final AtomicBoolean stopCalled = new AtomicBoolean();
+
+        @Override
+        public void stop() {
+            stopCalled.set(true);
+            throw new RuntimeException();
+        }
+
+    }
+
+}

src/java/test/org/apache/zookeeper/server/ZooKeeperServerMainTest.java

@@ -39,6 +39,12 @@ import org.apache.zookeeper.ZooKeeper;
 import org.apache.zookeeper.Watcher.Event.KeeperState;
 import org.apache.zookeeper.ZooDefs.Ids;
 import org.apache.zookeeper.common.PathUtils;
+import org.apache.zookeeper.metrics.BaseTestMetricsProvider;
+import org.apache.zookeeper.metrics.BaseTestMetricsProvider.MetricsProviderCapturingLifecycle;
+import org.apache.zookeeper.metrics.BaseTestMetricsProvider.MetricsProviderWithConfiguration;
+import org.apache.zookeeper.metrics.BaseTestMetricsProvider.MetricsProviderWithErrorInConfigure;
+import org.apache.zookeeper.metrics.BaseTestMetricsProvider.MetricsProviderWithErrorInStart;
+import org.apache.zookeeper.metrics.BaseTestMetricsProvider.MetricsProviderWithErrorInStop;
 import org.apache.zookeeper.server.persistence.FileTxnSnapLog;
 import org.apache.zookeeper.server.quorum.QuorumPeerConfig.ConfigException;
 import org.apache.zookeeper.test.ClientBase;
@@ -346,6 +352,197 @@ public class ZooKeeperServerMainTest extends ZKTestCase implements Watcher {
         }
     }
 
+    /**
+     * Test verifies that the server shouldn't boot with an invalid metrics provider
+     */
+    @Test
+    public void testInvalidMetricsProvider()
+            throws Exception {
+        ClientBase.setupTestEnv();
+
+        final int CLIENT_PORT = PortAssignment.unique();
+        final String configs = "metricsProvider.className=BadClass\n";
+        MainThread main = new MainThread(CLIENT_PORT, true, configs);
+        String args[] = new String[1];
+        args[0] = main.confFile.toString();
+        try {
+            main.main.initializeAndRun(args);
+            Assert.fail("Must throw exception as metrics provider is not "
+                    + "well configured");
+        } catch (ConfigException iae) {
+            // expected
+        }
+    }
+
+    /**
+     * Test verifies that the server shouldn't boot with a faulty metrics provider
+     */
+    @Test
+    public void testFaultyMetricsProviderOnStart()
+            throws Exception {
+        ClientBase.setupTestEnv();
+
+        final int CLIENT_PORT = PortAssignment.unique();
+        final String configs = "metricsProvider.className="+MetricsProviderWithErrorInStart.class.getName()+"\n";
+        MainThread main = new MainThread(CLIENT_PORT, true, configs);
+        String args[] = new String[1];
+        args[0] = main.confFile.toString();
+        try {
+            main.main.initializeAndRun(args);
+            Assert.fail("Must throw exception as metrics provider cannot boot");
+        } catch (IOException iae) {
+            // expected
+        }
+    }
+
+    /**
+     * Test verifies that the server shouldn't boot with a faulty metrics provider
+     */
+    @Test
+    public void testFaultyMetricsProviderOnConfigure()
+            throws Exception {
+        ClientBase.setupTestEnv();
+
+        final int CLIENT_PORT = PortAssignment.unique();
+        final String configs = "metricsProvider.className="+MetricsProviderWithErrorInConfigure.class.getName()+"\n";
+        MainThread main = new MainThread(CLIENT_PORT, true, configs);
+        String args[] = new String[1];
+        args[0] = main.confFile.toString();
+        try {
+            main.main.initializeAndRun(args);
+            Assert.fail("Must throw exception as metrics provider is cannot boot");
+        } catch (IOException iae) {
+            // expected
+        }
+    }
+
+    /**
+     * Test verifies that the server shouldn't be affected but runtime errors on stop()
+     */
+    @Test
+    public void testFaultyMetricsProviderOnStop()
+            throws Exception {
+        ClientBase.setupTestEnv();
+
+        final int CLIENT_PORT = PortAssignment.unique();
+        MetricsProviderWithErrorInStop.stopCalled.set(false);
+        final String configs = "metricsProvider.className="+MetricsProviderWithErrorInStop.class.getName()+"\n";
+        MainThread main = new MainThread(CLIENT_PORT, true, configs);
+        main.start();
+
+        Assert.assertTrue("waiting for server being up",
+                ClientBase.waitForServerUp("127.0.0.1:" + CLIENT_PORT,
+                        CONNECTION_TIMEOUT));
+
+        clientConnected = new CountDownLatch(1);
+        ZooKeeper zk = new ZooKeeper("127.0.0.1:" + CLIENT_PORT,
+                ClientBase.CONNECTION_TIMEOUT, this);
+        Assert.assertTrue("Failed to establish zkclient connection!",
+                clientConnected.await(CONNECTION_TIMEOUT, TimeUnit.MILLISECONDS));
+
+        zk.create("/foo", "foobar".getBytes(), Ids.OPEN_ACL_UNSAFE,
+                CreateMode.PERSISTENT);
+        Assert.assertEquals(new String(zk.getData("/foo", null, null)), "foobar");
+        zk.close();
+
+        main.shutdown();
+        main.join();
+        main.deleteDirs();
+
+        Assert.assertTrue("waiting for server down",
+                ClientBase.waitForServerDown("127.0.0.1:" + CLIENT_PORT,
+                        ClientBase.CONNECTION_TIMEOUT));
+        Assert.assertTrue(MetricsProviderWithErrorInStop.stopCalled.get());
+    }
+
+    /**
+     * Test verifies that configuration is passed to the MetricsProvider.
+     */
+    @Test
+    public void testMetricsProviderConfiguration()
+            throws Exception {
+        ClientBase.setupTestEnv();
+
+        final int CLIENT_PORT = PortAssignment.unique();
+        MetricsProviderWithConfiguration.httpPort.set(0);
+        final String configs = "metricsProvider.className="+MetricsProviderWithConfiguration.class.getName()+"\n"+
+                               "metricsProvider.httpPort=1234\n";
+        MainThread main = new MainThread(CLIENT_PORT, true, configs);
+        main.start();
+
+        Assert.assertTrue("waiting for server being up",
+                ClientBase.waitForServerUp("127.0.0.1:" + CLIENT_PORT,
+                        CONNECTION_TIMEOUT));
+
+        clientConnected = new CountDownLatch(1);
+        ZooKeeper zk = new ZooKeeper("127.0.0.1:" + CLIENT_PORT,
+                ClientBase.CONNECTION_TIMEOUT, this);
+        Assert.assertTrue("Failed to establish zkclient connection!",
+                clientConnected.await(CONNECTION_TIMEOUT, TimeUnit.MILLISECONDS));
+
+        zk.create("/foo", "foobar".getBytes(), Ids.OPEN_ACL_UNSAFE,
+                CreateMode.PERSISTENT);
+        Assert.assertEquals(new String(zk.getData("/foo", null, null)), "foobar");
+        zk.close();
+
+        main.shutdown();
+        main.join();
+        main.deleteDirs();
+
+        Assert.assertTrue("waiting for server down",
+                ClientBase.waitForServerDown("127.0.0.1:" + CLIENT_PORT,
+                        ClientBase.CONNECTION_TIMEOUT));
+        Assert.assertEquals(1234, MetricsProviderWithConfiguration.httpPort.get());
+    }
+
+    /**
+     * Test verifies that all of the lifecycle methods of the MetricsProvider are called.
+     */
+    @Test
+    public void testMetricsProviderLifecycle()
+            throws Exception {
+        ClientBase.setupTestEnv();
+        MetricsProviderCapturingLifecycle.reset();
+
+        final int CLIENT_PORT = PortAssignment.unique();
+        final String configs = "metricsProvider.className="+MetricsProviderCapturingLifecycle.class.getName()+"\n"+
+                               "metricsProvider.httpPort=1234\n";
+        MainThread main = new MainThread(CLIENT_PORT, true, configs);
+        main.start();
+
+        Assert.assertTrue("waiting for server being up",
+                ClientBase.waitForServerUp("127.0.0.1:" + CLIENT_PORT,
+                        CONNECTION_TIMEOUT));
+
+        clientConnected = new CountDownLatch(1);
+        ZooKeeper zk = new ZooKeeper("127.0.0.1:" + CLIENT_PORT,
+                ClientBase.CONNECTION_TIMEOUT, this);
+        Assert.assertTrue("Failed to establish zkclient connection!",
+                clientConnected.await(CONNECTION_TIMEOUT, TimeUnit.MILLISECONDS));
+
+        zk.create("/foo", "foobar".getBytes(), Ids.OPEN_ACL_UNSAFE,
+                CreateMode.PERSISTENT);
+        Assert.assertEquals(new String(zk.getData("/foo", null, null)), "foobar");
+        zk.close();
+
+        main.shutdown();
+        main.join();
+        main.deleteDirs();
+
+        Assert.assertTrue("waiting for server down",
+                ClientBase.waitForServerDown("127.0.0.1:" + CLIENT_PORT,
+                        ClientBase.CONNECTION_TIMEOUT));
+
+        Assert.assertTrue("metrics provider lifecycle error",
+                BaseTestMetricsProvider.MetricsProviderCapturingLifecycle.configureCalled.get());
+        Assert.assertTrue("metrics provider lifecycle error",
+                BaseTestMetricsProvider.MetricsProviderCapturingLifecycle.startCalled.get());
+        Assert.assertTrue("metrics provider lifecycle error",
+                BaseTestMetricsProvider.MetricsProviderCapturingLifecycle.getRootContextCalled.get());
+        Assert.assertTrue("metrics provider lifecycle error",
+                BaseTestMetricsProvider.MetricsProviderCapturingLifecycle.stopCalled.get());
+    }
+
     /**
      * Test verifies that the server is able to redefine if user configured only
      * minSessionTimeout limit

src/java/test/org/apache/zookeeper/server/quorum/QuorumPeerMainTest.java

@@ -58,6 +58,8 @@ import org.apache.zookeeper.ZooDefs.Ids;
 import org.apache.zookeeper.ZooKeeper.States;
 import org.apache.zookeeper.common.Time;
 import org.apache.zookeeper.data.Stat;
+import org.apache.zookeeper.metrics.BaseTestMetricsProvider;
+import org.apache.zookeeper.metrics.impl.NullMetricsProvider;
 import org.apache.zookeeper.server.persistence.FileTxnSnapLog;
 import org.apache.zookeeper.server.quorum.Leader.Proposal;
 import org.apache.zookeeper.test.ClientBase;
@@ -1146,6 +1148,8 @@ public class QuorumPeerMainTest extends QuorumPeerTestBase {
             QuorumPeerConfig configMock = mock(QuorumPeerConfig.class);
             when(configMock.getDataDir()).thenReturn(dataDir);
             when(configMock.getDataLogDir()).thenReturn(dataLogDir);
+            when(configMock.getMetricsProviderClassName())
+                    .thenReturn(NullMetricsProvider.class.getName());
 
             QuorumPeer qpMock = mock(QuorumPeer.class);
 
@@ -1486,6 +1490,377 @@ public class QuorumPeerMainTest extends QuorumPeerTestBase {
         }
     }
 
+    /**
+     * Verify boot works configuring a MetricsProvider
+     */
+    @Test
+    public void testMetricsProviderLifecycle() throws Exception {
+        ClientBase.setupTestEnv();
+        BaseTestMetricsProvider.MetricsProviderCapturingLifecycle.reset();
+
+        // setup the logger to capture all logs
+        ByteArrayOutputStream os = new ByteArrayOutputStream();
+        WriterAppender appender = getConsoleAppender(os, Level.WARN);
+        Logger qlogger = Logger.getLogger("org.apache.zookeeper.server.quorum");
+        qlogger.addAppender(appender);
+
+        try {
+            final int CLIENT_PORT_QP1 = PortAssignment.unique();
+            final int CLIENT_PORT_QP2 = PortAssignment.unique();
+
+            String quorumCfgSectionServer
+                    = "server.1=127.0.0.1:" + PortAssignment.unique()
+                    + ":" + PortAssignment.unique() + ";" + CLIENT_PORT_QP1 + "\n"
+                    + "server.2=127.0.0.1:" + PortAssignment.unique()
+                    + ":" + PortAssignment.unique() + ";" + CLIENT_PORT_QP2 + "\n";
+
+            // server 1 boots with a MetricsProvider
+            String quorumCfgSectionServer1 =
+                    quorumCfgSectionServer
+                    + "metricsProvider.className=" + BaseTestMetricsProvider.MetricsProviderCapturingLifecycle.class.getName() + "\n";
+
+            MainThread q1 = new MainThread(1, CLIENT_PORT_QP1, quorumCfgSectionServer1);
+            MainThread q2 = new MainThread(2, CLIENT_PORT_QP2, quorumCfgSectionServer);
+            q1.start();
+            q2.start();
+
+            boolean isup1
+                    = ClientBase.waitForServerUp("127.0.0.1:" + CLIENT_PORT_QP1,
+                            30000);
+            boolean isup2
+                    = ClientBase.waitForServerUp("127.0.0.1:" + CLIENT_PORT_QP2,
+                            30000);
+            Assert.assertTrue("Server 1 never came up", isup1);
+            Assert.assertTrue("Server 2 never came up", isup2);
+
+            q1.shutdown();
+            q2.shutdown();
+
+            Assert.assertTrue("waiting for server 1 down",
+                    ClientBase.waitForServerDown("127.0.0.1:" + CLIENT_PORT_QP1,
+                            ClientBase.CONNECTION_TIMEOUT));
+
+            Assert.assertTrue("waiting for server 2 down",
+                    ClientBase.waitForServerDown("127.0.0.1:" + CLIENT_PORT_QP2,
+                            ClientBase.CONNECTION_TIMEOUT));
+        } finally {
+            qlogger.removeAppender(appender);
+        }
+
+        Assert.assertTrue("metrics provider lifecycle error",
+                BaseTestMetricsProvider.MetricsProviderCapturingLifecycle.configureCalled.get());
+        Assert.assertTrue("metrics provider lifecycle error",
+                BaseTestMetricsProvider.MetricsProviderCapturingLifecycle.startCalled.get());
+        Assert.assertTrue("metrics provider lifecycle error",
+                BaseTestMetricsProvider.MetricsProviderCapturingLifecycle.getRootContextCalled.get());
+        Assert.assertTrue("metrics provider lifecycle error",
+                BaseTestMetricsProvider.MetricsProviderCapturingLifecycle.stopCalled.get());
+    }
+
+    /**
+     * Test verifies that configuration is passed to the MetricsProvider.
+     */
+    @Test
+    public void testMetricsProviderConfiguration() throws Exception {
+        ClientBase.setupTestEnv();
+        BaseTestMetricsProvider.MetricsProviderWithConfiguration.httpPort.set(0);
+
+        // setup the logger to capture all logs
+        ByteArrayOutputStream os = new ByteArrayOutputStream();
+        WriterAppender appender = getConsoleAppender(os, Level.WARN);
+        Logger qlogger = Logger.getLogger("org.apache.zookeeper.server.quorum");
+        qlogger.addAppender(appender);
+
+        try {
+            final int CLIENT_PORT_QP1 = PortAssignment.unique();
+            final int CLIENT_PORT_QP2 = PortAssignment.unique();
+
+            String quorumCfgSectionServer
+                    = "server.1=127.0.0.1:" + PortAssignment.unique()
+                    + ":" + PortAssignment.unique() + ";" + CLIENT_PORT_QP1 + "\n"
+                    + "server.2=127.0.0.1:" + PortAssignment.unique()
+                    + ":" + PortAssignment.unique() + ";" + CLIENT_PORT_QP2 + "\n";
+
+            // server 1 boots with a MetricsProvider
+            String quorumCfgSectionServer1 =
+                    quorumCfgSectionServer
+                    + "metricsProvider.className=" + BaseTestMetricsProvider.MetricsProviderWithConfiguration.class.getName() + "\n"
+                    + "metricsProvider.httpPort=1234";
+
+            MainThread q1 = new MainThread(1, CLIENT_PORT_QP1, quorumCfgSectionServer1);
+            MainThread q2 = new MainThread(2, CLIENT_PORT_QP2, quorumCfgSectionServer);
+            q1.start();
+            q2.start();
+
+            boolean isup1
+                    = ClientBase.waitForServerUp("127.0.0.1:" + CLIENT_PORT_QP1,
+                            30000);
+            boolean isup2
+                    = ClientBase.waitForServerUp("127.0.0.1:" + CLIENT_PORT_QP2,
+                            30000);
+            Assert.assertTrue("Server 1 never came up", isup1);
+            Assert.assertTrue("Server 2 never came up", isup2);
+
+            q1.shutdown();
+            q2.shutdown();
+
+            Assert.assertTrue("waiting for server 1 down",
+                    ClientBase.waitForServerDown("127.0.0.1:" + CLIENT_PORT_QP1,
+                            ClientBase.CONNECTION_TIMEOUT));
+
+            Assert.assertTrue("waiting for server 2 down",
+                    ClientBase.waitForServerDown("127.0.0.1:" + CLIENT_PORT_QP2,
+                            ClientBase.CONNECTION_TIMEOUT));
+        } finally {
+            qlogger.removeAppender(appender);
+        }
+
+        Assert.assertEquals(1234,
+                BaseTestMetricsProvider.MetricsProviderWithConfiguration.httpPort.get());
+    }
+
+    /**
+     * Test verifies that the server shouldn't be affected but runtime errors on stop()
+     */
+    @Test
+    public void testFaultyMetricsProviderOnStop() throws Exception {
+        ClientBase.setupTestEnv();
+        BaseTestMetricsProvider.MetricsProviderCapturingLifecycle.reset();
+
+        // setup the logger to capture all logs
+        ByteArrayOutputStream os = new ByteArrayOutputStream();
+        WriterAppender appender = getConsoleAppender(os, Level.WARN);
+        Logger qlogger = Logger.getLogger("org.apache.zookeeper.server.quorum");
+        qlogger.addAppender(appender);
+
+        try {
+            final int CLIENT_PORT_QP1 = PortAssignment.unique();
+            final int CLIENT_PORT_QP2 = PortAssignment.unique();
+
+            String quorumCfgSectionServer
+                    = "server.1=127.0.0.1:" + PortAssignment.unique()
+                    + ":" + PortAssignment.unique() + ";" + CLIENT_PORT_QP1 + "\n"
+                    + "server.2=127.0.0.1:" + PortAssignment.unique()
+                    + ":" + PortAssignment.unique() + ";" + CLIENT_PORT_QP2 + "\n";
+
+            // server 1 boots with a MetricsProvider
+            String quorumCfgSectionServer1 =
+                    quorumCfgSectionServer
+                    + "metricsProvider.className=" + BaseTestMetricsProvider.MetricsProviderWithErrorInStop.class.getName() + "\n";
+
+            MainThread q1 = new MainThread(1, CLIENT_PORT_QP1, quorumCfgSectionServer1);
+            MainThread q2 = new MainThread(2, CLIENT_PORT_QP2, quorumCfgSectionServer);
+            q1.start();
+            q2.start();
+
+            boolean isup1
+                    = ClientBase.waitForServerUp("127.0.0.1:" + CLIENT_PORT_QP1,
+                            30000);
+            boolean isup2
+                    = ClientBase.waitForServerUp("127.0.0.1:" + CLIENT_PORT_QP2,
+                            30000);
+            Assert.assertTrue("Server 1 never came up", isup1);
+            Assert.assertTrue("Server 2 never came up", isup2);
+
+            q1.shutdown();
+            q2.shutdown();
+
+            Assert.assertTrue("waiting for server 1 down",
+                    ClientBase.waitForServerDown("127.0.0.1:" + CLIENT_PORT_QP1,
+                            ClientBase.CONNECTION_TIMEOUT));
+
+            Assert.assertTrue("waiting for server 2 down",
+                    ClientBase.waitForServerDown("127.0.0.1:" + CLIENT_PORT_QP2,
+                            ClientBase.CONNECTION_TIMEOUT));
+        } finally {
+            qlogger.removeAppender(appender);
+        }
+
+        Assert.assertTrue("metrics provider lifecycle error",
+                BaseTestMetricsProvider.MetricsProviderWithErrorInStop.stopCalled.get());
+
+        LineNumberReader r = new LineNumberReader(new StringReader(os.toString()));
+        String line;
+        boolean found = false;
+        Pattern p
+                = Pattern.compile(".*Error while stopping metrics.*");
+        while ((line = r.readLine()) != null) {
+            found = p.matcher(line).matches();
+            if (found) {
+                break;
+            }
+        }
+        Assert.assertTrue("complains about metrics provider", found);
+    }
+
+    /**
+     * Verify boot fails with a bad MetricsProvider
+     */
+    @Test
+    public void testInvalidMetricsProvider() throws Exception {
+        ClientBase.setupTestEnv();
+
+        // setup the logger to capture all logs
+        ByteArrayOutputStream os = new ByteArrayOutputStream();
+        WriterAppender appender = getConsoleAppender(os, Level.WARN);
+        Logger qlogger = Logger.getLogger("org.apache.zookeeper.server.quorum");
+        qlogger.addAppender(appender);
+
+        try {
+            final int CLIENT_PORT_QP1 = PortAssignment.unique();
+
+            String quorumCfgSection
+                    = "server.1=127.0.0.1:" + PortAssignment.unique()
+                    + ":" + PortAssignment.unique() + ";" + CLIENT_PORT_QP1 + "\n"
+                    + "server.2=127.0.0.1:" + PortAssignment.unique()
+                    + ":" + PortAssignment.unique() + ";" + CLIENT_PORT_QP1 + "\n"
+                    + "metricsProvider.className=BadClass\n";
+
+            MainThread q1 = new MainThread(1, CLIENT_PORT_QP1, quorumCfgSection);
+            q1.start();
+
+            boolean isup
+                    = ClientBase.waitForServerUp("127.0.0.1:" + CLIENT_PORT_QP1,
+                            5000);
+
+            Assert.assertFalse("Server never came up", isup);
+
+            q1.shutdown();
+
+            Assert.assertTrue("waiting for server 1 down",
+                    ClientBase.waitForServerDown("127.0.0.1:" + CLIENT_PORT_QP1,
+                            ClientBase.CONNECTION_TIMEOUT));
+
+        } finally {
+            qlogger.removeAppender(appender);
+        }
+
+        LineNumberReader r = new LineNumberReader(new StringReader(os.toString()));
+        String line;
+        boolean found = false;
+        Pattern p
+                = Pattern.compile(".*BadClass.*");
+        while ((line = r.readLine()) != null) {
+            found = p.matcher(line).matches();
+            if (found) {
+                break;
+            }
+        }
+        Assert.assertTrue("complains about metrics provider", found);
+    }
+
+    /**
+     * Verify boot fails with a MetricsProvider with fails to start
+     */
+    @Test
+    public void testFaultyMetricsProviderOnStart() throws Exception {
+        ClientBase.setupTestEnv();
+
+        // setup the logger to capture all logs
+        ByteArrayOutputStream os = new ByteArrayOutputStream();
+        WriterAppender appender = getConsoleAppender(os, Level.WARN);
+        Logger qlogger = Logger.getLogger("org.apache.zookeeper.server.quorum");
+        qlogger.addAppender(appender);
+
+        try {
+            final int CLIENT_PORT_QP1 = PortAssignment.unique();
+
+            String quorumCfgSection
+                    = "server.1=127.0.0.1:" + PortAssignment.unique()
+                    + ":" + PortAssignment.unique() + ";" + CLIENT_PORT_QP1 + "\n"
+                    + "server.2=127.0.0.1:" + PortAssignment.unique()
+                    + ":" + PortAssignment.unique() + ";" + CLIENT_PORT_QP1 + "\n"
+                    + "metricsProvider.className=" + BaseTestMetricsProvider.MetricsProviderWithErrorInStart.class.getName() + "\n";
+
+            MainThread q1 = new MainThread(1, CLIENT_PORT_QP1, quorumCfgSection);
+            q1.start();
+
+            boolean isup
+                    = ClientBase.waitForServerUp("127.0.0.1:" + CLIENT_PORT_QP1,
+                            5000);
+
+            Assert.assertFalse("Server never came up", isup);
+
+            q1.shutdown();
+
+            Assert.assertTrue("waiting for server 1 down",
+                    ClientBase.waitForServerDown("127.0.0.1:" + CLIENT_PORT_QP1,
+                            ClientBase.CONNECTION_TIMEOUT));
+
+        } finally {
+            qlogger.removeAppender(appender);
+        }
+
+        LineNumberReader r = new LineNumberReader(new StringReader(os.toString()));
+        String line;
+        boolean found = false;
+        Pattern p
+                = Pattern.compile(".*MetricsProviderLifeCycleException.*");
+        while ((line = r.readLine()) != null) {
+            found = p.matcher(line).matches();
+            if (found) {
+                break;
+            }
+        }
+        Assert.assertTrue("complains about metrics provider MetricsProviderLifeCycleException", found);
+    }
+
+    /**
+     * Verify boot fails with a MetricsProvider with fails to start
+     */
+    @Test
+    public void testFaultyMetricsProviderOnConfigure() throws Exception {
+        ClientBase.setupTestEnv();
+
+        // setup the logger to capture all logs
+        ByteArrayOutputStream os = new ByteArrayOutputStream();
+        WriterAppender appender = getConsoleAppender(os, Level.WARN);
+        Logger qlogger = Logger.getLogger("org.apache.zookeeper.server.quorum");
+        qlogger.addAppender(appender);
+
+        try {
+            final int CLIENT_PORT_QP1 = PortAssignment.unique();
+
+            String quorumCfgSection
+                    = "server.1=127.0.0.1:" + PortAssignment.unique()
+                    + ":" + PortAssignment.unique() + ";" + CLIENT_PORT_QP1 + "\n"
+                    + "server.2=127.0.0.1:" + PortAssignment.unique()
+                    + ":" + PortAssignment.unique() + ";" + CLIENT_PORT_QP1 + "\n"
+                    + "metricsProvider.className=" + BaseTestMetricsProvider.MetricsProviderWithErrorInConfigure.class.getName() + "\n";
+
+            MainThread q1 = new MainThread(1, CLIENT_PORT_QP1, quorumCfgSection);
+            q1.start();
+
+            boolean isup
+                    = ClientBase.waitForServerUp("127.0.0.1:" + CLIENT_PORT_QP1,
+                            5000);
+
+            Assert.assertFalse("Server never came up", isup);
+
+            q1.shutdown();
+
+            Assert.assertTrue("waiting for server 1 down",
+                    ClientBase.waitForServerDown("127.0.0.1:" + CLIENT_PORT_QP1,
+                            ClientBase.CONNECTION_TIMEOUT));
+
+        } finally {
+            qlogger.removeAppender(appender);
+        }
+
+        LineNumberReader r = new LineNumberReader(new StringReader(os.toString()));
+        String line;
+        boolean found = false;
+        Pattern p
+                = Pattern.compile(".*MetricsProviderLifeCycleException.*");
+        while ((line = r.readLine()) != null) {
+            found = p.matcher(line).matches();
+            if (found) {
+                break;
+            }
+        }
+        Assert.assertTrue("complains about metrics provider MetricsProviderLifeCycleException", found);
+    }
+
     static class Context {
         boolean quitFollowing = false;
         boolean exitWhenAckNewLeader = false;