ZOOKEEPER-2014: Only admin should be allowed to reconfig a cluster.

This PR implements ZOOKEEPER-2014. For details, please refer to

JIRA: https://issues.apache.org/jira/browse/ZOOKEEPER-2014
Review board: https://reviews.apache.org/r/51546/

Author: Michael Han <hanm@cloudera.com>

Reviewers: fpj <fpj@apache.org>, breed <breed@apache.org>, rgs <rgs@itevenworks.net>

Closes #96 from hanm/ZOOKEEPER-2014

build.xml

@@ -528,6 +528,7 @@ xmlns:cs="antlib:com.puppycrawl.tools.checkstyle.ant">
           <include name="org/apache/zookeeper/WatchedEvent.java"/>
           <include name="org/apache/zookeeper/ZooDefs.java"/>
           <include name="org/apache/zookeeper/ZooKeeper.java"/>
+          <include name="org/apache/zookeeper/admin/ZooKeeperAdmin.java"/>
           <include name="org/apache/zookeeper/server/LogFormatter.java"/>
           <include name="org/apache/zookeeper/server/SnapshotFormatter.java"/>
           <include name="org/apache/zookeeper/server/PurgeTxnLog.java"/>

src/c/include/zookeeper.h

@@ -124,7 +124,8 @@ enum ZOO_ERRORS {
   ZNOTREADONLY = -119, /*!< state-changing request is passed to read-only server */
   ZEPHEMERALONLOCALSESSION = -120, /*!< Attempt to create ephemeral node on a local session */
   ZNOWATCHER = -121, /*!< The watcher couldn't be found */
-  ZRWSERVERFOUND = -122 /*!< r/w server found while in r/o mode */
+  ZRWSERVERFOUND = -122, /*!< r/w server found while in r/o mode */
+  ZRECONFIGDISABLED = -123 /*!< Attempts to perform a reconfiguration operation when reconfiguration feature is disabled */
 };
 
 #ifdef __cplusplus

src/c/tests/TestReconfigServer.cc

@@ -15,6 +15,9 @@
  * the License.
  */
 #include <algorithm>
+#include <sstream>
+#include <vector>
+#include <utility>
 #include <cppunit/extensions/HelperMacros.h>
 #include <unistd.h>
 #include "zookeeper.h"
@@ -28,6 +31,8 @@ class TestReconfigServer : public CPPUNIT_NS::TestFixture {
     CPPUNIT_TEST(testNonIncremental);
     CPPUNIT_TEST(testRemoveConnectedFollower);
     CPPUNIT_TEST(testRemoveFollower);
+    CPPUNIT_TEST(testReconfigFailureWithoutAuth);
+    CPPUNIT_TEST(testReconfigFailureWithoutServerSuperuserPasswordConfigured);
 #endif
     CPPUNIT_TEST_SUITE_END();
 
@@ -39,7 +44,8 @@ class TestReconfigServer : public CPPUNIT_NS::TestFixture {
     void testNonIncremental();
     void testRemoveConnectedFollower();
     void testRemoveFollower();
-
+    void testReconfigFailureWithoutAuth();
+    void testReconfigFailureWithoutServerSuperuserPasswordConfigured();
   private:
     static const uint32_t NUM_SERVERS;
     FILE* logfile_;
@@ -49,6 +55,7 @@ class TestReconfigServer : public CPPUNIT_NS::TestFixture {
     void parseConfig(char* buf, int len, std::vector<std::string>& servers,
                      std::string& version);
     bool waitForConnected(zhandle_t* zh, uint32_t timeout_sec);
+    zhandle_t* connectFollowers(std::vector<int32_t> &followers);
 };
 
 const uint32_t TestReconfigServer::NUM_SERVERS = 3;
@@ -70,7 +77,10 @@ TestReconfigServer::
 
 void TestReconfigServer::
 setUp() {
-    cluster_ = ZooKeeperQuorumServer::getCluster(NUM_SERVERS);
+    ZooKeeperQuorumServer::tConfigPairs configs;
+    configs.push_back(std::make_pair("reconfigEnabled", "true"));
+    cluster_ = ZooKeeperQuorumServer::getCluster(NUM_SERVERS, configs,
+        "SERVER_JVMFLAGS=-Dzookeeper.DigestAuthenticationProvider.superDigest=super:D/InIHSb7yEEbrWz8b9l71RjZJU="/* password is test */);
 }
 
 void TestReconfigServer::
@@ -151,7 +161,7 @@ testRemoveFollower() {
     zhandle_t* zk = zookeeper_init(host.c_str(), NULL, 10000, NULL, NULL, 0);
     CPPUNIT_ASSERT_EQUAL(true, waitForConnected(zk, 10));
     CPPUNIT_ASSERT_EQUAL((int)ZOK, zoo_getconfig(zk, 0, buf, &len, &stat));
-
+    CPPUNIT_ASSERT_EQUAL((int)ZOK, zoo_add_auth(zk, "digest", "super:test", 10, NULL,(void*)ZOK));
     // check if all the servers are listed in the config.
     parseConfig(buf, len, servers, version);
     // initially should be 1<<32, which is 0x100000000. This is the zxid
@@ -219,6 +229,7 @@ testNonIncremental() {
     zhandle_t* zk = zookeeper_init(host.c_str(), NULL, 10000, NULL, NULL, 0);
     CPPUNIT_ASSERT_EQUAL(true, waitForConnected(zk, 10));
     CPPUNIT_ASSERT_EQUAL((int)ZOK, zoo_getconfig(zk, 0, buf, &len, &stat));
+    CPPUNIT_ASSERT_EQUAL((int)ZOK, zoo_add_auth(zk, "digest", "super:test", 10, NULL,(void*)ZOK));
 
     // check if all the servers are listed in the config.
     parseConfig(buf, len, servers, version);
@@ -274,37 +285,46 @@ testNonIncremental() {
     zookeeper_close(zk);
 }
 
-/**
- * 1. Connect to a follower.
- * 2. Remove the follower the client is connected to.
- */
-void TestReconfigServer::
-testRemoveConnectedFollower() {
-    std::vector<std::string> servers;
-    std::string version;
-    struct Stat stat;
-    int len = 1024;
-    char buf[len];
-
-    // connect to a follower.
+zhandle_t* TestReconfigServer::
+connectFollowers(std::vector<int32_t> &followers) {
+    std::stringstream ss;
     int32_t leader = getLeader();
-    std::vector<int32_t> followers = getFollowers();
     CPPUNIT_ASSERT(leader >= 0);
     CPPUNIT_ASSERT_EQUAL(NUM_SERVERS - 1, (uint32_t)(followers.size()));
-    std::stringstream ss;
     for (int i = 0; i < followers.size(); i++) {
-      ss << cluster_[followers[i]]->getHostPort() << ",";
+        ss << cluster_[followers[i]]->getHostPort() << ",";
     }
     ss << cluster_[leader]->getHostPort();
     std::string hosts = ss.str().c_str();
     zoo_deterministic_conn_order(true);
     zhandle_t* zk = zookeeper_init(hosts.c_str(), NULL, 10000, NULL, NULL, 0);
     CPPUNIT_ASSERT_EQUAL(true, waitForConnected(zk, 10));
+
     std::string connectedHost(zoo_get_current_server(zk));
     std::string portString = connectedHost.substr(connectedHost.find(":") + 1);
     uint32_t port;
     std::istringstream (portString) >> port;
     CPPUNIT_ASSERT_EQUAL(cluster_[followers[0]]->getClientPort(), port);
+    return zk;
+}
+
+/**
+ * 1. Connect to a follower.
+ * 2. Remove the follower the client is connected to.
+ */
+void TestReconfigServer::
+testRemoveConnectedFollower() {
+    std::vector<std::string> servers;
+    std::string version;
+    struct Stat stat;
+    int len = 1024;
+    char buf[len];
+
+    // connect to a follower.
+    std::stringstream ss;
+    std::vector<int32_t> followers = getFollowers();
+    zhandle_t* zk = connectFollowers(followers);
+    CPPUNIT_ASSERT_EQUAL((int)ZOK, zoo_add_auth(zk, "digest", "super:test", 10, NULL,(void*)ZOK));
 
     // remove the follower.
     len = 1024;
@@ -324,4 +344,77 @@ testRemoveConnectedFollower() {
     zookeeper_close(zk);
 }
 
+/**
+ * ZOOKEEPER-2014: only admin or users who are explicitly granted permission can do reconfig.
+ */
+void TestReconfigServer::
+testReconfigFailureWithoutAuth() {
+    std::vector<std::string> servers;
+    std::string version;
+    struct Stat stat;
+    int len = 1024;
+    char buf[len];
+
+    // connect to a follower.
+    std::stringstream ss;
+    std::vector<int32_t> followers = getFollowers();
+    zhandle_t* zk = connectFollowers(followers);
+
+    // remove the follower.
+    len = 1024;
+    ss.str("");
+    ss << followers[0];
+    // No auth, should fail.
+    CPPUNIT_ASSERT_EQUAL((int)ZNOAUTH, zoo_reconfig(zk, NULL, ss.str().c_str(), NULL, -1, buf, &len, &stat));
+    // Wrong auth, should fail.
+    CPPUNIT_ASSERT_EQUAL((int)ZOK, zoo_add_auth(zk, "digest", "super:wrong", 11, NULL,(void*)ZOK));
+    CPPUNIT_ASSERT_EQUAL((int)ZNOAUTH, zoo_reconfig(zk, NULL, ss.str().c_str(), NULL, -1, buf, &len, &stat));
+    // Right auth, should pass.
+    CPPUNIT_ASSERT_EQUAL((int)ZOK, zoo_add_auth(zk, "digest", "super:test", 10, NULL,(void*)ZOK));
+    CPPUNIT_ASSERT_EQUAL((int)ZOK, zoo_reconfig(zk, NULL, ss.str().c_str(), NULL, -1, buf, &len, &stat));
+    CPPUNIT_ASSERT_EQUAL((int)ZOK, zoo_getconfig(zk, 0, buf, &len, &stat));
+    parseConfig(buf, len, servers, version);
+    CPPUNIT_ASSERT_EQUAL(NUM_SERVERS - 1, (uint32_t)(servers.size()));
+    for (int i = 0; i < cluster_.size(); i++) {
+        if (i == followers[0]) {
+            continue;
+        }
+        CPPUNIT_ASSERT(std::find(servers.begin(), servers.end(),
+                       cluster_[i]->getServerString()) != servers.end());
+    }
+    zookeeper_close(zk);
+}
+
+void TestReconfigServer::
+testReconfigFailureWithoutServerSuperuserPasswordConfigured() {
+    std::vector<std::string> servers;
+    std::string version;
+    struct Stat stat;
+    int len = 1024;
+    char buf[len];
+
+    // Create a new quorum with the super user's password not configured.
+    tearDown();
+    ZooKeeperQuorumServer::tConfigPairs configs;
+    configs.push_back(std::make_pair("reconfigEnabled", "true"));
+    cluster_ = ZooKeeperQuorumServer::getCluster(NUM_SERVERS, configs, "");
+
+    // connect to a follower.
+    std::stringstream ss;
+    std::vector<int32_t> followers = getFollowers();
+    zhandle_t* zk = connectFollowers(followers);
+
+    // remove the follower.
+    len = 1024;
+    ss.str("");
+    ss << followers[0];
+    // All cases should fail as server ensemble was not configured with the super user's password.
+    CPPUNIT_ASSERT_EQUAL((int)ZNOAUTH, zoo_reconfig(zk, NULL, ss.str().c_str(), NULL, -1, buf, &len, &stat));
+    CPPUNIT_ASSERT_EQUAL((int)ZOK, zoo_add_auth(zk, "digest", "super:", 11, NULL,(void*)ZOK));
+    CPPUNIT_ASSERT_EQUAL((int)ZNOAUTH, zoo_reconfig(zk, NULL, ss.str().c_str(), NULL, -1, buf, &len, &stat));
+    CPPUNIT_ASSERT_EQUAL((int)ZOK, zoo_add_auth(zk, "digest", "super:test", 10, NULL,(void*)ZOK));
+    CPPUNIT_ASSERT_EQUAL((int)ZNOAUTH, zoo_reconfig(zk, NULL, ss.str().c_str(), NULL, -1, buf, &len, &stat));
+    zookeeper_close(zk);
+}
+
 CPPUNIT_TEST_SUITE_REGISTRATION(TestReconfigServer);

src/c/tests/ZooKeeperQuorumServer.cc

@@ -21,18 +21,21 @@
 #include <cstdlib>
 #include <fstream>
 #include <sstream>
+#include <vector>
+#include <utility>
 #include <unistd.h>
 
 ZooKeeperQuorumServer::
-ZooKeeperQuorumServer(uint32_t id, uint32_t numServers) :
+ZooKeeperQuorumServer(uint32_t id, uint32_t numServers, std::string config, std::string env) :
     id_(id),
+    env_(env),
     numServers_(numServers) {
     const char* root = getenv("ZKROOT");
     if (root == NULL) {
         assert(!"Environment variable 'ZKROOT' is not set");
     }
     root_ = root;
-    createConfigFile();
+    createConfigFile(config);
     createDataDirectory();
     start();
 }
@@ -58,6 +61,9 @@ void ZooKeeperQuorumServer::
 start() {
     std::string command = root_ + "/bin/zkServer.sh start " +
                           getConfigFileName();
+    if (!env_.empty()) {
+        command = env_ + " " + command;
+    }
     assert(system(command.c_str()) == 0);
 }
 
@@ -102,7 +108,7 @@ isFollower() {
 }
 
 void ZooKeeperQuorumServer::
-createConfigFile() {
+createConfigFile(std::string config) {
     std::string command = "mkdir -p " + root_ + "/build/test/test-cppunit/conf";
     assert(system(command.c_str()) == 0);
     std::ofstream confFile;
@@ -118,6 +124,10 @@ createConfigFile() {
     for (int i = 0; i < numServers_; i++) {
         confFile << getServerString(i) << "\n";
     }
+    // Append additional config, if any.
+    if (!config.empty()) {
+      confFile << config << std::endl;
+    }
     confFile.close();
 }
 
@@ -188,3 +198,33 @@ getCluster(uint32_t numServers) {
     }
     assert(!"The cluster didn't start for 10 seconds");
 }
+
+std::vector<ZooKeeperQuorumServer*> ZooKeeperQuorumServer::
+getCluster(uint32_t numServers, ZooKeeperQuorumServer::tConfigPairs configs, std::string env) {
+    std::vector<ZooKeeperQuorumServer*> cluster;
+    std::string config;
+    for (ZooKeeperQuorumServer::tConfigPairs::const_iterator iter = configs.begin(); iter != configs.end(); ++iter) {
+        std::pair<std::string, std::string> pair = *iter;
+        config += (pair.first + "=" + pair.second + "\n");
+    }
+    for (int i = 0; i < numServers; i++) {
+        cluster.push_back(new ZooKeeperQuorumServer(i, numServers, config, env));
+    }
+
+    // Wait until all the servers start, and fail if they don't start within 10
+    // seconds.
+    for (int i = 0; i < 10; i++) {
+        int j = 0;
+        for (; j < cluster.size(); j++) {
+            if (cluster[j]->getMode() == "") {
+                // The server hasn't started.
+                sleep(1);
+                break;
+            }
+        }
+        if (j == cluster.size()) {
+            return cluster;
+        }
+    }
+    assert(!"The cluster didn't start for 10 seconds");
+}

src/c/tests/ZooKeeperQuorumServer.h

@@ -20,11 +20,16 @@
 #include <stdint.h>
 #include <string>
 #include <vector>
+#include <utility>
 
 class ZooKeeperQuorumServer {
   public:
     ~ZooKeeperQuorumServer();
+    typedef std::vector<std::pair<std::string, std::string> > tConfigPairs;
     static std::vector<ZooKeeperQuorumServer*> getCluster(uint32_t numServers);
+    static std::vector<ZooKeeperQuorumServer*> getCluster(uint32_t numServers,
+        tConfigPairs configs, /* Additional config options as a list of key/value pairs. */
+        std::string env       /* Additional environment variables when starting zkServer.sh. */);
     std::string getHostPort();
     uint32_t getClientPort();
     void start();
@@ -35,10 +40,11 @@ class ZooKeeperQuorumServer {
 
   private:
     ZooKeeperQuorumServer();
-    ZooKeeperQuorumServer(uint32_t id, uint32_t numServers);
+    ZooKeeperQuorumServer(uint32_t id, uint32_t numServers, std::string config = "",
+                          std::string env = "");
     ZooKeeperQuorumServer(const ZooKeeperQuorumServer& that);
     const ZooKeeperQuorumServer& operator=(const ZooKeeperQuorumServer& that);
-    void createConfigFile();
+    void createConfigFile(std::string config = "");
     std::string getConfigFileName();
     void createDataDirectory();
     std::string getDataDirectory();
@@ -52,6 +58,7 @@ class ZooKeeperQuorumServer {
     uint32_t numServers_;
     uint32_t id_;
     std::string root_;
+    std::string env_;
 };
 
 #endif  // ZOOKEEPER_QUORUM_SERVER_H

src/docs/src/documentation/content/xdocs/zookeeperAdmin.xml

@@ -931,6 +931,7 @@ server.3=zoo3:2888:3888</programlisting>
               feature. Default is "true"</para>
             </listitem>
           </varlistentry>
+
         </variablelist>
       </section>
 
@@ -1108,6 +1109,42 @@ server.3=zoo3:2888:3888</programlisting>
               </para>
             </listitem>
           </varlistentry>
+
+          <varlistentry>
+            <term>reconfigEnabled</term>
+
+            <listitem>
+              <para>(No Java system property)</para>
+
+              <para><emphasis role="bold">New in 3.5.3:</emphasis>
+                This controls the enabling or disabling of
+                <ulink url="zookeeperReconfig.html">
+                  Dynamic Reconfiguration</ulink> feature. When the feature
+                is enabled, users can perform reconfigure operations through
+                the ZooKeeper client API or through ZooKeeper command line tools
+                assuming users are authorized to perform such operations.
+                When the feature is disabled, no user, including the super user,
+                can perform a reconfiguration. Any attempt to reconfigure will return an error.
+                <emphasis role="bold">"reconfigEnabled"</emphasis> option can be set as
+                <emphasis role="bold">"reconfigEnabled=false"</emphasis> or
+                <emphasis role="bold">"reconfigEnabled=true"</emphasis>
+                to a server's config file, or using QuorumPeerConfig's
+                setReconfigEnabled method. The default value is false.
+
+                If present, the value should be consistent across every server in
+                the entire ensemble. Setting the value as true on some servers and false
+                on other servers will cause inconsistent behavior depending on which server
+                is elected as leader. If the leader has a setting of
+                <emphasis role="bold">"reconfigEnabled=true"</emphasis>, then the ensemble
+                will have reconfig feature enabled. If the leader has a setting of
+                <emphasis role="bold">"reconfigEnabled=false"</emphasis>, then the ensemble
+                will have reconfig feature disabled. It is thus recommended to have a consistent
+                value for <emphasis role="bold">"reconfigEnabled"</emphasis> across servers
+                in the ensemble.
+              </para>
+            </listitem>
+          </varlistentry>
+
         </variablelist>
         <para></para>
       </section>

src/docs/src/documentation/content/xdocs/zookeeperReconfig.xml

@@ -83,6 +83,11 @@
         </listitem>
       </varlistentry>
     </variablelist>
+    <para><emphasis role="bold">Note:</emphasis> Starting with 3.5.3, the dynamic reconfiguration
+      feature is disabled by default, and has to be explicitly turned on via
+      <ulink url="zookeeperAdmin.html#sc_advancedConfiguration">
+        reconfigEnabled </ulink> configuration option.
+    </para>
   </section>
   <section id="ch_reconfig_format">
     <title>Changes to Configuration Format</title>
@@ -142,6 +147,25 @@
         recommend setting the flag to <emphasis>false</emphasis>. We expect that
         the legacy Standalone mode will be deprecated in the future.</para>
     </section>
+    <section id="sc_reconfig_reconfigEnabled">
+      <title>The <emphasis>reconfigEnabled</emphasis> flag</title>
+      <para>Starting with 3.5.0 and prior to 3.5.3, there is no way to disable
+        dynamic reconfiguration feature. We would like to offer the option of
+        disabling reconfiguration feature because with reconfiguration enabled,
+        we have a security concern that a malicious actor can make arbitrary changes
+        to the configuration of a ZooKeeper ensemble, including adding a compromised
+        server to the ensemble. We prefer to leave to the discretion of the user to
+        decide whether to enable it or not and make sure that the appropriate security
+        measure are in place. So in 3.5.3 the <ulink url="zookeeperAdmin.html#sc_advancedConfiguration">
+          reconfigEnabled </ulink> configuration option is introduced
+        such that the reconfiguration feature can be completely disabled and any attempts
+        to reconfigure a cluster through reconfig API with or without authentication
+        will fail by default, unless <emphasis role="bold">reconfigEnabled</emphasis> is set to
+        <emphasis role="bold">true</emphasis>.
+      </para>
+      <para>To set the option to true, the configuration file (zoo.cfg) should contain:</para>
+      <para><computeroutput>reconfigEnabled=true</computeroutput></para>
+    </section>
     <section id="sc_reconfig_file">
       <title>Dynamic configuration file</title>
       <para>Starting with 3.5.0 we're distinguishing between dynamic
@@ -252,6 +276,7 @@ server.3=125.23.63.25:2782:2785:participant</programlisting>
       clientPort/clientPortAddress statements (although if you specify client
       ports in the new format, these statements are now redundant).</para>
   </section>
+
   <section id="ch_reconfig_dyn">
     <title>Dynamic Reconfiguration of the ZooKeeper Ensemble</title>
     <para>The ZooKeeper Java and C API were extended with getConfig and reconfig
@@ -260,6 +285,110 @@ server.3=125.23.63.25:2782:2785:participant</programlisting>
       here using the Java CLI, but note that you can similarly use the C CLI or
       invoke the commands directly from a program just like any other ZooKeeper
       command.</para>
+
+    <section id="ch_reconfig_api">
+      <title>API</title>
+      <para>There are two sets of APIs for both Java and C client.
+      </para>
+      <variablelist>
+        <varlistentry>
+          <term><emphasis role="bold">Reconfiguration API</emphasis></term>
+
+          <listitem>
+            <para>Reconfiguration API is used to reconfigure the ZooKeeper cluster.
+              Starting with 3.5.3, reconfiguration Java APIs are moved into ZooKeeperAdmin class
+              from ZooKeeper class, and use of this API requires ACL setup and user
+              authentication (see <xref linkend="sc_reconfig_access_control"/> for more information.).
+            </para>
+          </listitem>
+        </varlistentry>
+
+        <varlistentry>
+          <term><emphasis role="bold">Get Configuration API</emphasis></term>
+          <listitem>
+            <para>Get configuration APIs are used to retrieve ZooKeeper cluster configuration information
+              stored in /zookeeper/config znode. Use of this API does not require specific setup or authentication,
+            because /zookeeper/config is readable to any users.</para>
+          </listitem>
+        </varlistentry>
+      </variablelist>
+    </section>
+
+    <section id="sc_reconfig_access_control">
+      <title>Security</title>
+      <para>Prior to <emphasis role="bold">3.5.3</emphasis>, there is no enforced security mechanism
+        over reconfig so any ZooKeeper clients that can connect to ZooKeeper server ensemble
+        will have the ability to change the state of a ZooKeeper cluster via reconfig.
+        It is thus possible for a malicious client to add compromised server to an ensemble,
+        e.g., add a compromised server, or remove legitimate servers.
+        Cases like these could be security vulnerabilities on a case by case basis.
+      </para>
+      <para>To address this security concern, we introduced access control over reconfig
+        starting from <emphasis role="bold">3.5.3</emphasis> such that only a specific set of users
+        can use reconfig commands or APIs, and these users need be configured explicitly. In addition,
+        the setup of ZooKeeper cluster must enable authentication so ZooKeeper clients can be authenticated.
+      </para>
+      <para>
+        We also provides an escape hatch for users who operate and interact with a ZooKeeper ensemble in a secured
+        environment (i.e. behind company firewall). For those users who want to use reconfiguration feature but
+        don't want the overhead of configuring an explicit list of authorized user for reconfig access checks,
+        they can set <ulink url="zookeeperAdmin.html#sc_authOptions">"skipACL"</ulink> to "yes" which will
+        skip ACL check and allow any user to reconfigure cluster.
+      </para>
+      <para>
+        Overall, ZooKeeper provides flexible configuration options for the reconfigure feature
+        that allow a user to choose based on user's security requirement.
+        We leave to the discretion of the user to decide appropriate security measure are in place.
+      </para>
+      <variablelist>
+        <varlistentry>
+          <term><emphasis role="bold">Access Control</emphasis></term>
+
+          <listitem>
+            <para>The dynamic configuration is stored in a special znode
+              ZooDefs.CONFIG_NODE = /zookeeper/config. This node by default is read only
+              for all users, except super user and users that's explicitly configured for write
+              access.
+            </para>
+
+            <para>Clients that need to use reconfig commands or reconfig API should be configured as users
+              that have write access to CONFIG_NODE. By default, only the super user has full control including
+              write access to CONFIG_NODE. Additional users can be granted write access through superuser
+              by setting an ACL that has write permission associated with specified user.
+            </para>
+
+            <para>A few examples of how to setup ACLs and use reconfiguration API with authentication can be found in
+              ReconfigExceptionTest.java and TestReconfigServer.cc.</para>
+          </listitem>
+        </varlistentry>
+
+        <varlistentry>
+          <term><emphasis role="bold">Authentication</emphasis></term>
+
+          <listitem>
+            <para>Authentication of users is orthogonal to the access control and is delegated to
+              existing authentication mechanism supported by ZooKeeper's pluggable authentication schemes.
+              See <ulink
+                      url="https://cwiki.apache.org/confluence/display/ZOOKEEPER/Zookeeper+and+SASL"
+              >ZooKeeper and SASL</ulink> for more details on this topic.
+            </para>
+          </listitem>
+        </varlistentry>
+
+        <varlistentry>
+          <term><emphasis role="bold">Disable ACL check</emphasis></term>
+          <listitem>
+            <para>
+              ZooKeeper supports <ulink
+                    url="zookeeperAdmin.html#sc_authOptions">"skipACL"</ulink> option such that ACL
+              check will be completely skipped, if skipACL is set to "yes". In such cases any unauthenticated
+              users can use reconfig API.
+            </para>
+          </listitem>
+        </varlistentry>
+      </variablelist>
+    </section>
+
     <section id="sc_reconfig_retrieving">
       <title>Retrieving the current dynamic configuration</title>
       <para>The dynamic configuration is stored in a special znode

src/java/main/org/apache/zookeeper/ClientCnxn.java

@@ -1523,14 +1523,14 @@ public class ClientCnxn {
         sendThread.sendPacket(p);
     }
 
-    Packet queuePacket(RequestHeader h, ReplyHeader r, Record request,
+    public Packet queuePacket(RequestHeader h, ReplyHeader r, Record request,
             Record response, AsyncCallback cb, String clientPath,
             String serverPath, Object ctx, WatchRegistration watchRegistration) {
         return queuePacket(h, r, request, response, cb, clientPath, serverPath,
                 ctx, watchRegistration, null);
     }
 
-    Packet queuePacket(RequestHeader h, ReplyHeader r, Record request,
+    public Packet queuePacket(RequestHeader h, ReplyHeader r, Record request,
             Record response, AsyncCallback cb, String clientPath,
             String serverPath, Object ctx, WatchRegistration watchRegistration,
             WatchDeregistration watchDeregistration) {

src/java/main/org/apache/zookeeper/KeeperException.java

@@ -139,6 +139,8 @@ public abstract class KeeperException extends Exception {
                 return new EphemeralOnLocalSessionException();
             case NOWATCHER:
                 return new NoWatcherException();
+            case RECONFIGDISABLED:
+                return new ReconfigDisabledException();
             case OK:
             default:
                 throw new IllegalArgumentException("Invalid exception code");
@@ -384,7 +386,9 @@ public abstract class KeeperException extends Exception {
         /** Attempt to create ephemeral node on a local session */
         EPHEMERALONLOCALSESSION (EphemeralOnLocalSession),
         /** Attempts to remove a non-existing watcher */
-        NOWATCHER (-121);
+        NOWATCHER (-121),
+        /** Attempts to perform a reconfiguration operation when reconfiguration feature is disabled. */
+        RECONFIGDISABLED(-123);
 
         private static final Map<Integer,Code> lookup
             = new HashMap<Integer,Code>();
@@ -469,6 +473,8 @@ public abstract class KeeperException extends Exception {
                 return "Ephemeral node on local session";
             case NOWATCHER:
                 return "No such watcher";
+            case RECONFIGDISABLED:
+                return "Reconfig is disabled";
             default:
                 return "Unknown error " + code;
         }
@@ -515,7 +521,7 @@ public abstract class KeeperException extends Exception {
 
     @Override
     public String getMessage() {
-        if (path == null) {
+        if (path == null || path.isEmpty()) {
             return "KeeperErrorCode = " + getCodeMessage(code);
         }
         return "KeeperErrorCode = " + getCodeMessage(code) + " for " + path;
@@ -795,4 +801,14 @@ public abstract class KeeperException extends Exception {
             super(Code.NOWATCHER, path);
         }
     }
+
+    /**
+     * @see Code#RECONFIGDISABLED
+     */
+    public static class ReconfigDisabledException extends KeeperException {
+        public ReconfigDisabledException() { super(Code.RECONFIGDISABLED); }
+        public ReconfigDisabledException(String path) {
+            super(Code.RECONFIGDISABLED, path);
+        }
+    }
 }

src/java/main/org/apache/zookeeper/ZooKeeper.java

@@ -39,7 +39,6 @@ import org.apache.zookeeper.client.StaticHostProvider;
 import org.apache.zookeeper.client.ZKClientConfig;
 import org.apache.zookeeper.client.ZooKeeperSaslClient;
 import org.apache.zookeeper.common.PathUtils;
-import org.apache.zookeeper.common.StringUtils;
 import org.apache.zookeeper.data.ACL;
 import org.apache.zookeeper.data.Stat;
 import org.apache.zookeeper.proto.CheckWatchesRequest;
@@ -57,7 +56,6 @@ import org.apache.zookeeper.proto.GetChildrenRequest;
 import org.apache.zookeeper.proto.GetChildrenResponse;
 import org.apache.zookeeper.proto.GetDataRequest;
 import org.apache.zookeeper.proto.GetDataResponse;
-import org.apache.zookeeper.proto.ReconfigRequest;
 import org.apache.zookeeper.proto.RemoveWatchesRequest;
 import org.apache.zookeeper.proto.ReplyHeader;
 import org.apache.zookeeper.proto.RequestHeader;
@@ -156,7 +154,7 @@ public class ZooKeeper {
         Environment.logEnv("Client environment:", LOG);
     }
 
-    private final HostProvider hostProvider;
+    protected final HostProvider hostProvider;
 
     /**
      * This function allows a client to update the connection string by providing 
@@ -215,7 +213,7 @@ public class ZooKeeper {
         return cnxn.zooKeeperSaslClient;
     }
 
-    private final ZKWatchManager watchManager;
+    protected final ZKWatchManager watchManager;
 
     private final ZKClientConfig clientConfig;
 
@@ -223,19 +221,19 @@ public class ZooKeeper {
         return clientConfig;
     }
 
-    List<String> getDataWatches() {
+    protected List<String> getDataWatches() {
         synchronized(watchManager.dataWatches) {
             List<String> rc = new ArrayList<String>(watchManager.dataWatches.keySet());
             return rc;
         }
     }
-    List<String> getExistWatches() {
+    protected List<String> getExistWatches() {
         synchronized(watchManager.existWatches) {
             List<String> rc =  new ArrayList<String>(watchManager.existWatches.keySet());
             return rc;
         }
     }
-    List<String> getChildWatches() {
+    protected List<String> getChildWatches() {
         synchronized(watchManager.childWatches) {
             List<String> rc = new ArrayList<String>(watchManager.childWatches.keySet());
             return rc;
@@ -262,7 +260,7 @@ public class ZooKeeper {
             this.disableAutoWatchReset = disableAutoWatchReset;
         }
 
-        private volatile Watcher defaultWatcher;
+        protected volatile Watcher defaultWatcher;
 
         final private void addTo(Set<Watcher> from, Set<Watcher> to) {
             if (from != null) {
@@ -529,7 +527,7 @@ public class ZooKeeper {
     /**
      * Register a watcher for a particular path.
      */
-    abstract class WatchRegistration {
+    public abstract class WatchRegistration {
         private Watcher watcher;
         private String clientPath;
         public WatchRegistration(Watcher watcher, String clientPath)
@@ -2177,85 +2175,6 @@ public class ZooKeeper {
     public void getConfig(boolean watch, DataCallback cb, Object ctx) {
         getConfig(watch ? watchManager.defaultWatcher : null, cb, ctx);
     }
-    
-    /**
-     * Reconfigure - add/remove servers. Return the new configuration.
-     * @param joiningServers
-     *                a comma separated list of servers being added (incremental reconfiguration)
-     * @param leavingServers
-     *                a comma separated list of servers being removed (incremental reconfiguration)
-     * @param newMembers
-     *                a comma separated list of new membership (non-incremental reconfiguration)
-     * @param fromConfig
-     *                version of the current configuration (optional - causes reconfiguration to throw an exception if configuration is no longer current)
-     * @param stat the stat of /zookeeper/config znode will be copied to this
-     *             parameter if not null.
-     * @return new configuration
-     * @throws InterruptedException If the server transaction is interrupted.
-     * @throws KeeperException If the server signals an error with a non-zero error code.     
-     */
-    public byte[] reconfig(String joiningServers, String leavingServers, String newMembers, long fromConfig, Stat stat) throws KeeperException, InterruptedException
-    {
-        RequestHeader h = new RequestHeader();
-        h.setType(ZooDefs.OpCode.reconfig);       
-        ReconfigRequest request = new ReconfigRequest(joiningServers, leavingServers, newMembers, fromConfig);        
-        GetDataResponse response = new GetDataResponse();       
-        ReplyHeader r = cnxn.submitRequest(h, request, response, null);
-        if (r.getErr() != 0) {
-            throw KeeperException.create(KeeperException.Code.get(r.getErr()), "");
-        }
-        if (stat != null) {
-            DataTree.copyStat(response.getStat(), stat);
-        }
-        return response.getData();
-    }
-
-    /**
-     * Convenience wrapper around reconfig that takes Lists of strings instead of comma-separated servers.
-     *
-     * @see #reconfig
-     *
-     */
-    public byte[] reconfig(List<String> joiningServers, List<String> leavingServers, List<String> newMembers, long fromConfig, Stat stat) throws KeeperException, InterruptedException
-    {
-        return reconfig(StringUtils.joinStrings(joiningServers, ","), 
-        		StringUtils.joinStrings(leavingServers, ","), 
-        		StringUtils.joinStrings(newMembers, ","), 
-        		fromConfig, stat);
-    }
-
-    /**
-     * The Asynchronous version of reconfig. 
-     *
-     * @see #reconfig
-     *      
-     **/
-    public void reconfig(String joiningServers, String leavingServers,
-        String newMembers, long fromConfig, DataCallback cb, Object ctx)
-    {
-        RequestHeader h = new RequestHeader();
-        h.setType(ZooDefs.OpCode.reconfig);       
-        ReconfigRequest request = new ReconfigRequest(joiningServers, leavingServers, newMembers, fromConfig);
-        GetDataResponse response = new GetDataResponse();
-        cnxn.queuePacket(h, new ReplyHeader(), request, response, cb,
-               ZooDefs.CONFIG_NODE, ZooDefs.CONFIG_NODE, ctx, null);
-    }
- 
-    /**
-     * Convenience wrapper around asynchronous reconfig that takes Lists of strings instead of comma-separated servers.
-     *
-     * @see #reconfig
-     *
-     */
-    public void reconfig(List<String> joiningServers,
-        List<String> leavingServers, List<String> newMembers, long fromConfig,
-        DataCallback cb, Object ctx)
-    {
-        reconfig(StringUtils.joinStrings(joiningServers, ","), 
-        		StringUtils.joinStrings(leavingServers, ","), 
-        		StringUtils.joinStrings(newMembers, ","), 
-        		fromConfig, cb, ctx);
-    }
    
     /**
      * Set the data for the node of the given path if such a node exists and the

src/java/main/org/apache/zookeeper/ZooKeeperMain.java

@@ -66,6 +66,7 @@ import org.apache.zookeeper.cli.SetQuotaCommand;
 import org.apache.zookeeper.cli.StatCommand;
 import org.apache.zookeeper.cli.SyncCommand;
 import org.apache.zookeeper.client.ZKClientConfig;
+import org.apache.zookeeper.admin.ZooKeeperAdmin;
 
 /**
  * The command line client to ZooKeeper.
@@ -275,15 +276,14 @@ public class ZooKeeperMain {
         if (zk != null && zk.getState().isAlive()) {
             zk.close();
         }
+
         host = newHost;
         boolean readOnly = cl.getOption("readonly") != null;
         if (cl.getOption("secure") != null) {
             System.setProperty(ZKClientConfig.SECURE_CLIENT, "true");
             System.out.println("Secure connection is enabled");
         }
-        zk = new ZooKeeper(host,
-                 Integer.parseInt(cl.getOption("timeout")),
-                 new MyWatcher(), readOnly);
+        zk = new ZooKeeperAdmin(host, Integer.parseInt(cl.getOption("timeout")), new MyWatcher(), readOnly);
     }
     
     public static void main(String args[]) throws CliException, IOException, InterruptedException
@@ -296,8 +296,6 @@ public class ZooKeeperMain {
         cl.parseOptions(args);
         System.out.println("Connecting to " + cl.getOption("server"));
         connectToZK(cl.getOption("server"));
-        //zk = new ZooKeeper(cl.getOption("server"),
-//                Integer.parseInt(cl.getOption("timeout")), new MyWatcher());
     }
 
     public ZooKeeperMain(ZooKeeper zk) {

src/java/main/org/apache/zookeeper/admin/ZooKeeperAdmin.java

@@ -0,0 +1,250 @@
+/**
+ * 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.admin;
+
+import java.io.IOException;
+import java.util.List;
+
+import org.apache.zookeeper.ZooKeeper;
+import org.apache.zookeeper.ZooDefs;
+import org.apache.zookeeper.Watcher;
+import org.apache.zookeeper.KeeperException;
+import org.apache.zookeeper.AsyncCallback.DataCallback;
+import org.apache.zookeeper.client.ZKClientConfig;
+import org.apache.zookeeper.common.StringUtils;
+import org.apache.zookeeper.data.Stat;
+import org.apache.zookeeper.proto.GetDataResponse;
+import org.apache.zookeeper.proto.ReconfigRequest;
+import org.apache.zookeeper.proto.ReplyHeader;
+import org.apache.zookeeper.proto.RequestHeader;
+import org.apache.zookeeper.server.DataTree;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * This is the main class for ZooKeeperAdmin client library.
+ * This library is used to perform cluster administration tasks,
+ * such as reconfigure cluster membership. The ZooKeeperAdmin class
+ * inherits ZooKeeper and has similar usage pattern as ZooKeeper class.
+ * Please check {@link ZooKeeper} class document for more details.
+ *
+ * @since 3.5.3
+ */
+public class ZooKeeperAdmin extends ZooKeeper {
+    private static final Logger LOG = LoggerFactory.getLogger(ZooKeeperAdmin.class);
+
+    /**
+     * Create a ZooKeeperAdmin object which is used to perform dynamic reconfiguration
+     * operations.
+     *
+     * @param connectString
+     *            comma separated host:port pairs, each corresponding to a zk
+     *            server. e.g. "127.0.0.1:3000,127.0.0.1:3001,127.0.0.1:3002" If
+     *            the optional chroot suffix is used the example would look
+     *            like: "127.0.0.1:3000,127.0.0.1:3001,127.0.0.1:3002/app/a"
+     *            where the client would be rooted at "/app/a" and all paths
+     *            would be relative to this root - ie getting/setting/etc...
+     *            "/foo/bar" would result in operations being run on
+     *            "/app/a/foo/bar" (from the server perspective).
+     * @param sessionTimeout
+     *            session timeout in milliseconds
+     * @param watcher
+     *            a watcher object which will be notified of state changes, may
+     *            also be notified for node events
+     *
+     * @throws IOException
+     *             in cases of network failure
+     * @throws IllegalArgumentException
+     *             if an invalid chroot path is specified
+     *
+     * @see ZooKeeper#ZooKeeper(String, int, Watcher)
+     *
+     */
+    public ZooKeeperAdmin(String connectString, int sessionTimeout, Watcher watcher)
+        throws IOException {
+        super(connectString, sessionTimeout, watcher);
+    }
+
+    /**
+     * Create a ZooKeeperAdmin object which is used to perform dynamic reconfiguration
+     * operations.
+     *
+     * @param connectString
+     *            comma separated host:port pairs, each corresponding to a zk
+     *            server. e.g. "127.0.0.1:3000,127.0.0.1:3001,127.0.0.1:3002" If
+     *            the optional chroot suffix is used the example would look
+     *            like: "127.0.0.1:3000,127.0.0.1:3001,127.0.0.1:3002/app/a"
+     *            where the client would be rooted at "/app/a" and all paths
+     *            would be relative to this root - ie getting/setting/etc...
+     *            "/foo/bar" would result in operations being run on
+     *            "/app/a/foo/bar" (from the server perspective).
+     * @param sessionTimeout
+     *            session timeout in milliseconds
+     * @param watcher
+     *            a watcher object which will be notified of state changes, may
+     *            also be notified for node events
+     * @param conf
+     *            passing this conf object gives each client the flexibility of
+     *            configuring properties differently compared to other instances
+     *
+     * @throws IOException
+     *             in cases of network failure
+     * @throws IllegalArgumentException
+     *             if an invalid chroot path is specified
+     *
+     * @see ZooKeeper#ZooKeeper(String, int, Watcher, ZKClientConfig)
+     */
+    public ZooKeeperAdmin(String connectString, int sessionTimeout, Watcher watcher,
+            ZKClientConfig conf) throws IOException {
+        super(connectString, sessionTimeout, watcher, conf);
+    }
+
+    /**
+     * Create a ZooKeeperAdmin object which is used to perform dynamic reconfiguration
+     * operations.
+     *
+     * @param connectString
+     *            comma separated host:port pairs, each corresponding to a zk
+     *            server. e.g. "127.0.0.1:3000,127.0.0.1:3001,127.0.0.1:3002" If
+     *            the optional chroot suffix is used the example would look
+     *            like: "127.0.0.1:3000,127.0.0.1:3001,127.0.0.1:3002/app/a"
+     *            where the client would be rooted at "/app/a" and all paths
+     *            would be relative to this root - ie getting/setting/etc...
+     *            "/foo/bar" would result in operations being run on
+     *            "/app/a/foo/bar" (from the server perspective).
+     * @param sessionTimeout
+     *            session timeout in milliseconds
+     * @param watcher
+     *            a watcher object which will be notified of state changes, may
+     *            also be notified for node events
+     * @param canBeReadOnly
+     *            whether the created client is allowed to go to
+     *            read-only mode in case of partitioning. Read-only mode
+     *            basically means that if the client can't find any majority
+     *            servers but there's partitioned server it could reach, it
+     *            connects to one in read-only mode, i.e. read requests are
+     *            allowed while write requests are not. It continues seeking for
+     *            majority in the background.
+     *
+     * @throws IOException
+     *             in cases of network failure
+     * @throws IllegalArgumentException
+     *             if an invalid chroot path is specified
+     *
+     * @see ZooKeeper#ZooKeeper(String, int, Watcher, boolean)
+     */
+    public ZooKeeperAdmin(String connectString, int sessionTimeout, Watcher watcher,
+                     boolean canBeReadOnly) throws IOException {
+        super(connectString, sessionTimeout, watcher, canBeReadOnly);
+    }
+
+    /**
+     * Reconfigure - add/remove servers. Return the new configuration.
+     * @param joiningServers
+     *                a comma separated list of servers being added (incremental reconfiguration)
+     * @param leavingServers
+     *                a comma separated list of servers being removed (incremental reconfiguration)
+     * @param newMembers
+     *                a comma separated list of new membership (non-incremental reconfiguration)
+     * @param fromConfig
+     *                version of the current configuration
+     *                (optional - causes reconfiguration to throw an exception if configuration is no longer current)
+     * @param stat the stat of /zookeeper/config znode will be copied to this
+     *             parameter if not null.
+     * @return new configuration
+     * @throws InterruptedException If the server transaction is interrupted.
+     * @throws KeeperException If the server signals an error with a non-zero error code.
+     */
+    public byte[] reconfig(String joiningServers, String leavingServers,
+                           String newMembers, long fromConfig, Stat stat) throws KeeperException, InterruptedException {
+        RequestHeader h = new RequestHeader();
+        h.setType(ZooDefs.OpCode.reconfig);
+        ReconfigRequest request = new ReconfigRequest(joiningServers, leavingServers, newMembers, fromConfig);
+        GetDataResponse response = new GetDataResponse();
+        ReplyHeader r = cnxn.submitRequest(h, request, response, null);
+        if (r.getErr() != 0) {
+            throw KeeperException.create(KeeperException.Code.get(r.getErr()), "");
+        }
+        if (stat != null) {
+            DataTree.copyStat(response.getStat(), stat);
+        }
+        return response.getData();
+    }
+
+    /**
+     * Convenience wrapper around reconfig that takes Lists of strings instead of comma-separated servers.
+     *
+     * @see #reconfig
+     *
+     */
+    public byte[] reconfig(List<String> joiningServers, List<String> leavingServers,
+                           List<String> newMembers, long fromConfig,
+                           Stat stat) throws KeeperException, InterruptedException {
+        return reconfig(StringUtils.joinStrings(joiningServers, ","),
+                        StringUtils.joinStrings(leavingServers, ","),
+                        StringUtils.joinStrings(newMembers, ","),
+                        fromConfig, stat);
+    }
+
+    /**
+     * The Asynchronous version of reconfig.
+     *
+     * @see #reconfig
+     *
+     **/
+    public void reconfig(String joiningServers, String leavingServers,
+        String newMembers, long fromConfig, DataCallback cb, Object ctx) {
+        RequestHeader h = new RequestHeader();
+        h.setType(ZooDefs.OpCode.reconfig);
+        ReconfigRequest request = new ReconfigRequest(joiningServers, leavingServers, newMembers, fromConfig);
+        GetDataResponse response = new GetDataResponse();
+        cnxn.queuePacket(h, new ReplyHeader(), request, response, cb,
+               ZooDefs.CONFIG_NODE, ZooDefs.CONFIG_NODE, ctx, null);
+    }
+
+    /**
+     * Convenience wrapper around asynchronous reconfig that takes Lists of strings instead of comma-separated servers.
+     *
+     * @see #reconfig
+     *
+     */
+    public void reconfig(List<String> joiningServers,
+        List<String> leavingServers, List<String> newMembers, long fromConfig,
+        DataCallback cb, Object ctx) {
+        reconfig(StringUtils.joinStrings(joiningServers, ","),
+                 StringUtils.joinStrings(leavingServers, ","),
+                 StringUtils.joinStrings(newMembers, ","),
+                 fromConfig, cb, ctx);
+    }
+
+    /**
+     * String representation of this ZooKeeperAdmin client. Suitable for things
+     * like logging.
+     *
+     * Do NOT count on the format of this string, it may change without
+     * warning.
+     *
+     * @since 3.5.3
+     */
+    @Override
+    public String toString() {
+        return super.toString();
+    }
+}

src/java/main/org/apache/zookeeper/cli/CliCommand.java

@@ -25,7 +25,6 @@ import org.apache.zookeeper.ZooKeeper;
  * base class for all CLI commands
  */
 abstract public class CliCommand {
-
     protected ZooKeeper zk;
     protected PrintStream out;
     protected PrintStream err;
@@ -63,7 +62,7 @@ abstract public class CliCommand {
 
     /**
      * set the zookeper instance
-     * @param zk the zookeper instance
+     * @param zk the ZooKeeper instance.
      */
     public void setZk(ZooKeeper zk) {
         this.zk = zk;

src/java/main/org/apache/zookeeper/cli/ReconfigCommand.java

@@ -18,12 +18,11 @@
 package org.apache.zookeeper.cli;
 
 import java.io.FileInputStream;
-import java.util.ArrayList;
-import java.util.List;
 import java.util.Properties;
 
 import org.apache.commons.cli.*;
 import org.apache.zookeeper.KeeperException;
+import org.apache.zookeeper.admin.ZooKeeperAdmin;
 import org.apache.zookeeper.data.Stat;
 import org.apache.zookeeper.server.quorum.QuorumPeerConfig;
 
@@ -146,7 +145,16 @@ public class ReconfigCommand extends CliCommand {
     public boolean exec() throws CliException {
         try {
             Stat stat = new Stat();
-            byte[] curConfig = zk.reconfig(joining,
+            if (!(zk instanceof ZooKeeperAdmin)) {
+                // This should never happen when executing reconfig command line,
+                // because it is guaranteed that we have a ZooKeeperAdmin instance ready
+                // to use in CliCommand stack.
+                // The only exception would be in test code where clients can directly set
+                // ZooKeeper object to ZooKeeperMain.
+                return false;
+            }
+
+            byte[] curConfig = ((ZooKeeperAdmin)zk).reconfig(joining,
                     leaving, members, version, stat);
             out.println("Committed new configuration:\n" + new String(curConfig));
             

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

@@ -245,15 +245,23 @@ public class DataTree {
         addConfigNode();
     }
 
-     public void addConfigNode() {
-    	 DataNode zookeeperZnode = nodes.get(procZookeeper);
-         if (zookeeperZnode!=null) { // should always be the case
-        	 zookeeperZnode.addChild(configChildZookeeper);
-         } else {
-        	 LOG.error("There's no /zookeeper znode - this should never happen");
-         }
-         nodes.put(configZookeeper, configDataNode);   
-     }
+    public void addConfigNode() {
+        DataNode zookeeperZnode = nodes.get(procZookeeper);
+        if (zookeeperZnode != null) { // should always be the case
+            zookeeperZnode.addChild(configChildZookeeper);
+        } else {
+            assert false : "There's no /zookeeper znode - this should never happen.";
+        }
+
+        nodes.put(configZookeeper, configDataNode);
+        try {
+            // Reconfig node is access controlled by default (ZOOKEEPER-2014).
+            setACL(configZookeeper, ZooDefs.Ids.READ_ACL_UNSAFE, -1);
+        } catch (KeeperException.NoNodeException e) {
+            assert false : "There's no " + configZookeeper +
+                    " znode - this should never happen.";
+        }
+    }
 
     /**
      * is the path one of the special paths owned by zookeeper.

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

@@ -425,6 +425,15 @@ public class PrepRequestProcessor extends ZooKeeperCriticalThread implements
                 addChangeRecord(nodeRecord);
                 break;
             case OpCode.reconfig:
+                if (!QuorumPeerConfig.isReconfigEnabled()) {
+                    LOG.error("Reconfig operation requested but reconfig feature is disabled.");
+                    throw new KeeperException.ReconfigDisabledException();
+                }
+
+                if (skipACL) {
+                    LOG.warn("skipACL is set, reconfig operation will skip ACL checks!");
+                }
+
                 zks.sessionTracker.checkSession(request.sessionId, request.getOwner());
                 ReconfigRequest reconfigRequest = (ReconfigRequest)record; 
                 LeaderZooKeeperServer lzks;

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

@@ -169,7 +169,8 @@ public class ZooKeeperServer implements SessionExpirer, ServerStats.Provider {
      * @param tickTime the ticktime for the server
      * @throws IOException
      */
-    public ZooKeeperServer(FileTxnSnapLog txnLogFactory, int tickTime) throws IOException {
+    public ZooKeeperServer(FileTxnSnapLog txnLogFactory, int tickTime)
+            throws IOException {
         this(txnLogFactory, tickTime, -1, -1, new ZKDatabase(txnLogFactory));
     }
 

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

@@ -110,7 +110,8 @@ public class ZooKeeperServerMain {
      * @throws IOException
      * @throws AdminServerException
      */
-    public void runFromConfig(ServerConfig config) throws IOException, AdminServerException {
+    public void runFromConfig(ServerConfig config)
+            throws IOException, AdminServerException {
         LOG.info("Starting server");
         FileTxnSnapLog txnLog = null;
         try {

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

@@ -60,6 +60,7 @@ public class QuorumPeerConfig {
     public static final String nextDynamicConfigFileSuffix = ".dynamic.next";
 
     private static boolean standaloneEnabled = true;
+    private static boolean reconfigEnabled = false;
 
     protected InetSocketAddress clientPortAddress;
     protected InetSocketAddress secureClientPortAddress;
@@ -279,7 +280,15 @@ public class QuorumPeerConfig {
                 } else if (value.toLowerCase().equals("false")) {
                     setStandaloneEnabled(false);
                 } else {
-                    throw new ConfigException("Invalid option for standalone mode. Choose 'true' or 'false.'");
+                    throw new ConfigException("Invalid option " + value + " for standalone mode. Choose 'true' or 'false.'");
+                }
+            } else if (key.equals("reconfigEnabled")) {
+                if (value.toLowerCase().equals("true")) {
+                    setReconfigEnabled(true);
+                } else if (value.toLowerCase().equals("false")) {
+                    setReconfigEnabled(false);
+                } else {
+                    throw new ConfigException("Invalid option " + value + " for reconfigEnabled flag. Choose 'true' or 'false.'");
                 }
             } else if ((key.startsWith("server.") || key.startsWith("group") || key.startsWith("weight")) && zkProp.containsKey("dynamicConfigFile")) {
                 throw new ConfigException("parameter: " + key + " must be in a separate dynamic config file");
@@ -732,7 +741,13 @@ public class QuorumPeerConfig {
     }
     
     public static void setStandaloneEnabled(boolean enabled) {
-	standaloneEnabled = enabled;
+        standaloneEnabled = enabled;
+    }
+
+    public static boolean isReconfigEnabled() { return reconfigEnabled; }
+
+    public static void setReconfigEnabled(boolean enabled) {
+        reconfigEnabled = enabled;
     }
 
 }

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

@@ -126,7 +126,9 @@ public class QuorumPeerMain {
         }
     }
 
-    public void runFromConfig(QuorumPeerConfig config) throws IOException, AdminServerException {
+    public void runFromConfig(QuorumPeerConfig config)
+            throws IOException, AdminServerException
+    {
       try {
           ManagedUtil.registerLog4jMBeans();
       } catch (JMException e) {

src/java/systest/org/apache/zookeeper/test/system/BaseSysTest.java

@@ -25,7 +25,6 @@ import java.net.InetSocketAddress;
 import java.net.UnknownHostException;
 import java.util.HashMap;
 
-
 import org.apache.zookeeper.WatchedEvent;
 import org.apache.zookeeper.Watcher;
 import org.apache.zookeeper.ZooKeeper;

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

@@ -25,10 +25,11 @@ import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.TimeUnit;
 
 import org.apache.jute.Record;
+import org.apache.zookeeper.admin.ZooKeeperAdmin;
 import org.apache.zookeeper.proto.ReplyHeader;
 import org.apache.zookeeper.proto.RequestHeader;
 
-public class TestableZooKeeper extends ZooKeeper {
+public class TestableZooKeeper extends ZooKeeperAdmin {
 
     public TestableZooKeeper(String host, int sessionTimeout,
             Watcher watcher) throws IOException {

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

@@ -26,12 +26,10 @@ import org.apache.zookeeper.WatchedEvent;
 import org.apache.zookeeper.Watcher;
 import org.apache.zookeeper.ZKTestCase;
 import org.apache.zookeeper.data.Stat;
-import org.apache.zookeeper.server.DataTree;
 import org.junit.After;
 import org.junit.Assert;
 import org.junit.Before;
 import org.junit.Test;
-import org.apache.zookeeper.server.DataNode;
 import java.io.ByteArrayInputStream;
 import java.io.ByteArrayOutputStream;
 import java.io.DataOutputStream;
@@ -200,29 +198,34 @@ public class DataTreeTest extends ZKTestCase {
         BinaryOutputArchive oa = new BinaryOutputArchive(out) {
             @Override
             public void writeRecord(Record r, String tag) throws IOException {
-                DataNode node = (DataNode) r;
-                if (node.data.length == 1 && node.data[0] == 42) {
-                    final Semaphore semaphore = new Semaphore(0);
-                    new Thread(new Runnable() {
-                        @Override
-                        public void run() {
-                            synchronized (markerNode) {
-                                //When we lock markerNode, allow writeRecord to continue
-                                semaphore.release();
+                // Need check if the record is a DataNode instance because of changes in ZOOKEEPER-2014
+                // which adds default ACL to config node.
+                if (r instanceof DataNode) {
+                    DataNode node = (DataNode) r;
+                    if (node.data.length == 1 && node.data[0] == 42) {
+                        final Semaphore semaphore = new Semaphore(0);
+                        new Thread(new Runnable() {
+                            @Override
+                            public void run() {
+                                synchronized (markerNode) {
+                                    //When we lock markerNode, allow writeRecord to continue
+                                    semaphore.release();
+                                }
                             }
+                        }).start();
+
+                        try {
+                            boolean acquired = semaphore.tryAcquire(30, TimeUnit.SECONDS);
+                            //This is the real assertion - could another thread lock
+                            //the DataNode we're currently writing
+                            Assert.assertTrue("Couldn't acquire a lock on the DataNode while we were calling tree.serialize", acquired);
+                        } catch (InterruptedException e1) {
+                            throw new RuntimeException(e1);
                         }
-                    }).start();
-
-                    try {
-                        boolean acquired = semaphore.tryAcquire(30, TimeUnit.SECONDS);
-                        //This is the real assertion - could another thread lock
-                        //the DataNode we're currently writing
-                        Assert.assertTrue("Couldn't acquire a lock on the DataNode while we were calling tree.serialize", acquired);
-                    } catch (InterruptedException e1) {
-                        throw new RuntimeException(e1);
+                        ranTestCase.set(true);
                     }
-                    ranTestCase.set(true);
                 }
+
                 super.writeRecord(r, tag);
             }
         };

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

@@ -49,7 +49,8 @@ public class LearnerTest extends ZKTestCase {
 
         Learner learner;
 
-        public SimpleLearnerZooKeeperServer(FileTxnSnapLog ftsl, QuorumPeer self) throws IOException {
+        public SimpleLearnerZooKeeperServer(FileTxnSnapLog ftsl, QuorumPeer self)
+                throws IOException {
             super(ftsl, 2000, 2000, 2000, new ZKDatabase(ftsl), self);
         }
 

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

@@ -151,7 +151,8 @@ public class RaceConditionTest extends QuorumPeerTestBase {
         }
 
         public CustomQuorumPeer(Map<Long, QuorumServer> quorumPeers, File snapDir, File logDir, int clientPort,
-                int electionAlg, long myid, int tickTime, int initLimit, int syncLimit) throws IOException {
+                int electionAlg, long myid, int tickTime, int initLimit, int syncLimit)
+                throws IOException {
             super(quorumPeers, snapDir, logDir, electionAlg, myid, tickTime, initLimit, syncLimit, false,
                     ServerCnxnFactory.createFactory(new InetSocketAddress(clientPort), -1), new QuorumMaj(quorumPeers));
         }
@@ -234,7 +235,8 @@ public class RaceConditionTest extends QuorumPeerTestBase {
 
     private static class MockTestQPMain extends TestQPMain {
         @Override
-        public void runFromConfig(QuorumPeerConfig config) throws IOException, AdminServerException {
+        public void runFromConfig(QuorumPeerConfig config)
+                throws IOException, AdminServerException {
             quorumPeer = new CustomQuorumPeer(config.getQuorumVerifier().getAllMembers(), config.getDataDir(),
                     config.getDataLogDir(), config.getClientPortAddress().getPort(), config.getElectionAlg(),
                     config.getServerId(), config.getTickTime(), config.getInitLimit(), config.getSyncLimit());

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

@@ -20,6 +20,7 @@ package org.apache.zookeeper.server.quorum;
 
 import org.apache.zookeeper.PortAssignment;
 import org.apache.zookeeper.ZooKeeper;
+import org.apache.zookeeper.admin.ZooKeeperAdmin;
 import org.apache.zookeeper.common.StringUtils;
 import org.apache.zookeeper.test.ClientBase;
 import org.apache.zookeeper.test.ReconfigTest;
@@ -61,6 +62,8 @@ public class ReconfigBackupTest extends QuorumPeerTestBase {
     @Before
     public void setup() {
         ClientBase.setupTestEnv();
+        System.setProperty("zookeeper.DigestAuthenticationProvider.superDigest",
+                "super:D/InIHSb7yEEbrWz8b9l71RjZJU="/* password is 'test'*/);
     }
 
     /**
@@ -80,6 +83,7 @@ public class ReconfigBackupTest extends QuorumPeerTestBase {
                     + clientPorts[i];
             sb.append(server + "\n");
         }
+
         String currentQuorumCfgSection = sb.toString();
 
         MainThread mt[] = new MainThread[SERVER_COUNT];
@@ -145,14 +149,16 @@ public class ReconfigBackupTest extends QuorumPeerTestBase {
             oldServers.add(servers[i]);
             sb.append(servers[i] + "\n");
         }
+
         String quorumCfgSection = sb.toString();
 
         MainThread mt[] = new MainThread[NEW_SERVER_COUNT];
         ZooKeeper zk[] = new ZooKeeper[NEW_SERVER_COUNT];
+        ZooKeeperAdmin zkAdmin[] = new ZooKeeperAdmin[NEW_SERVER_COUNT];
 
         // start old cluster
         for (int i = 0; i < SERVER_COUNT; i++) {
-            mt[i] = new MainThread(i, clientPorts[i], quorumCfgSection);
+            mt[i] = new MainThread(i, clientPorts[i], quorumCfgSection, "reconfigEnabled=true\n");
             mt[i].start();
         }
 
@@ -164,6 +170,9 @@ public class ReconfigBackupTest extends QuorumPeerTestBase {
                     ClientBase.waitForServerUp("127.0.0.1:" + clientPorts[i],
                             CONNECTION_TIMEOUT));
             zk[i] = ClientBase.createZKClient("127.0.0.1:" + clientPorts[i]);
+            zkAdmin[i] = new ZooKeeperAdmin("127.0.0.1:" + clientPorts[i],
+                    ClientBase.CONNECTION_TIMEOUT, this);
+            zkAdmin[i].addAuthInfo("digest", "super:test".getBytes());
 
             Properties cfg = ReconfigLegacyTest.readPropertiesFromFile(mt[i].confFile);
             String filename = cfg.getProperty("dynamicConfigFile", "");
@@ -186,7 +195,7 @@ public class ReconfigBackupTest extends QuorumPeerTestBase {
             }
         }
 
-        ReconfigTest.reconfig(zk[1], null, null, newServers, -1);
+        ReconfigTest.reconfig(zkAdmin[1], null, null, newServers, -1);
 
         // start additional new servers
         for (int i = SERVER_COUNT; i < NEW_SERVER_COUNT; i++) {
@@ -230,6 +239,7 @@ public class ReconfigBackupTest extends QuorumPeerTestBase {
         for (int i = 0; i < SERVER_COUNT; i++) {
             mt[i].shutdown();
             zk[i].close();
+            zkAdmin[i].close();
         }
     }
 

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

@@ -30,6 +30,7 @@ import org.apache.zookeeper.CreateMode;
 import org.apache.zookeeper.PortAssignment;
 import org.apache.zookeeper.ZooDefs;
 import org.apache.zookeeper.ZooKeeper;
+import org.apache.zookeeper.admin.ZooKeeperAdmin;
 import org.apache.zookeeper.server.ServerCnxnFactory;
 import org.apache.zookeeper.server.admin.AdminServer.AdminServerException;
 import org.apache.zookeeper.server.persistence.FileTxnSnapLog;
@@ -38,6 +39,7 @@ import org.apache.zookeeper.test.ClientBase;
 import org.apache.zookeeper.test.ClientBase.CountdownWatcher;
 import org.junit.After;
 import org.junit.Assert;
+import org.junit.Before;
 import org.junit.Test;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -47,6 +49,13 @@ public class ReconfigDuringLeaderSyncTest extends QuorumPeerTestBase {
     private static int SERVER_COUNT = 3;
     private MainThread[] mt;
 
+    @Before
+    public void setup() {
+        System.setProperty("zookeeper.DigestAuthenticationProvider.superDigest",
+                "super:D/InIHSb7yEEbrWz8b9l71RjZJU="/* password is 'test'*/);
+        QuorumPeerConfig.setReconfigEnabled(true);
+    }
+
     /**
      * <pre>
      * Test case for https://issues.apache.org/jira/browse/ZOOKEEPER-2172.
@@ -86,8 +95,9 @@ public class ReconfigDuringLeaderSyncTest extends QuorumPeerTestBase {
                     ClientBase.waitForServerUp("127.0.0.1:" + clientPorts[i], CONNECTION_TIMEOUT));
         }
         CountdownWatcher watch = new CountdownWatcher();
-        ZooKeeper preReconfigClient = new ZooKeeper("127.0.0.1:" + clientPorts[0], ClientBase.CONNECTION_TIMEOUT,
-                watch);
+        ZooKeeperAdmin preReconfigClient = new ZooKeeperAdmin("127.0.0.1:" + clientPorts[0],
+                ClientBase.CONNECTION_TIMEOUT, watch);
+        preReconfigClient.addAuthInfo("digest", "super:test".getBytes());
         watch.waitForConnected(ClientBase.CONNECTION_TIMEOUT);
 
         // new server joining
@@ -198,7 +208,8 @@ public class ReconfigDuringLeaderSyncTest extends QuorumPeerTestBase {
         private boolean newLeaderMessage = false;
 
         public CustomQuorumPeer(Map<Long, QuorumServer> quorumPeers, File snapDir, File logDir, int clientPort,
-                int electionAlg, long myid, int tickTime, int initLimit, int syncLimit) throws IOException {
+                int electionAlg, long myid, int tickTime, int initLimit, int syncLimit)
+                throws IOException {
             super(quorumPeers, snapDir, logDir, electionAlg, myid, tickTime, initLimit, syncLimit, false,
                     ServerCnxnFactory.createFactory(new InetSocketAddress(clientPort), -1), new QuorumMaj(quorumPeers));
         }
@@ -241,7 +252,8 @@ public class ReconfigDuringLeaderSyncTest extends QuorumPeerTestBase {
 
     private static class MockTestQPMain extends TestQPMain {
         @Override
-        public void runFromConfig(QuorumPeerConfig config) throws IOException, AdminServerException {
+        public void runFromConfig(QuorumPeerConfig config)
+                throws IOException, AdminServerException {
             quorumPeer = new CustomQuorumPeer(config.getQuorumVerifier().getAllMembers(), config.getDataDir(),
                     config.getDataLogDir(), config.getClientPortAddress().getPort(), config.getElectionAlg(),
                     config.getServerId(), config.getTickTime(), config.getInitLimit(), config.getSyncLimit());

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

@@ -27,20 +27,28 @@ import java.util.List;
 
 import org.apache.zookeeper.KeeperException;
 import org.apache.zookeeper.KeeperException.NewConfigNoQuorum;
-import org.apache.zookeeper.PortAssignment;
 import org.apache.zookeeper.ZooKeeper;
+import org.apache.zookeeper.admin.ZooKeeperAdmin;
 import org.apache.zookeeper.data.Stat;
 import org.apache.zookeeper.test.ClientBase;
 import org.apache.zookeeper.test.QuorumUtil;
 import org.apache.zookeeper.test.ReconfigTest;
 import org.junit.After;
 import org.junit.Assert;
+import org.junit.Before;
 import org.junit.Test;
 
 public class ReconfigFailureCasesTest extends QuorumPeerTestBase {
 
     private QuorumUtil qu;
 
+    @Before
+    public void setup() {
+        QuorumPeerConfig.setReconfigEnabled(true);
+        System.setProperty("zookeeper.DigestAuthenticationProvider.superDigest",
+                "super:D/InIHSb7yEEbrWz8b9l71RjZJU="/* password is 'test'*/);
+    }
+
     @After
     public void tearDown() throws Exception {
         if (qu != null) {
@@ -57,6 +65,7 @@ public class ReconfigFailureCasesTest extends QuorumPeerTestBase {
         qu.disableJMXTest = true;
         qu.startAll();
         ZooKeeper[] zkArr = ReconfigTest.createHandles(qu);
+        ZooKeeperAdmin[] zkAdminArr = ReconfigTest.createAdminHandles(qu);
 
         ArrayList<String> members = new ArrayList<String>();
         members.add("group.1=3:4:5");
@@ -75,14 +84,14 @@ public class ReconfigFailureCasesTest extends QuorumPeerTestBase {
         }
 
         // Change the quorum system from majority to hierarchical.
-        ReconfigTest.reconfig(zkArr[1], null, null, members, -1);
+        ReconfigTest.reconfig(zkAdminArr[1], null, null, members, -1);
         ReconfigTest.testNormalOperation(zkArr[1], zkArr[2]);
 
         // Attempt an incremental reconfig.
         List<String> leavingServers = new ArrayList<String>();
         leavingServers.add("3");
         try {
-             zkArr[1].reconfig(null, leavingServers, null, -1, null);
+             zkAdminArr[1].reconfig(null, leavingServers, null, -1, null);
             Assert.fail("Reconfig should have failed since the current config isn't Majority QS");
         } catch (KeeperException.BadArgumentsException e) {
             // We expect this to happen.
@@ -90,7 +99,7 @@ public class ReconfigFailureCasesTest extends QuorumPeerTestBase {
             Assert.fail("Should have been BadArgumentsException!");
         }
 
-        ReconfigTest.closeAllHandles(zkArr);
+        ReconfigTest.closeAllHandles(zkArr, zkAdminArr);
     }
 
     /*
@@ -106,12 +115,13 @@ public class ReconfigFailureCasesTest extends QuorumPeerTestBase {
         qu.disableJMXTest = true;
         qu.startAll();
         ZooKeeper[] zkArr = ReconfigTest.createHandles(qu);
+        ZooKeeperAdmin[] zkAdminArr = ReconfigTest.createAdminHandles(qu);
 
         List<String> leavingServers = new ArrayList<String>();
         leavingServers.add("2");
         leavingServers.add("3");
         try {
-             zkArr[1].reconfig(null, leavingServers, null, -1, null);
+             zkAdminArr[1].reconfig(null, leavingServers, null, -1, null);
             Assert.fail("Reconfig should have failed since the current config version is not 8");
         } catch (KeeperException.BadArgumentsException e) {
             // We expect this to happen.
@@ -119,7 +129,7 @@ public class ReconfigFailureCasesTest extends QuorumPeerTestBase {
             Assert.fail("Should have been BadArgumentsException!");
         }
 
-        ReconfigTest.closeAllHandles(zkArr);
+        ReconfigTest.closeAllHandles(zkArr, zkAdminArr);
     }
 
     /*
@@ -132,11 +142,12 @@ public class ReconfigFailureCasesTest extends QuorumPeerTestBase {
         qu.disableJMXTest = true;
         qu.startAll();
         ZooKeeper[] zkArr = ReconfigTest.createHandles(qu);
+        ZooKeeperAdmin[] zkAdminArr = ReconfigTest.createAdminHandles(qu);
 
         List<String> leavingServers = new ArrayList<String>();
         leavingServers.add("3");
         try {
-             zkArr[1].reconfig(null, leavingServers, null, 8, null);
+             zkAdminArr[1].reconfig(null, leavingServers, null, 8, null);
             Assert.fail("Reconfig should have failed since the current config version is not 8");
         } catch (KeeperException.BadVersionException e) {
             // We expect this to happen.
@@ -144,7 +155,7 @@ public class ReconfigFailureCasesTest extends QuorumPeerTestBase {
             Assert.fail("Should have been BadVersionException!");
         }
 
-        ReconfigTest.closeAllHandles(zkArr);
+        ReconfigTest.closeAllHandles(zkArr, zkAdminArr);
     }
 
     /*
@@ -158,6 +169,7 @@ public class ReconfigFailureCasesTest extends QuorumPeerTestBase {
         qu.disableJMXTest = true;
         qu.startAll();
         ZooKeeper[] zkArr = ReconfigTest.createHandles(qu);
+        ZooKeeperAdmin[] zkAdminArr = ReconfigTest.createAdminHandles(qu);
 
         List<String> leavingServers = new ArrayList<String>();
         leavingServers.add("3");
@@ -170,7 +182,7 @@ public class ReconfigFailureCasesTest extends QuorumPeerTestBase {
             // We try to remove server 3, which requires a quorum of {1,2,3}
             // (we have that) and of {1,2}, but 2 is down so we won't get a
             // quorum of new config ACKs.
-            zkArr[1].reconfig(null, leavingServers, null, -1, null);
+            zkAdminArr[1].reconfig(null, leavingServers, null, -1, null);
             Assert.fail("Reconfig should have failed since we don't have quorum of new config");
         } catch (KeeperException.ConnectionLossException e) {
             // We expect leader to lose quorum of proposed config and time out
@@ -186,7 +198,7 @@ public class ReconfigFailureCasesTest extends QuorumPeerTestBase {
                 qu.getPeer(1).peer.getServerState());
         Assert.assertEquals(QuorumStats.Provider.LOOKING_STATE,
                 qu.getPeer(3).peer.getServerState());
-        ReconfigTest.closeAllHandles(zkArr);
+        ReconfigTest.closeAllHandles(zkArr, zkAdminArr);
     }
 
     /*
@@ -222,6 +234,7 @@ public class ReconfigFailureCasesTest extends QuorumPeerTestBase {
 
         MainThread mt[] = new MainThread[SERVER_COUNT];
         ZooKeeper zk[] = new ZooKeeper[SERVER_COUNT];
+        ZooKeeperAdmin zkAdmin[] = new ZooKeeperAdmin[SERVER_COUNT];
 
         // Server 0 stays down
         for (int i = 1; i < SERVER_COUNT; i++) {
@@ -230,6 +243,9 @@ public class ReconfigFailureCasesTest extends QuorumPeerTestBase {
             mt[i].start();
             zk[i] = new ZooKeeper("127.0.0.1:" + ports[i][2],
                     ClientBase.CONNECTION_TIMEOUT, this);
+            zkAdmin[i] = new ZooKeeperAdmin("127.0.0.1:" + ports[i][2],
+                    ClientBase.CONNECTION_TIMEOUT, this);
+            zkAdmin[i].addAuthInfo("digest", "super:test".getBytes());
         }
 
         for (int i = 1; i < SERVER_COUNT; i++) {
@@ -239,7 +255,7 @@ public class ReconfigFailureCasesTest extends QuorumPeerTestBase {
         }
 
         try {
-            zk[1].reconfig("", "", nextQuorumCfgSection, -1, new Stat());
+            zkAdmin[1].reconfig("", "", nextQuorumCfgSection, -1, new Stat());
             Assert.fail("Reconfig should have failed with NewConfigNoQuorum");
         } catch (NewConfigNoQuorum e) {
             // This is expected case since server 0 is down and 3 can't vote
@@ -250,19 +266,20 @@ public class ReconfigFailureCasesTest extends QuorumPeerTestBase {
         // In this scenario to change 3's role to participant we need to remove it first
         ArrayList<String> leavingServers = new ArrayList<String>();
         leavingServers.add("3");
-        ReconfigTest.reconfig(zk[1], null, leavingServers, null, -1);
+        ReconfigTest.reconfig(zkAdmin[1], null, leavingServers, null, -1);
         ReconfigTest.testNormalOperation(zk[2], zk[3]);
         ReconfigTest.testServerHasConfig(zk[3], null, leavingServers);
 
         // Now we're adding it back as a participant and everything should work.
         List<String> newMembers = Arrays.asList(nextQuorumCfgSection.split("\n"));
-        ReconfigTest.reconfig(zk[1], null, null, newMembers, -1);
+        ReconfigTest.reconfig(zkAdmin[1], null, null, newMembers, -1);
         ReconfigTest.testNormalOperation(zk[2], zk[3]);
         for (int i = 1; i < SERVER_COUNT; i++) {
             ReconfigTest.testServerHasConfig(zk[i], newMembers, null);
         }
         for (int i = 1; i < SERVER_COUNT; i++) {
             zk[i].close();
+            zkAdmin[i].close();
             mt[i].shutdown();
         }
     }

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

@@ -31,6 +31,7 @@ import org.apache.zookeeper.CreateMode;
 import org.apache.zookeeper.PortAssignment;
 import org.apache.zookeeper.ZooDefs.Ids;
 import org.apache.zookeeper.ZooKeeper;
+import org.apache.zookeeper.admin.ZooKeeperAdmin;
 import org.apache.zookeeper.test.ClientBase;
 import org.apache.zookeeper.test.ReconfigTest;
 import org.junit.Assert;
@@ -44,6 +45,9 @@ public class ReconfigLegacyTest extends QuorumPeerTestBase {
     @Before
     public void setup() {
         ClientBase.setupTestEnv();
+        QuorumPeerConfig.setReconfigEnabled(true);
+        System.setProperty("zookeeper.DigestAuthenticationProvider.superDigest",
+                "super:D/InIHSb7yEEbrWz8b9l71RjZJU="/* password is 'test'*/);
     }
 
     /**
@@ -176,6 +180,7 @@ public class ReconfigLegacyTest extends QuorumPeerTestBase {
 
         MainThread mt[] = new MainThread[SERVER_COUNT];
         ZooKeeper zk[] = new ZooKeeper[SERVER_COUNT];
+        ZooKeeperAdmin zkAdmin[] = new ZooKeeperAdmin[SERVER_COUNT];
 
         // Start the servers with a static config file, without a dynamic config file.
         for (int i = 0; i < SERVER_COUNT; i++) {
@@ -190,6 +195,9 @@ public class ReconfigLegacyTest extends QuorumPeerTestBase {
                     ClientBase.waitForServerUp("127.0.0.1:" + clientPorts[i],
                             CONNECTION_TIMEOUT));
             zk[i] = ClientBase.createZKClient("127.0.0.1:" + clientPorts[i]);
+            zkAdmin[i] = new ZooKeeperAdmin("127.0.0.1:" + clientPorts[i],
+                    ClientBase.CONNECTION_TIMEOUT, this);
+            zkAdmin[i].addAuthInfo("digest", "super:test".getBytes());
 
             ReconfigTest.testServerHasConfig(zk[i], allServers, null);
             Properties cfg = readPropertiesFromFile(mt[i].confFile);
@@ -199,7 +207,7 @@ public class ReconfigLegacyTest extends QuorumPeerTestBase {
         }
         ReconfigTest.testNormalOperation(zk[0], zk[1]);
 
-        ReconfigTest.reconfig(zk[1], null, null, newServers, -1);
+        ReconfigTest.reconfig(zkAdmin[1], null, null, newServers, -1);
         ReconfigTest.testNormalOperation(zk[0], zk[1]);
 
         // Sleep since writing the config files may take time.
@@ -222,6 +230,7 @@ public class ReconfigLegacyTest extends QuorumPeerTestBase {
         for (int i = 0; i < SERVER_COUNT; i++) {
             mt[i].shutdown();
             zk[i].close();
+            zkAdmin[i].close();
         }
     }
 

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

@@ -27,6 +27,7 @@ import org.apache.zookeeper.KeeperException;
 import org.apache.zookeeper.data.Stat;
 import org.apache.zookeeper.PortAssignment;
 import org.apache.zookeeper.ZooKeeper;
+import org.apache.zookeeper.admin.ZooKeeperAdmin;
 import org.apache.zookeeper.test.ClientBase;
 import org.apache.zookeeper.test.ReconfigTest;
 import org.junit.Assert;
@@ -37,6 +38,7 @@ public class StandaloneDisabledTest extends QuorumPeerTestBase {
     private final int NUM_SERVERS = 5;
     private MainThread peers[];
     private ZooKeeper zkHandles[];
+    private ZooKeeperAdmin zkAdminHandles[];
     private int clientPorts[];
     private final int leaderId = 0;
     private final int follower1 = 1;
@@ -75,7 +77,7 @@ public class StandaloneDisabledTest extends QuorumPeerTestBase {
         reconfigServers.clear();
         reconfigServers.add(Integer.toString(follower2));
         try {
-            ReconfigTest.reconfig(zkHandles[follower1], null, reconfigServers, null, -1);
+            ReconfigTest.reconfig(zkAdminHandles[follower1], null, reconfigServers, null, -1);
             Assert.fail("reconfig completed successfully even though there is no quorum up in new config!");
         } catch (KeeperException.NewConfigNoQuorum e) { }
 
@@ -92,7 +94,7 @@ public class StandaloneDisabledTest extends QuorumPeerTestBase {
         reconfigServers.clear();
         reconfigServers.add(Integer.toString(follower2));
         try {
-            zkHandles[follower2].reconfig(null, reconfigServers, null, -1, new Stat());
+            zkAdminHandles[follower2].reconfig(null, reconfigServers, null, -1, new Stat());
             Assert.fail("reconfig completed successfully even though there is no quorum up in new config!");
         } catch (KeeperException.BadArgumentsException e) {
             // This is expected.
@@ -118,11 +120,15 @@ public class StandaloneDisabledTest extends QuorumPeerTestBase {
     private void setUpData() throws Exception {
         ClientBase.setupTestEnv();
         QuorumPeerConfig.setStandaloneEnabled(false);
+        QuorumPeerConfig.setReconfigEnabled(true);
         peers = new MainThread[NUM_SERVERS];
         zkHandles = new ZooKeeper[NUM_SERVERS];
+        zkAdminHandles = new ZooKeeperAdmin[NUM_SERVERS];
         clientPorts = new int[NUM_SERVERS];
         serverStrings = buildServerStrings();
         reconfigServers = new ArrayList<String>();
+        System.setProperty("zookeeper.DigestAuthenticationProvider.superDigest",
+                "super:D/InIHSb7yEEbrWz8b9l71RjZJU="/* password is 'test'*/);
     }
 
     /**
@@ -131,6 +137,7 @@ public class StandaloneDisabledTest extends QuorumPeerTestBase {
     private void shutDownData() throws Exception {
         for (int i = 0; i < NUM_SERVERS; i++) {
             zkHandles[i].close();
+            zkAdminHandles[i].close();
         }
         for (int i = 1; i < NUM_SERVERS; i++) {
             peers[i].shutdown();
@@ -167,6 +174,8 @@ public class StandaloneDisabledTest extends QuorumPeerTestBase {
         Assert.assertTrue("Error- Server started in Standalone Mode!",
                 peers[id].isQuorumPeerRunning());
         zkHandles[id] = ClientBase.createZKClient("127.0.0.1:" + clientPorts[id]);
+        zkAdminHandles[id] = new ZooKeeperAdmin("127.0.0.1:" + clientPorts[id], CONNECTION_TIMEOUT, this);
+        zkAdminHandles[id].addAuthInfo("digest", "super:test".getBytes());
     }
 
     /**
@@ -221,14 +230,14 @@ public class StandaloneDisabledTest extends QuorumPeerTestBase {
     private void testReconfig(int id, boolean adding,
                               ArrayList<String> servers) throws Exception {
         if (adding) {
-            ReconfigTest.reconfig(zkHandles[id], servers, null, null, -1);
+            ReconfigTest.reconfig(zkAdminHandles[id], servers, null, null, -1);
             for (String server : servers) {
                 int id2 = Integer.parseInt(server.substring(7, 8)); //server.#
                 ReconfigTest.testNormalOperation(zkHandles[id], zkHandles[id2]);
             }
             ReconfigTest.testServerHasConfig(zkHandles[id], servers, null);
         } else {
-            ReconfigTest.reconfig(zkHandles[id], null, servers, null, -1);
+            ReconfigTest.reconfig(zkAdminHandles[id], null, servers, null, -1);
             ReconfigTest.testServerHasConfig(zkHandles[id], null, servers);
         }
 

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

@@ -1216,13 +1216,13 @@ public class Zab1_0Test extends ZKTestCase {
     }
     
     private Leader createLeader(File tmpDir, QuorumPeer peer)
-    throws IOException, NoSuchFieldException, IllegalAccessException{
+    throws IOException, NoSuchFieldException, IllegalAccessException {
         LeaderZooKeeperServer zk = prepareLeader(tmpDir, peer);
         return new Leader(peer, zk);
     }
     
     private Leader createMockLeader(File tmpDir, QuorumPeer peer)
-    throws IOException, NoSuchFieldException, IllegalAccessException{
+    throws IOException, NoSuchFieldException, IllegalAccessException {
         LeaderZooKeeperServer zk = prepareLeader(tmpDir, peer);
         return new MockLeader(peer, zk);
     }

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

@@ -115,7 +115,8 @@ public class ACLTest extends ZKTestCase implements Watcher {
                 zk.create(path, path.getBytes(), Ids.OPEN_ACL_UNSAFE,
                         CreateMode.PERSISTENT);
             }
-            Assert.assertTrue("size of the acl map ", (1 == zks.getZKDatabase().getAclSize()));
+            int size = zks.getZKDatabase().getAclSize();
+            Assert.assertTrue("size of the acl map ", (2 == zks.getZKDatabase().getAclSize()));
             for (int j = 100; j < 200; j++) {
                 path = "/" + j;
                 ACL acl = new ACL();
@@ -128,7 +129,7 @@ public class ACLTest extends ZKTestCase implements Watcher {
                 list.add(acl);
                 zk.create(path, path.getBytes(), list, CreateMode.PERSISTENT);
             }
-            Assert.assertTrue("size of the acl map ", (101 == zks.getZKDatabase().getAclSize()));
+            Assert.assertTrue("size of the acl map ", (102 == zks.getZKDatabase().getAclSize()));
         } finally {
             // now shutdown the server and restart it
             f.shutdown();
@@ -145,7 +146,7 @@ public class ACLTest extends ZKTestCase implements Watcher {
             Assert.assertTrue("waiting for server up",
                        ClientBase.waitForServerUp(HOSTPORT, CONNECTION_TIMEOUT));
             zk = ClientBase.createZKClient(HOSTPORT);
-            Assert.assertTrue("acl map ", (101 == zks.getZKDatabase().getAclSize()));
+            Assert.assertTrue("acl map ", (102 == zks.getZKDatabase().getAclSize()));
             for (int j = 200; j < 205; j++) {
                 path = "/" + j;
                 ACL acl = new ACL();
@@ -158,7 +159,7 @@ public class ACLTest extends ZKTestCase implements Watcher {
                 list.add(acl);
                 zk.create(path, path.getBytes(), list, CreateMode.PERSISTENT);
             }
-            Assert.assertTrue("acl map ", (106 == zks.getZKDatabase().getAclSize()));
+            Assert.assertTrue("acl map ", (107 == zks.getZKDatabase().getAclSize()));
     
             zk.close();
         } finally {

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

@@ -0,0 +1,220 @@
+/**
+ * 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.test;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.concurrent.TimeoutException;
+
+import org.apache.zookeeper.ZKTestCase;
+import org.apache.zookeeper.KeeperException;
+import org.apache.zookeeper.ZooDefs;
+import org.apache.zookeeper.PortAssignment;
+import org.apache.zookeeper.admin.ZooKeeperAdmin;
+import org.apache.zookeeper.data.ACL;
+import org.apache.zookeeper.data.Id;
+import org.apache.zookeeper.data.Stat;
+import org.apache.zookeeper.server.quorum.QuorumPeerConfig;
+import org.junit.After;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class ReconfigExceptionTest extends ZKTestCase {
+    private static final Logger LOG = LoggerFactory
+            .getLogger(ReconfigExceptionTest.class);
+    private static String authProvider = "zookeeper.DigestAuthenticationProvider.superDigest";
+    // Use DigestAuthenticationProvider.base64Encode or
+    // run ZooKeeper jar with org.apache.zookeeper.server.auth.DigestAuthenticationProvider to generate password.
+    // An example:
+    // java -cp zookeeper-3.6.0-SNAPSHOT.jar:lib/log4j-1.2.17.jar:lib/slf4j-log4j12-1.7.5.jar:
+    // lib/slf4j-api-1.7.5.jar org.apache.zookeeper.server.auth.DigestAuthenticationProvider super:test
+    // The password here is 'test'.
+    private static String superDigest = "super:D/InIHSb7yEEbrWz8b9l71RjZJU=";
+    private QuorumUtil qu;
+    private ZooKeeperAdmin zkAdmin;
+
+    @Before
+    public void setup() throws InterruptedException {
+        System.setProperty(authProvider, superDigest);
+        QuorumPeerConfig.setReconfigEnabled(false);
+
+        // Get a three server quorum.
+        qu = new QuorumUtil(1);
+        qu.disableJMXTest = true;
+
+        try {
+            qu.startAll();
+        } catch (IOException e) {
+            Assert.fail("Fail to start quorum servers.");
+        }
+
+        resetZKAdmin();
+    }
+
+    @After
+    public void tearDown() throws Exception {
+        System.clearProperty(authProvider);
+        try {
+            if (qu != null) {
+                qu.tearDown();
+            }
+            if (zkAdmin != null) {
+                zkAdmin.close();
+            }
+        } catch (Exception e) {
+            // Ignore.
+        }
+    }
+
+    @Test(timeout = 10000)
+    public void testReconfigDisabledByDefault() throws InterruptedException {
+        try {
+            reconfigPort();
+            Assert.fail("Reconfig should be disabled by default.");
+        } catch (KeeperException e) {
+            Assert.assertTrue(e.code() == KeeperException.Code.RECONFIGDISABLED);
+        }
+    }
+
+    @Test(timeout = 10000)
+    public void testReconfigFailWithoutAuth() throws InterruptedException {
+        // Now enable reconfig feature by turning on the switch.
+        QuorumPeerConfig.setReconfigEnabled(true);
+
+        try {
+            reconfigPort();
+            Assert.fail("Reconfig should fail without auth.");
+        } catch (KeeperException e) {
+            // However a failure is still expected as user is not authenticated, so ACL check will fail.
+            Assert.assertTrue(e.code() == KeeperException.Code.NOAUTH);
+        }
+    }
+
+    @Test(timeout = 10000)
+    public void testReconfigEnabledWithSuperUser() throws InterruptedException {
+        QuorumPeerConfig.setReconfigEnabled(true);
+
+        try {
+            zkAdmin.addAuthInfo("digest", "super:test".getBytes());
+            Assert.assertTrue(reconfigPort());
+        } catch (KeeperException e) {
+            Assert.fail("Reconfig should not fail, but failed with exception : " + e.getMessage());
+        }
+    }
+
+    @Test(timeout = 10000)
+    public void testReconfigFailWithAuthWithNoACL() throws InterruptedException {
+        resetZKAdmin();
+        QuorumPeerConfig.setReconfigEnabled(true);
+
+        try {
+            zkAdmin.addAuthInfo("digest", "user:test".getBytes());
+            reconfigPort();
+            Assert.fail("Reconfig should fail without a valid ACL associated with user.");
+        } catch (KeeperException e) {
+            // Again failure is expected because no ACL is associated with this user.
+            Assert.assertTrue(e.code() == KeeperException.Code.NOAUTH);
+        }
+    }
+
+    @Test(timeout = 10000)
+    public void testReconfigEnabledWithAuthAndWrongACL() throws InterruptedException {
+        resetZKAdmin();
+        QuorumPeerConfig.setReconfigEnabled(true);
+
+        try {
+            zkAdmin.addAuthInfo("digest", "super:test".getBytes());
+            // There is ACL however the permission is wrong - need WRITE permission at leaste.
+            ArrayList<ACL> acls = new ArrayList<ACL>(
+                    Collections.singletonList(
+                            new ACL(ZooDefs.Perms.READ,
+                                    new Id("digest", "user:tl+z3z0vO6PfPfEENfLF96E6pM0="/* password is test */))));
+            zkAdmin.setACL(ZooDefs.CONFIG_NODE, acls, -1);
+            resetZKAdmin();
+            zkAdmin.addAuthInfo("digest", "user:test".getBytes());
+            reconfigPort();
+            Assert.fail("Reconfig should fail with an ACL that is read only!");
+        } catch (KeeperException e) {
+            Assert.assertTrue(e.code() == KeeperException.Code.NOAUTH);
+        }
+    }
+
+    @Test(timeout = 10000)
+    public void testReconfigEnabledWithAuthAndACL() throws InterruptedException {
+        resetZKAdmin();
+        QuorumPeerConfig.setReconfigEnabled(true);
+
+        try {
+            zkAdmin.addAuthInfo("digest", "super:test".getBytes());
+            ArrayList<ACL> acls = new ArrayList<ACL>(
+                    Collections.singletonList(
+                            new ACL(ZooDefs.Perms.WRITE,
+                            new Id("digest", "user:tl+z3z0vO6PfPfEENfLF96E6pM0="/* password is test */))));
+            zkAdmin.setACL(ZooDefs.CONFIG_NODE, acls, -1);
+            resetZKAdmin();
+            zkAdmin.addAuthInfo("digest", "user:test".getBytes());
+            Assert.assertTrue(reconfigPort());
+        } catch (KeeperException e) {
+            Assert.fail("Reconfig should not fail, but failed with exception : " + e.getMessage());
+        }
+    }
+
+    // Utility method that recreates a new ZooKeeperAdmin handle, and wait for the handle to connect to
+    // quorum servers.
+    private void resetZKAdmin() throws InterruptedException {
+        String cnxString;
+        ClientBase.CountdownWatcher watcher = new ClientBase.CountdownWatcher();
+        try {
+            cnxString = "127.0.0.1:" + qu.getPeer(1).peer.getClientPort();
+            if (zkAdmin != null) {
+                zkAdmin.close();
+            }
+            zkAdmin = new ZooKeeperAdmin(cnxString,
+                    ClientBase.CONNECTION_TIMEOUT, watcher);
+        } catch (IOException e) {
+            Assert.fail("Fail to create ZooKeeperAdmin handle.");
+            return;
+        }
+
+        try {
+            watcher.waitForConnected(ClientBase.CONNECTION_TIMEOUT);
+        } catch (InterruptedException | TimeoutException e) {
+            Assert.fail("ZooKeeper admin client can not connect to " + cnxString);
+        }
+    }
+
+    private boolean reconfigPort() throws KeeperException, InterruptedException {
+        List<String> joiningServers = new ArrayList<String>();
+        int leaderId = 1;
+        while (qu.getPeer(leaderId).peer.leader == null)
+            leaderId++;
+        int followerId = leaderId == 1 ? 2 : 1;
+        joiningServers.add("server." + followerId + "=localhost:"
+                + qu.getPeer(followerId).peer.getQuorumAddress().getPort() /*quorum port*/
+                + ":" + qu.getPeer(followerId).peer.getElectionAddress().getPort() /*election port*/
+                + ":participant;localhost:" + PortAssignment.unique()/* new client port */);
+        zkAdmin.reconfig(joiningServers, null, null, -1, new Stat());
+        return true;
+    }
+}

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

@@ -0,0 +1,130 @@
+/**
+ * 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.test;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.TimeoutException;
+
+import org.apache.zookeeper.ZKTestCase;
+import org.apache.zookeeper.KeeperException;
+import org.apache.zookeeper.PortAssignment;
+import org.apache.zookeeper.admin.ZooKeeperAdmin;
+import org.apache.zookeeper.data.Stat;
+import org.apache.zookeeper.server.quorum.QuorumPeerConfig;
+import org.junit.After;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class ReconfigMisconfigTest extends ZKTestCase {
+    private static final Logger LOG = LoggerFactory.getLogger(ReconfigMisconfigTest.class);
+    private QuorumUtil qu;
+    private ZooKeeperAdmin zkAdmin;
+    private static String errorMsg = "Reconfig should fail without configuring the super " +
+            "user's password on server side first.";
+
+    @Before
+    public void setup() throws InterruptedException {
+        QuorumPeerConfig.setReconfigEnabled(false);
+        // Get a three server quorum.
+        qu = new QuorumUtil(1);
+        qu.disableJMXTest = true;
+        try {
+            qu.startAll();
+        } catch (IOException e) {
+            Assert.fail("Fail to start quorum servers.");
+        }
+
+        instantiateZKAdmin();
+    }
+
+    @After
+    public void tearDown() throws Exception {
+        try {
+            if (qu != null) {
+                qu.tearDown();
+            }
+            if (zkAdmin != null) {
+                zkAdmin.close();
+            }
+        } catch (Exception e) {
+            // Ignore.
+        }
+    }
+
+    @Test(timeout = 10000)
+    public void testReconfigFailWithoutSuperuserPasswordConfiguredOnServer() throws InterruptedException {
+        // This tests the case where ZK ensemble does not have the super user's password configured.
+        // Reconfig should fail as the super user has to be explicitly configured via
+        // zookeeper.DigestAuthenticationProvider.superDigest.
+        QuorumPeerConfig.setReconfigEnabled(true);
+        try {
+            reconfigPort();
+            Assert.fail(errorMsg);
+        } catch (KeeperException e) {
+            Assert.assertTrue(e.getCode() == KeeperException.Code.NoAuth);
+        }
+
+        try {
+            zkAdmin.addAuthInfo("digest", "super:".getBytes());
+            reconfigPort();
+            Assert.fail(errorMsg);
+        } catch (KeeperException e) {
+            Assert.assertTrue(e.getCode() == KeeperException.Code.NoAuth);
+        }
+    }
+
+    private void instantiateZKAdmin() throws InterruptedException {
+        String cnxString;
+        ClientBase.CountdownWatcher watcher = new ClientBase.CountdownWatcher();
+        try {
+            cnxString = "127.0.0.1:" + qu.getPeer(1).peer.getClientPort();
+            zkAdmin = new ZooKeeperAdmin(cnxString,
+                    ClientBase.CONNECTION_TIMEOUT, watcher);
+        } catch (IOException e) {
+            Assert.fail("Fail to create ZooKeeperAdmin handle.");
+            return;
+        }
+
+        try {
+            watcher.waitForConnected(ClientBase.CONNECTION_TIMEOUT);
+        } catch (InterruptedException | TimeoutException e) {
+            Assert.fail("ZooKeeper admin client can not connect to " + cnxString);
+        }
+    }
+
+    private boolean reconfigPort() throws KeeperException, InterruptedException {
+        List<String> joiningServers = new ArrayList<String>();
+        int leaderId = 1;
+        while (qu.getPeer(leaderId).peer.leader == null)
+            leaderId++;
+        int followerId = leaderId == 1 ? 2 : 1;
+        joiningServers.add("server." + followerId + "=localhost:"
+                + qu.getPeer(followerId).peer.getQuorumAddress().getPort() /*quorum port*/
+                + ":" + qu.getPeer(followerId).peer.getElectionAddress().getPort() /*election port*/
+                + ":participant;localhost:" + PortAssignment.unique()/* new client port */);
+        zkAdmin.reconfig(joiningServers, null, null, -1, new Stat());
+        return true;
+    }
+}
+

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

@@ -29,25 +29,28 @@ 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;
-import org.apache.zookeeper.WatchedEvent;
-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.QuorumPeer.QuorumServer;
 import org.apache.zookeeper.server.quorum.QuorumPeer.ServerState;
+import org.apache.zookeeper.server.quorum.QuorumPeerConfig;
 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.ZooKeeper;
+import org.apache.zookeeper.admin.ZooKeeperAdmin;
+import org.apache.zookeeper.Watcher;
+import org.apache.zookeeper.WatchedEvent;
+import org.apache.zookeeper.ZooDefs;
+import org.apache.zookeeper.CreateMode;
+import org.apache.zookeeper.PortAssignment;
 import org.junit.After;
 import org.junit.Assert;
+import org.junit.Before;
 import org.junit.Test;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -58,6 +61,13 @@ public class ReconfigTest extends ZKTestCase implements DataCallback{
 
     private QuorumUtil qu;
 
+    @Before
+    public void setup() {
+        System.setProperty("zookeeper.DigestAuthenticationProvider.superDigest",
+                "super:D/InIHSb7yEEbrWz8b9l71RjZJU="/* password is 'test'*/);
+        QuorumPeerConfig.setReconfigEnabled(true);
+    }
+
     @After
     public void tearDown() throws Exception {
         if (qu != null) {
@@ -65,13 +75,13 @@ public class ReconfigTest extends ZKTestCase implements DataCallback{
         }
     }
 
-    public static String reconfig(ZooKeeper zk, List<String> joiningServers,
-            List<String> leavingServers, List<String> newMembers, long fromConfig)
+    public static String reconfig(ZooKeeperAdmin zkAdmin, List<String> joiningServers,
+                                  List<String> leavingServers, List<String> newMembers, long fromConfig)
             throws KeeperException, InterruptedException {
         byte[] config = null;
         for (int j = 0; j < 30; j++) {
             try {
-                config = zk.reconfig(joiningServers, leavingServers,
+                config = zkAdmin.reconfig(joiningServers, leavingServers,
                         newMembers, fromConfig, new Stat());
                 break;
             } catch (KeeperException.ConnectionLossException e) {
@@ -208,19 +218,40 @@ public class ReconfigTest extends ZKTestCase implements DataCallback{
         return zkArr;
     }
 
-    public static void closeAllHandles(ZooKeeper[] zkArr) throws InterruptedException {
+    public static ZooKeeperAdmin[] createAdminHandles(QuorumUtil qu) throws IOException {
+        // create an extra handle, so we can index the handles from 1 to qu.ALL
+        // using the server id.
+        ZooKeeperAdmin[] zkAdminArr = new ZooKeeperAdmin[qu.ALL + 1];
+        zkAdminArr[0] = null; // not used.
+        for (int i = 1; i <= qu.ALL; i++) {
+            // server ids are 1, 2 and 3
+            zkAdminArr[i] = new ZooKeeperAdmin("127.0.0.1:"
+                    + qu.getPeer(i).peer.getClientPort(),
+                    ClientBase.CONNECTION_TIMEOUT, new Watcher() {
+                public void process(WatchedEvent event) {
+                }});
+            zkAdminArr[i].addAuthInfo("digest", "super:test".getBytes());
+        }
+
+        return zkAdminArr;
+    }
+
+    public static void closeAllHandles(ZooKeeper[] zkArr, ZooKeeperAdmin[] zkAdminArr) throws InterruptedException {
         for (ZooKeeper zk : zkArr)
             if (zk != null)
                 zk.close();
+        for (ZooKeeperAdmin zkAdmin : zkAdminArr)
+            if (zkAdmin != null)
+                zkAdmin.close();
     }
 
- 
     @Test
     public void testRemoveAddOne() throws Exception {
         qu = new QuorumUtil(1); // create 3 servers
         qu.disableJMXTest = true;
         qu.startAll();
         ZooKeeper[] zkArr = createHandles(qu);
+        ZooKeeperAdmin[] zkAdminArr = createAdminHandles(qu);
 
         List<String> leavingServers = new ArrayList<String>();
         List<String> joiningServers = new ArrayList<String>();
@@ -242,6 +273,10 @@ public class ReconfigTest extends ZKTestCase implements DataCallback{
                     : zkArr[(leaderIndex % qu.ALL) + 1];
             ZooKeeper zk2 = (leavingIndex == leaderIndex) ? zkArr[(leaderIndex % qu.ALL) + 1]
                     : zkArr[leaderIndex];
+            ZooKeeperAdmin zkAdmin1 = (leavingIndex == leaderIndex) ? zkAdminArr[leaderIndex]
+                    : zkAdminArr[(leaderIndex % qu.ALL) + 1];
+            ZooKeeperAdmin zkAdmin2 = (leavingIndex == leaderIndex) ? zkAdminArr[(leaderIndex % qu.ALL) + 1]
+                    : zkAdminArr[leaderIndex];
 
             leavingServers.add(Integer.toString(leavingIndex));
 
@@ -256,7 +291,7 @@ public class ReconfigTest extends ZKTestCase implements DataCallback{
                             .getPort() + ":participant;localhost:"
                     + qu.getPeer(leavingIndex).peer.getClientPort());
 
-            String configStr = reconfig(zk1, null, leavingServers, null, -1);
+            String configStr = reconfig(zkAdmin1, null, leavingServers, null, -1);
             testServerHasConfig(zk2, null, leavingServers);
             testNormalOperation(zk2, zk1);
 
@@ -265,13 +300,13 @@ public class ReconfigTest extends ZKTestCase implements DataCallback{
 
             // checks that conditioning on version works properly
             try {
-                reconfig(zk2, joiningServers, null, null, version + 1);
+                reconfig(zkAdmin2, joiningServers, null, null, version + 1);
                 Assert.fail("reconfig succeeded even though version condition was incorrect!");
             } catch (KeeperException.BadVersionException e) {
 
             }
 
-            reconfig(zk2, joiningServers, null, null, version);
+            reconfig(zkAdmin2, joiningServers, null, null, version);
 
             testNormalOperation(zk1, zk2);
             testServerHasConfig(zk1, joiningServers, null);
@@ -283,7 +318,7 @@ public class ReconfigTest extends ZKTestCase implements DataCallback{
             joiningServers.clear();
         }
 
-        closeAllHandles(zkArr);
+        closeAllHandles(zkArr, zkAdminArr);
     }
 
     /**
@@ -298,6 +333,7 @@ public class ReconfigTest extends ZKTestCase implements DataCallback{
         qu.disableJMXTest = true;
         qu.startAll();
         ZooKeeper[] zkArr = createHandles(qu);
+        ZooKeeperAdmin[] zkAdminArr = createAdminHandles(qu);
 
         List<String> leavingServers = new ArrayList<String>();
         List<String> joiningServers = new ArrayList<String>();
@@ -345,7 +381,7 @@ public class ReconfigTest extends ZKTestCase implements DataCallback{
         qu.shutdown(leavingIndex2);
 
         // 3 servers still up so this should work
-        reconfig(zkArr[stayingIndex2], null, leavingServers, null, -1);
+        reconfig(zkAdminArr[stayingIndex2], null, leavingServers, null, -1);
         
         qu.shutdown(stayingIndex2);
 
@@ -366,7 +402,7 @@ public class ReconfigTest extends ZKTestCase implements DataCallback{
         Thread.sleep(10000);
 
         try {
-            reconfig(zkArr[stayingIndex1], joiningServers, null, null, -1);
+            reconfig(zkAdminArr[stayingIndex1], joiningServers, null, null, -1);
             Assert.fail("reconfig completed successfully even though there is no quorum up in new config!");
         } catch (KeeperException.NewConfigNoQuorum e) {
 
@@ -375,7 +411,7 @@ public class ReconfigTest extends ZKTestCase implements DataCallback{
         // now start the third server so that new config has quorum
         qu.restart(stayingIndex2);
 
-        reconfig(zkArr[stayingIndex1], joiningServers, null, null, -1);
+        reconfig(zkAdminArr[stayingIndex1], joiningServers, null, null, -1);
         testNormalOperation(zkArr[stayingIndex2], zkArr[stayingIndex3]);
         testServerHasConfig(zkArr[stayingIndex2], joiningServers, null);
 
@@ -388,7 +424,7 @@ public class ReconfigTest extends ZKTestCase implements DataCallback{
         testNormalOperation(zkArr[stayingIndex2], zkArr[leavingIndex2]);
         testServerHasConfig(zkArr[leavingIndex2], joiningServers, null);
 
-        closeAllHandles(zkArr);
+        closeAllHandles(zkArr, zkAdminArr);
     }
 
     @Test
@@ -397,6 +433,7 @@ public class ReconfigTest extends ZKTestCase implements DataCallback{
         qu.disableJMXTest = true;
         qu.startAll();
         ZooKeeper[] zkArr = createHandles(qu);
+        ZooKeeperAdmin[] zkAdminArr = createAdminHandles(qu);
 
         // new config will have three of the servers as followers
         // two of the servers as observers, and all ports different
@@ -413,7 +450,7 @@ public class ReconfigTest extends ZKTestCase implements DataCallback{
         qu.shutdown(6);
         qu.shutdown(7);
         
-        reconfig(zkArr[1], null, null, newServers, -1);
+        reconfig(zkAdminArr[1], null, null, newServers, -1);
         testNormalOperation(zkArr[1], zkArr[2]);
        
         testServerHasConfig(zkArr[1], newServers, null);
@@ -426,7 +463,7 @@ public class ReconfigTest extends ZKTestCase implements DataCallback{
         
         testNormalOperation(zkArr[1], zkArr[2]);
 
-        closeAllHandles(zkArr);
+        closeAllHandles(zkArr, zkAdminArr);
     }
 
     @Test
@@ -435,6 +472,7 @@ public class ReconfigTest extends ZKTestCase implements DataCallback{
         qu.disableJMXTest = true;
         qu.startAll();
         ZooKeeper[] zkArr = createHandles(qu);
+        ZooKeeperAdmin[] zkAdminArr = createAdminHandles(qu);
 
         List<String> leavingServers = new ArrayList<String>();
        
@@ -443,7 +481,7 @@ public class ReconfigTest extends ZKTestCase implements DataCallback{
  
         LinkedList<Integer> results = new LinkedList<Integer>();
         
-        zkArr[1].reconfig(null, leavingServers, null, -1, this, results);   
+        zkAdminArr[1].reconfig(null, leavingServers, null, -1, this, results);
         
         synchronized (results) {
             while (results.size() < 1) {
@@ -456,7 +494,7 @@ public class ReconfigTest extends ZKTestCase implements DataCallback{
         for (int i=1; i<=5; i++)
             testServerHasConfig(zkArr[i], null, leavingServers);
 
-        closeAllHandles(zkArr);
+        closeAllHandles(zkArr, zkAdminArr);
     }
 
     @SuppressWarnings("unchecked")
@@ -475,6 +513,7 @@ public class ReconfigTest extends ZKTestCase implements DataCallback{
         qu.disableJMXTest = true;
         qu.startAll();
         ZooKeeper[] zkArr = createHandles(qu);
+        ZooKeeperAdmin[] zkAdminArr = createAdminHandles(qu);
 
         // changing a server's role / port is done by "adding" it with the same
         // id but different role / port
@@ -501,6 +540,8 @@ public class ReconfigTest extends ZKTestCase implements DataCallback{
             // to removed server
             ZooKeeper zk1 = (changingIndex == leaderIndex) ? zkArr[leaderIndex]
                     : zkArr[(leaderIndex % qu.ALL) + 1];
+            ZooKeeperAdmin zkAdmin1 = (changingIndex == leaderIndex) ? zkAdminArr[leaderIndex]
+                    : zkAdminArr[(leaderIndex % qu.ALL) + 1];
 
             // exactly as it is now, except for role change
             joiningServers.add("server."
@@ -513,7 +554,7 @@ public class ReconfigTest extends ZKTestCase implements DataCallback{
                             .getPort() + ":" + newRole + ";localhost:"
                     + qu.getPeer(changingIndex).peer.getClientPort());
 
-            reconfig(zk1, joiningServers, null, null, -1);
+            reconfig(zkAdmin1, joiningServers, null, null, -1);
             testNormalOperation(zkArr[changingIndex], zk1);
 
             if (newRole.equals("observer")) {
@@ -540,7 +581,7 @@ public class ReconfigTest extends ZKTestCase implements DataCallback{
                 changingIndex = leaderIndex;
             }
         }
-        closeAllHandles(zkArr);
+        closeAllHandles(zkArr, zkAdminArr);
     }
 
     @Test
@@ -549,6 +590,7 @@ public class ReconfigTest extends ZKTestCase implements DataCallback{
         qu.disableJMXTest = true;
         qu.startAll();
         ZooKeeper[] zkArr = createHandles(qu);
+        ZooKeeperAdmin[] zkAdminArr = createAdminHandles(qu);
 
         List<String> joiningServers = new ArrayList<String>();
 
@@ -568,7 +610,7 @@ public class ReconfigTest extends ZKTestCase implements DataCallback{
         // any reconfig is invoked
         testNormalOperation(zkArr[followerIndex], zkArr[leaderIndex]);
 
-        reconfig(zkArr[followerIndex], joiningServers, null, null, -1);
+        reconfig(zkAdminArr[followerIndex], joiningServers, null, null, -1);
 
         try {
           for (int i=0; i < 20; i++) {
@@ -584,6 +626,14 @@ public class ReconfigTest extends ZKTestCase implements DataCallback{
                 + oldClientPort,
                 ClientBase.CONNECTION_TIMEOUT, new Watcher() {
                     public void process(WatchedEvent event) {}});
+
+        zkAdminArr[followerIndex].close();
+        zkAdminArr[followerIndex] = new ZooKeeperAdmin("127.0.0.1:"
+                + oldClientPort,
+                ClientBase.CONNECTION_TIMEOUT, new Watcher() {
+            public void process(WatchedEvent event) {}});
+        zkAdminArr[followerIndex].addAuthInfo("digest", "super:test".getBytes());
+
         for (int i = 0; i < 10; i++) {
             try {
                 Thread.sleep(1000);
@@ -599,6 +649,13 @@ public class ReconfigTest extends ZKTestCase implements DataCallback{
                 ClientBase.CONNECTION_TIMEOUT, new Watcher() {
                     public void process(WatchedEvent event) {}});
 
+        zkAdminArr[followerIndex].close();
+        zkAdminArr[followerIndex] = new ZooKeeperAdmin("127.0.0.1:"
+                + newClientPort,
+                ClientBase.CONNECTION_TIMEOUT, new Watcher() {
+            public void process(WatchedEvent event) {}});
+        zkAdminArr[followerIndex].addAuthInfo("digest", "super:test".getBytes());
+
         testNormalOperation(zkArr[followerIndex], zkArr[leaderIndex]);
         testServerHasConfig(zkArr[followerIndex], joiningServers, null);
         Assert.assertEquals(newClientPort, qu.getPeer(followerIndex).peer.getClientPort());
@@ -615,7 +672,7 @@ public class ReconfigTest extends ZKTestCase implements DataCallback{
                 + ":participant;localhost:"
                 + qu.getPeer(leaderIndex).peer.getClientPort());
 
-        reconfig(zkArr[leaderIndex], joiningServers, null, null, -1);
+        reconfig(zkAdminArr[leaderIndex], joiningServers, null, null, -1);
 
         testNormalOperation(zkArr[followerIndex], zkArr[leaderIndex]);
 
@@ -634,7 +691,7 @@ public class ReconfigTest extends ZKTestCase implements DataCallback{
                     + qu.getPeer(i).peer.getClientPort());
         }
 
-        reconfig(zkArr[1], joiningServers, null, null, -1);
+        reconfig(zkAdminArr[1], joiningServers, null, null, -1);
 
         leaderIndex = getLeaderId(qu);
         int follower1 = leaderIndex == 1 ? 2 : 1;
@@ -650,7 +707,7 @@ public class ReconfigTest extends ZKTestCase implements DataCallback{
         testServerHasConfig(zkArr[follower1], joiningServers, null);
         testServerHasConfig(zkArr[follower2], joiningServers, null);
 
-        closeAllHandles(zkArr);
+        closeAllHandles(zkArr, zkAdminArr);
     }
 
     @Test
@@ -667,6 +724,7 @@ public class ReconfigTest extends ZKTestCase implements DataCallback{
         qu.disableJMXTest = true;
         qu.startAll();
         ZooKeeper[] zkArr = createHandles(qu);
+        ZooKeeperAdmin[] zkAdminArr = createAdminHandles(qu);
 
         List<String> joiningServers = new ArrayList<String>();
 
@@ -692,7 +750,7 @@ public class ReconfigTest extends ZKTestCase implements DataCallback{
             testNormalOperation(zkArr[followerIndex], zkArr[leaderIndex]);
 
             // Reconfigure
-            reconfig(zkArr[reconfigIndex], joiningServers, null, null, -1);
+            reconfig(zkAdminArr[reconfigIndex], joiningServers, null, null, -1);
             Thread.sleep(1000);
 
             // The follower reconfiguration will have failed
@@ -702,6 +760,12 @@ public class ReconfigTest extends ZKTestCase implements DataCallback{
                     ClientBase.CONNECTION_TIMEOUT, new Watcher() {
                         public void process(WatchedEvent event) {}});
 
+            zkAdminArr[serverIndex].close();
+            zkAdminArr[serverIndex] = new ZooKeeperAdmin("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);
@@ -721,7 +785,7 @@ public class ReconfigTest extends ZKTestCase implements DataCallback{
             joiningServers.add("server." + serverIndex + "=localhost:" + quorumPort
                     + ":" + electionPort + ":participant;localhost:" + oldClientPort);
 
-            reconfig(zkArr[reconfigIndex], joiningServers, null, null, -1);
+            reconfig(zkAdminArr[reconfigIndex], joiningServers, null, null, -1);
 
             zkArr[serverIndex].close();
             zkArr[serverIndex] = new ZooKeeper("127.0.0.1:"
@@ -733,7 +797,7 @@ public class ReconfigTest extends ZKTestCase implements DataCallback{
             testServerHasConfig(zkArr[serverIndex], joiningServers, null);
             Assert.assertEquals(oldClientPort, qu.getPeer(serverIndex).peer.getClientPort());
         }
-        closeAllHandles(zkArr);
+        closeAllHandles(zkArr, zkAdminArr);
     }
 
     @Test
@@ -754,6 +818,7 @@ public class ReconfigTest extends ZKTestCase implements DataCallback{
         qu.disableJMXTest = true;
         qu.startAll();
         ZooKeeper[] zkArr = createHandles(qu);
+        ZooKeeperAdmin[] zkAdminArr = createAdminHandles(qu);
 
         ArrayList<String> members = new ArrayList<String>();
         members.add("group.1=3:4:5");
@@ -771,7 +836,7 @@ public class ReconfigTest extends ZKTestCase implements DataCallback{
                     + "127.0.0.1:" + qu.getPeer(i).peer.getClientPort());
         }
 
-        reconfig(zkArr[1], null, null, members, -1);
+        reconfig(zkAdminArr[1], null, null, members, -1);
 
         // this should flush the config to servers 2, 3, 4 and 5
         testNormalOperation(zkArr[2], zkArr[3]);
@@ -803,7 +868,7 @@ public class ReconfigTest extends ZKTestCase implements DataCallback{
                     + "127.0.0.1:" + qu.getPeer(i).peer.getClientPort());
         }
 
-        reconfig(zkArr[1], null, null, members, -1);
+        reconfig(zkAdminArr[1], null, null, members, -1);
 
         // flush the config to server 2
         testNormalOperation(zkArr[1], zkArr[2]);
@@ -821,7 +886,7 @@ public class ReconfigTest extends ZKTestCase implements DataCallback{
                         + " doesn't think the quorum system is a majority quorum system!");
         }
 
-        closeAllHandles(zkArr);
+        closeAllHandles(zkArr, zkAdminArr);
     }
     
     @Test
@@ -849,6 +914,7 @@ public class ReconfigTest extends ZKTestCase implements DataCallback{
         qu.disableJMXTest = true;
         qu.startAll();
         ZooKeeper[] zkArr = createHandles(qu);
+        ZooKeeperAdmin[] zkAdminArr = createAdminHandles(qu);
 
         List<String> leavingServers = new ArrayList<String>();
         List<String> joiningServers = new ArrayList<String>();
@@ -873,6 +939,7 @@ public class ReconfigTest extends ZKTestCase implements DataCallback{
         assertRemotePeerMXBeanAttributes(leavingQS3, remotePeerBean3);
 
         ZooKeeper zk = zkArr[leavingIndex];
+        ZooKeeperAdmin zkAdmin = zkAdminArr[leavingIndex];
 
         leavingServers.add(Integer.toString(leavingIndex));
 
@@ -885,7 +952,7 @@ public class ReconfigTest extends ZKTestCase implements DataCallback{
                 + qu.getPeer(leavingIndex).peer.getClientPort());
 
         // Remove ReplicatedServer_1 from the ensemble
-        reconfig(zk, null, leavingServers, null, -1);
+        reconfig(zkAdmin, null, leavingServers, null, -1);
 
         // localPeerBean.1 of ReplicatedServer_1
         QuorumPeer removedPeer = qu.getPeer(leavingIndex).peer;
@@ -900,7 +967,7 @@ public class ReconfigTest extends ZKTestCase implements DataCallback{
         JMXEnv.ensureNone(remotePeerBean3);
 
         // Add ReplicatedServer_1 back to the ensemble
-        reconfig(zk, joiningServers, null, null, -1);
+        reconfig(zkAdmin, joiningServers, null, null, -1);
 
         // localPeerBean.1 of ReplicatedServer_1
         assertLocalPeerMXBeanAttributes(removedPeer, localPeerBean, true);
@@ -913,7 +980,7 @@ public class ReconfigTest extends ZKTestCase implements DataCallback{
         leavingQS3 = peer3.getView().get(new Long(leavingIndex));
         assertRemotePeerMXBeanAttributes(leavingQS3, remotePeerBean3);
 
-        closeAllHandles(zkArr);
+        closeAllHandles(zkArr, zkAdminArr);
     }
 
     /**
@@ -926,6 +993,7 @@ public class ReconfigTest extends ZKTestCase implements DataCallback{
         qu.disableJMXTest = true;
         qu.startAll();
         ZooKeeper[] zkArr = createHandles(qu);
+        ZooKeeperAdmin[] zkAdminArr = createAdminHandles(qu);
 
         // changing a server's role / port is done by "adding" it with the same
         // id but different role / port
@@ -953,6 +1021,7 @@ public class ReconfigTest extends ZKTestCase implements DataCallback{
         String newRole = "observer";
 
         ZooKeeper zk = zkArr[changingIndex];
+        ZooKeeperAdmin zkAdmin = zkAdminArr[changingIndex];
 
         // exactly as it is now, except for role change
         joiningServers.add("server." + changingIndex + "=127.0.0.1:"
@@ -962,7 +1031,7 @@ public class ReconfigTest extends ZKTestCase implements DataCallback{
                 + ":" + newRole + ";127.0.0.1:"
                 + qu.getPeer(changingIndex).peer.getClientPort());
 
-        reconfig(zk, joiningServers, null, null, -1);
+        reconfig(zkAdmin, joiningServers, null, null, -1);
         testNormalOperation(zkArr[changingIndex], zk);
 
         Assert.assertTrue(qu.getPeer(changingIndex).peer.observer != null
@@ -986,7 +1055,7 @@ public class ReconfigTest extends ZKTestCase implements DataCallback{
         changingQS3 = peer3.getView().get(new Long(changingIndex));
         assertRemotePeerMXBeanAttributes(changingQS3, remotePeerBean3);
 
-        closeAllHandles(zkArr);
+        closeAllHandles(zkArr, zkAdminArr);
     }
 
     private void assertLocalPeerMXBeanAttributes(QuorumPeer qp,

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

@@ -24,17 +24,20 @@ import java.io.File;
 import java.util.ArrayList;
 import java.util.List;
 
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-import org.apache.zookeeper.KeeperException;
-import org.apache.zookeeper.PortAssignment;
 import org.apache.zookeeper.Watcher;
-import org.apache.zookeeper.ZooKeeper;
+import org.apache.zookeeper.PortAssignment;
+import org.apache.zookeeper.KeeperException;
+import org.apache.zookeeper.server.quorum.QuorumPeerConfig;
 import org.apache.zookeeper.data.Stat;
 import org.apache.zookeeper.server.ServerCnxnFactory;
 import org.apache.zookeeper.server.ZooKeeperServer;
 import org.apache.zookeeper.server.quorum.QuorumPeerTestBase;
 import org.apache.zookeeper.test.ClientBase.CountdownWatcher;
+import org.apache.zookeeper.ZooKeeper;
+import org.apache.zookeeper.admin.ZooKeeperAdmin;
+import org.junit.Before;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 import org.junit.Assert;
 import org.junit.Test;
 
@@ -45,6 +48,13 @@ public class StandaloneTest extends QuorumPeerTestBase implements Watcher{
     protected static final Logger LOG =
         LoggerFactory.getLogger(StandaloneTest.class);
 
+    @Before
+    public void setup() {
+        System.setProperty("zookeeper.DigestAuthenticationProvider.superDigest",
+                "super:D/InIHSb7yEEbrWz8b9l71RjZJU="/* password is 'test'*/);
+        QuorumPeerConfig.setReconfigEnabled(true);
+    }
+
     /**
      * This test wouldn't create any dynamic config.
      * However, it adds a "clientPort=XXX" in static config file.
@@ -133,13 +143,15 @@ public class StandaloneTest extends QuorumPeerTestBase implements Watcher{
 
         CountdownWatcher watcher = new CountdownWatcher();
         ZooKeeper zk = new ZooKeeper(HOSTPORT, CONNECTION_TIMEOUT, watcher);
+        ZooKeeperAdmin zkAdmin = new ZooKeeperAdmin(HOSTPORT, CONNECTION_TIMEOUT, watcher);
         watcher.waitForConnected(CONNECTION_TIMEOUT);
 
         List<String> joiners = new ArrayList<String>();
         joiners.add("server.2=localhost:1234:1235;1236");
         // generate some transactions that will get logged
         try {
-            zk.reconfig(joiners, null, null, -1, new Stat());
+            zkAdmin.addAuthInfo("digest", "super:test".getBytes());
+            zkAdmin.reconfig(joiners, null, null, -1, new Stat());
             Assert.fail("Reconfiguration in standalone should trigger " +
                         "UnimplementedException");
         } catch (KeeperException.UnimplementedException ex) {