ZOOKEEPER-3720: Fix rolling upgrade failure (invalid protocol version)

The multi-address feature introduced in ZOOKEEPER-3188 required
changes in the Quorum protocol as we had to send all addresses in
the connection initiation message to enable the receiving side to
choose a reachable address in case of network failure.

The new code can handle both the old and the new protocol versions to
avoid 'invalid protocol' error e.g. during rolling restarts. However,
the new protocol version still can not be used during rolling upgrade
if the old servers are not supporting this protocol. In this case the
old and the new servers would form two distinct partitions until all
the servers get upgraded. To support rolling upgrades too, we want to
disable the MultiAddress feature by default and use the old protocol.

If the user would like enable the MultiAddress feature on a 3.6.0
cluster, she/he can do it either by 1) starting the cluster from
scratch (without rolling upgrade), or 2) performing a rolling upgrade
without the MultiAddress feature enabled then doing a rolling restart
with a new configuration where the MultiAddress feature is enabled.
During the rolling restart there will be no partitions, as all the
servers in the cluster now will run ZooKeeper version 3.6.0 which
understands now both protocols.

The changes in this patch:
- introducing new config property: multiAddress.enabled, disabled
by default
- updating QuorumCnxManager to be able to use both protocol versions
and to use the old one if MultiAddress is disabled
- failing with ConfigException if the user provides multiple
addresses in the config while having MultiAddress disabled
- updating the existing MultiAddress related tests to enable the
feature first
- add some new tests
- update the documentation

Testing:
- I ran all the unit tests
- Using https://github.com/symat/zk-rolling-upgrade-test
  - I tested rolling upgrade from 3.5.6
  - I tested rolling restart to enable the MultiAddress feature
- Using https://github.com/symat/zookeeper-docker-test
  - I tested the MultiAddress feature by disabling some virtual
    interfaces and waiting for the cluster to recover

Author: Mate Szalay-Beko <szalay.beko.mate@gmail.com>

Reviewers: Enrico Olivelli <eolivelli@apache.org>, Norbert Kalmar <nkalmar@apache.org>, Andor Molnar <andor@apache.org>

Closes #1251 from symat/ZOOKEEPER-3720

zookeeper-docs/src/main/resources/markdown/zookeeperAdmin.md

@@ -1211,7 +1211,8 @@ of servers -- that is, when deploying clusters of servers.
     <a name="id_multi_address"></a>
     Since ZooKeeper 3.6.0 it is possible to specify **multiple addresses** for each
     ZooKeeper server (see [ZOOKEEPER-3188](https://issues.apache.org/jira/projects/ZOOKEEPER/issues/ZOOKEEPER-3188)).
-    This helps to increase availability and adds network level 
+    To enable this feature, you must set the *multiAddress.enabled* configuration property
+    to *true*. This helps to increase availability and adds network level 
     resiliency to ZooKeeper. When multiple physical network interfaces are used 
     for the servers, ZooKeeper is able to bind on all interfaces and runtime switching 
     to a working interface in case a network error. The different addresses can be specified
@@ -1220,7 +1221,18 @@ of servers -- that is, when deploying clusters of servers.
         server.1=zoo1-net1:2888:3888|zoo1-net2:2889:3889
         server.2=zoo2-net1:2888:3888|zoo2-net2:2889:3889
         server.3=zoo3-net1:2888:3888|zoo3-net2:2889:3889
-       
+
+
+    ###### Note
+    >By enabling this feature, the Quorum protocol (ZooKeeper Server-Server protocol) will change.
+    The users will not notice this and when anyone starts a ZooKeeper cluster with the new config,
+    everything will work normally. However, it's not possible to enable this feature and specify
+    multiple addresses during a rolling upgrade if the old ZooKeeper cluster didn't support the
+    *multiAddress* feature (and the new Quorum protocol). In case if you need this feature but you
+    also need to perform a rolling upgrade from a ZooKeeper cluster older than *3.6.0*, then you
+    first need to do the rolling upgrade without enabling the MultiAddress feature and later make
+    a separate rolling restart with the new configuration where **multiAddress.enabled** is set
+    to **true** and multiple addresses are provided.
 
 * *syncLimit* :
     (No Java system property)
@@ -1584,6 +1596,16 @@ and [SASL authentication for ZooKeeper](https://cwiki.apache.org/confluence/disp
     (e.g. the zk/myhost@EXAMPLE.COM client principal will be authenticated in ZooKeeper as zk/myhost)
     Default: false
 
+* *multiAddress.enabled* :
+    (Java system property: **zookeeper.multiAddress.enabled**)
+    **New in 3.6.0:**
+    Since ZooKeeper 3.6.0 you can also [specify multiple addresses](#id_multi_address) 
+    for each ZooKeeper server instance (this can increase availability when multiple physical 
+    network interfaces can be used parallel in the cluster). Setting this parameter to 
+    **true** will enable this feature. Please note, that you can not enable this feature
+    during a rolling upgrade if the version of the old ZooKeeper cluster is prior to 3.6.0.
+    The default value is **false**.
+
 * *multiAddress.reachabilityCheckTimeoutMs* :
     (Java system property: **zookeeper.multiAddress.reachabilityCheckTimeoutMs**)
     **New in 3.6.0:**
@@ -1596,7 +1618,8 @@ and [SASL authentication for ZooKeeper](https://cwiki.apache.org/confluence/disp
     in parallel for the different addresses, so the timeout you set here is the maximum time will be taken
     by checking the reachability of all addresses.
     The default value is **1000**.
-    
+
+    This parameter has no effect, unless you enable the MultiAddress feature by setting *multiAddress.enabled=true*.    
 
 <a name="Experimental+Options%2FFeatures"></a>
 
@@ -1688,6 +1711,7 @@ the variable does.
     Please note, disabling the reachability check will cause the cluster not to be able to reconfigure 
     itself properly during network problems, so the disabling is advised only during testing. 
 
+    This parameter has no effect, unless you enable the MultiAddress feature by setting *multiAddress.enabled=true*.
 
 <a name="Disabling+data+directory+autocreation"></a>
 

zookeeper-server/src/main/java/org/apache/zookeeper/server/quorum/QuorumCnxManager.java

@@ -37,6 +37,8 @@ import java.nio.ByteBuffer;
 import java.nio.channels.UnresolvedAddressException;
 import java.time.Duration;
 import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collection;
 import java.util.Collections;
 import java.util.Enumeration;
 import java.util.HashSet;
@@ -113,9 +115,12 @@ public class QuorumCnxManager {
     private AtomicLong observerCounter = new AtomicLong(-1);
 
     /*
-     * Protocol identifier used among peers
+     * Protocol identifier used among peers (must be a negative number for backward compatibility reasons)
      */
-    public static final long PROTOCOL_VERSION = -65535L;
+    // the following protocol version was sent in every connection initiation message since ZOOKEEPER-107 released in 3.5.0
+    public static final long PROTOCOL_VERSION_V1 = -65536L;
+    // ZOOKEEPER-3188 introduced multiple addresses in the connection initiation message, released in 3.6.0
+    public static final long PROTOCOL_VERSION_V2 = -65535L;
 
     /*
      * Max buffer size to be read from the network.
@@ -218,7 +223,7 @@ public class QuorumCnxManager {
         public static InitialMessage parse(Long protocolVersion, DataInputStream din) throws InitialMessageException, IOException {
             Long sid;
 
-            if (protocolVersion != PROTOCOL_VERSION) {
+            if (protocolVersion != PROTOCOL_VERSION_V1 && protocolVersion != PROTOCOL_VERSION_V2) {
                 throw new InitialMessageException("Got unrecognized protocol version %s", protocolVersion);
             }
 
@@ -236,6 +241,8 @@ public class QuorumCnxManager {
                 throw new InitialMessageException("Read only %s bytes out of %s sent by server %s", num_read, remaining, sid);
             }
 
+            // in PROTOCOL_VERSION_V1 we expect to get a single address here represented as a 'host:port' string
+            // in PROTOCOL_VERSION_V2 we expect to get multiple addresses like: 'host1:port1|host2:port2|...'
             String[] addressStrings = new String(b).split("\\|");
             List<InetSocketAddress> addresses = new ArrayList<>(addressStrings.length);
             for (String addr : addressStrings) {
@@ -416,10 +423,20 @@ public class QuorumCnxManager {
 
             // Sending id and challenge
 
-            // represents protocol version (in other words - message type)
-            dout.writeLong(PROTOCOL_VERSION);
+            // First sending the protocol version (in other words - message type).
+            // For backward compatibility reasons we stick to the old protocol version, unless the MultiAddress
+            // feature is enabled. During rolling upgrade, we must make sure that all the servers can
+            // understand the protocol version we use to avoid multiple partitions. see ZOOKEEPER-3720
+            long protocolVersion = self.isMultiAddressEnabled() ? PROTOCOL_VERSION_V2 : PROTOCOL_VERSION_V1;
+            dout.writeLong(protocolVersion);
             dout.writeLong(self.getId());
-            String addr = self.getElectionAddress().getAllAddresses().stream()
+
+            // now we send our election address. For the new protocol version, we can send multiple addresses.
+            Collection<InetSocketAddress> addressesToSend = protocolVersion == PROTOCOL_VERSION_V2
+                    ? self.getElectionAddress().getAllAddresses()
+                    : Arrays.asList(self.getElectionAddress().getOne());
+
+            String addr = addressesToSend.stream()
                     .map(NetUtils::formatInetAddr).collect(Collectors.joining("|"));
             byte[] addr_bytes = addr.getBytes();
             dout.writeInt(addr_bytes.length);
@@ -639,7 +656,7 @@ public class QuorumCnxManager {
     synchronized boolean connectOne(long sid, MultipleAddresses electionAddr) {
         if (senderWorkerMap.get(sid) != null) {
             LOG.debug("There is a connection already for server {}", sid);
-            if (electionAddr.size() > 1 && self.isMultiAddressReachabilityCheckEnabled()) {
+            if (self.isMultiAddressEnabled() && electionAddr.size() > 1 && self.isMultiAddressReachabilityCheckEnabled()) {
                 // since ZOOKEEPER-3188 we can use multiple election addresses to reach a server. It is possible, that the
                 // one we are using is already dead and we need to clean-up, so when we will create a new connection
                 // then we will choose an other one, which is actually reachable
@@ -710,7 +727,7 @@ public class QuorumCnxManager {
     synchronized void connectOne(long sid) {
         if (senderWorkerMap.get(sid) != null) {
             LOG.debug("There is a connection already for server {}", sid);
-            if (self.isMultiAddressReachabilityCheckEnabled()) {
+            if (self.isMultiAddressEnabled() && self.isMultiAddressReachabilityCheckEnabled()) {
                 // since ZOOKEEPER-3188 we can use multiple election addresses to reach a server. It is possible, that the
                 // one we are using is already dead and we need to clean-up, so when we will create a new connection
                 // then we will choose an other one, which is actually reachable

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

@@ -161,8 +161,22 @@ public class QuorumPeer extends ZooKeeperThread implements QuorumStats.Provider
         this.observerMasterPort = observerMasterPort;
     }
 
-    private int multiAddressReachabilityCheckTimeoutMs = (int) MultipleAddresses.DEFAULT_TIMEOUT.toMillis();
+    public static final String CONFIG_KEY_MULTI_ADDRESS_ENABLED = "zookeeper.multiAddress.enabled";
+    public static final String CONFIG_DEFAULT_MULTI_ADDRESS_ENABLED = "false";
+
+    private boolean multiAddressEnabled = true;
+    public boolean isMultiAddressEnabled() {
+        return multiAddressEnabled;
+    }
+
+    public void setMultiAddressEnabled(boolean multiAddressEnabled) {
+        this.multiAddressEnabled = multiAddressEnabled;
+        LOG.info("multiAddress.enabled set to {}", multiAddressEnabled);
+    }
 
+    public static final String CONFIG_KEY_MULTI_ADDRESS_REACHABILITY_CHECK_TIMEOUT_MS = "zookeeper.multiAddress.reachabilityCheckTimeoutMs";
+
+    private int multiAddressReachabilityCheckTimeoutMs = (int) MultipleAddresses.DEFAULT_TIMEOUT.toMillis();
     public int getMultiAddressReachabilityCheckTimeoutMs() {
         return multiAddressReachabilityCheckTimeoutMs;
     }
@@ -172,6 +186,8 @@ public class QuorumPeer extends ZooKeeperThread implements QuorumStats.Provider
         LOG.info("multiAddress.reachabilityCheckTimeoutMs set to {}", multiAddressReachabilityCheckTimeoutMs);
     }
 
+    public static final String CONFIG_KEY_MULTI_ADDRESS_REACHABILITY_CHECK_ENABLED = "zookeeper.multiAddress.reachabilityCheckEnabled";
+
     private boolean multiAddressReachabilityCheckEnabled = true;
 
     public boolean isMultiAddressReachabilityCheckEnabled() {
@@ -274,6 +290,12 @@ public class QuorumPeer extends ZooKeeperThread implements QuorumStats.Provider
                 }
             }
 
+            boolean multiAddressEnabled = Boolean.parseBoolean(
+                System.getProperty(QuorumPeer.CONFIG_KEY_MULTI_ADDRESS_ENABLED, QuorumPeer.CONFIG_DEFAULT_MULTI_ADDRESS_ENABLED));
+            if (!multiAddressEnabled && serverAddresses.length > 1) {
+                throw new ConfigException("Multiple address feature is disabled, but multiple addresses were specified for sid " + sid);
+            }
+
             for (String serverAddress : serverAddresses) {
                 String serverParts[] = ConfigUtils.getHostAndPort(serverAddress);
                 if ((serverClientParts.length > 2) || (serverParts.length < 3)

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

@@ -121,13 +121,13 @@ public class QuorumPeerConfig {
     protected int quorumCnxnThreadsSize;
 
     // multi address related configs
+    private boolean multiAddressEnabled = Boolean.parseBoolean(
+        System.getProperty(QuorumPeer.CONFIG_KEY_MULTI_ADDRESS_ENABLED, QuorumPeer.CONFIG_DEFAULT_MULTI_ADDRESS_ENABLED));
     private boolean multiAddressReachabilityCheckEnabled =
-      Boolean.parseBoolean(System.getProperty("zookeeper.multiAddress.reachabilityCheckEnabled",
-                                              "true"));
+        Boolean.parseBoolean(System.getProperty(QuorumPeer.CONFIG_KEY_MULTI_ADDRESS_REACHABILITY_CHECK_ENABLED, "true"));
     private int multiAddressReachabilityCheckTimeoutMs =
-      Integer.parseInt(System.getProperty("zookeeper.multiAddress.reachabilityCheckTimeoutMs",
-                                          String.valueOf(MultipleAddresses.DEFAULT_TIMEOUT.toMillis())));
-
+        Integer.parseInt(System.getProperty(QuorumPeer.CONFIG_KEY_MULTI_ADDRESS_REACHABILITY_CHECK_TIMEOUT_MS,
+                                            String.valueOf(MultipleAddresses.DEFAULT_TIMEOUT.toMillis())));
 
     /**
      * Minimum snapshot retain count.
@@ -398,6 +398,8 @@ public class QuorumPeerConfig {
             } else if (key.startsWith("metricsProvider.")) {
                 String keyForMetricsProvider = key.substring(16);
                 metricsProviderConfiguration.put(keyForMetricsProvider, value);
+            } else if (key.equals("multiAddress.enabled")) {
+                multiAddressEnabled = Boolean.parseBoolean(value);
             } else if (key.equals("multiAddress.reachabilityCheckTimeoutMs")) {
                 multiAddressReachabilityCheckTimeoutMs = Integer.parseInt(value);
             } else if (key.equals("multiAddress.reachabilityCheckEnabled")) {
@@ -939,6 +941,10 @@ public class QuorumPeerConfig {
         return quorumListenOnAllIPs;
     }
 
+    public boolean isMultiAddressEnabled() {
+        return multiAddressEnabled;
+    }
+
     public boolean isMultiAddressReachabilityCheckEnabled() {
         return multiAddressReachabilityCheckEnabled;
     }

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

@@ -204,6 +204,7 @@ public class QuorumPeerMain {
             if (config.sslQuorumReloadCertFiles) {
                 quorumPeer.getX509Util().enableCertFileReloading();
             }
+            quorumPeer.setMultiAddressEnabled(config.isMultiAddressEnabled());
             quorumPeer.setMultiAddressReachabilityCheckEnabled(config.isMultiAddressReachabilityCheckEnabled());
             quorumPeer.setMultiAddressReachabilityCheckTimeoutMs(config.getMultiAddressReachabilityCheckTimeoutMs());
 

zookeeper-server/src/test/java/org/apache/zookeeper/server/quorum/CnxManagerTest.java

@@ -261,7 +261,7 @@ public class CnxManagerTest extends ZKTestCase {
 
         InetSocketAddress otherAddr = peers.get(2L).electionAddr.getReachableOrOne();
         DataOutputStream dout = new DataOutputStream(sc.socket().getOutputStream());
-        dout.writeLong(QuorumCnxManager.PROTOCOL_VERSION);
+        dout.writeLong(QuorumCnxManager.PROTOCOL_VERSION_V1);
         dout.writeLong(2);
         String addr = otherAddr.getHostString() + ":" + otherAddr.getPort();
         byte[] addr_bytes = addr.getBytes();
@@ -604,7 +604,7 @@ public class CnxManagerTest extends ZKTestCase {
             dout.writeBytes(hostport);
 
             din = new DataInputStream(new ByteArrayInputStream(bos.toByteArray()));
-            msg = InitialMessage.parse(QuorumCnxManager.PROTOCOL_VERSION, din);
+            msg = InitialMessage.parse(QuorumCnxManager.PROTOCOL_VERSION_V1, din);
             fail("long message accepted");
         } catch (InitialMessage.InitialMessageException ex) {
         }
@@ -620,7 +620,7 @@ public class CnxManagerTest extends ZKTestCase {
             dout.writeBytes(hostport);
 
             din = new DataInputStream(new ByteArrayInputStream(bos.toByteArray()));
-            msg = InitialMessage.parse(QuorumCnxManager.PROTOCOL_VERSION, din);
+            msg = InitialMessage.parse(QuorumCnxManager.PROTOCOL_VERSION_V1, din);
             fail("bad hostport accepted");
         } catch (InitialMessage.InitialMessageException ex) {
         }
@@ -637,7 +637,7 @@ public class CnxManagerTest extends ZKTestCase {
 
             // now parse it
             din = new DataInputStream(new ByteArrayInputStream(bos.toByteArray()));
-            msg = InitialMessage.parse(QuorumCnxManager.PROTOCOL_VERSION, din);
+            msg = InitialMessage.parse(QuorumCnxManager.PROTOCOL_VERSION_V1, din);
             assertEquals(Long.valueOf(5), msg.sid);
             assertEquals(Arrays.asList(new InetSocketAddress("10.0.0.2", 3888)), msg.electionAddr);
         } catch (InitialMessage.InitialMessageException ex) {
@@ -656,7 +656,7 @@ public class CnxManagerTest extends ZKTestCase {
 
             // now parse it
             din = new DataInputStream(new ByteArrayInputStream(bos.toByteArray()));
-            msg = InitialMessage.parse(QuorumCnxManager.PROTOCOL_VERSION, din);
+            msg = InitialMessage.parse(QuorumCnxManager.PROTOCOL_VERSION_V2, din);
             assertEquals(Long.valueOf(5), msg.sid);
             assertEquals(Arrays.asList(new InetSocketAddress("1.1.1.1", 9999),
                                        new InetSocketAddress("2.2.2.2", 8888),

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

@@ -53,6 +53,7 @@ import org.apache.zookeeper.test.TestUtils;
 import org.apache.zookeeper.txn.CreateTxn;
 import org.apache.zookeeper.txn.TxnHeader;
 import org.apache.zookeeper.util.ServiceUtils;
+import org.junit.After;
 import org.junit.Test;
 
 public class LearnerTest extends ZKTestCase {
@@ -134,6 +135,11 @@ public class LearnerTest extends ZKTestCase {
         }
     }
 
+    @After
+    public void cleanup() {
+        System.clearProperty(QuorumPeer.CONFIG_KEY_MULTI_ADDRESS_ENABLED);
+    }
+
     @Test(expected = IOException.class)
     public void connectionRetryTimeoutTest() throws Exception {
         Learner learner = new TestLearner();
@@ -178,6 +184,7 @@ public class LearnerTest extends ZKTestCase {
 
     @Test
     public void shouldTryMultipleAddresses() throws Exception {
+        System.setProperty(QuorumPeer.CONFIG_KEY_MULTI_ADDRESS_ENABLED, "true");
         TestLearner learner = new TestLearner();
         learner.self = new QuorumPeer();
         learner.self.setTickTime(2000);
@@ -205,6 +212,7 @@ public class LearnerTest extends ZKTestCase {
 
     @Test
     public void multipleAddressesSomeAreFailing() throws Exception {
+        System.setProperty(QuorumPeer.CONFIG_KEY_MULTI_ADDRESS_ENABLED, "true");
         TestLearner learner = new TestLearner();
         learner.self = new QuorumPeer();
         learner.self.setTickTime(2000);

zookeeper-server/src/test/java/org/apache/zookeeper/server/quorum/QuorumPeerMainMultiAddressTest.java

@@ -19,6 +19,7 @@
 package org.apache.zookeeper.server.quorum;
 
 import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.fail;
 import java.io.IOException;
 import java.util.Arrays;
 import java.util.List;
@@ -52,6 +53,7 @@ public class QuorumPeerMainMultiAddressTest extends QuorumPeerTestBase {
 
   @Before
   public void setUp() throws Exception {
+    System.setProperty(QuorumPeer.CONFIG_KEY_MULTI_ADDRESS_ENABLED, "true");
     ClientBase.setupTestEnv();
     System.setProperty("zookeeper.DigestAuthenticationProvider.superDigest", "super:D/InIHSb7yEEbrWz8b9l71RjZJU="/* password is 'test'*/);
     QuorumPeerConfig.setReconfigEnabled(true);
@@ -63,6 +65,7 @@ public class QuorumPeerMainMultiAddressTest extends QuorumPeerTestBase {
   @After
   public void tearDown() throws Exception {
     super.tearDown();
+    System.clearProperty(QuorumPeer.CONFIG_KEY_MULTI_ADDRESS_ENABLED);
     System.clearProperty("zookeeper.jmx.log4j.disable");
   }
 
@@ -230,6 +233,34 @@ public class QuorumPeerMainMultiAddressTest extends QuorumPeerTestBase {
     checkIfZooKeeperQuorumWorks(newQuorumConfig);
   }
 
+  @Test
+  public void shouldFailToReconfigWithMultipleAddressesWhenFeatureIsDisabled() throws Exception {
+    System.setProperty(QuorumPeer.CONFIG_KEY_MULTI_ADDRESS_ENABLED, "false");
+
+    // we have three ZK servers, each server has a single quorumPort and single electionPort registered
+    QuorumServerConfigBuilder initialQuorumConfig = new QuorumServerConfigBuilder(hostName, 3, 1);
+
+    // we launch the three servers, each server should use the same initial config
+    launchServers(Arrays.asList(initialQuorumConfig, initialQuorumConfig, initialQuorumConfig));
+
+    checkIfZooKeeperQuorumWorks(initialQuorumConfig);
+
+    // we create a new config where we add a new address to one of the servers with random available ports
+    QuorumServerConfigBuilder newQuorumConfig = new QuorumServerConfigBuilder(initialQuorumConfig)
+            .addNewServerAddress(FIRST_SERVER);
+
+    ZooKeeperAdmin zkAdmin = newZooKeeperAdmin(initialQuorumConfig);
+
+    // initiating a new incremental reconfig by using the updated ports, expecting exceptions here
+    try {
+      ReconfigTest.reconfig(zkAdmin, newQuorumConfig.buildAsStringList(), null, null, -1);
+      fail("Reconfig succeeded with multiple addresses without exception when the MultiAddress feature is disabled");
+    } catch (KeeperException.BadArgumentsException e) {
+      // do nothing, this is what we expected
+    } catch (Exception e) {
+      fail("Reconfig failed in a wrong way. We expected KeeperException.BadArgumentsException.");
+    }
+  }
 
   private void launchServers(List<QuorumServerConfigBuilder> builders) throws IOException, InterruptedException {
 

zookeeper-server/src/test/java/org/apache/zookeeper/server/quorum/QuorumSSLTest.java

@@ -430,6 +430,7 @@ public class QuorumSSLTest extends QuorumPeerTestBase {
 
     @After
     public void cleanUp() throws Exception {
+        System.clearProperty(QuorumPeer.CONFIG_KEY_MULTI_ADDRESS_ENABLED);
         clearSSLSystemProperties();
         if (q1 != null) {
             q1.shutdown();
@@ -480,6 +481,7 @@ public class QuorumSSLTest extends QuorumPeerTestBase {
 
     @Test
     public void testQuorumSSLWithMultipleAddresses() throws Exception {
+        System.setProperty(QuorumPeer.CONFIG_KEY_MULTI_ADDRESS_ENABLED, "true");
         quorumConfiguration = generateMultiAddressQuorumConfiguration();
 
         q1 = new MainThread(1, clientPortQp1, quorumConfiguration, SSL_QUORUM_ENABLED);
@@ -598,6 +600,7 @@ public class QuorumSSLTest extends QuorumPeerTestBase {
 
     @Test
     public void testHostnameVerificationForInvalidMultiAddressServerConfig() throws Exception {
+        System.setProperty(QuorumPeer.CONFIG_KEY_MULTI_ADDRESS_ENABLED, "true");
         quorumConfiguration = generateMultiAddressQuorumConfiguration();
 
         String badhostnameKeystorePath = tmpDir + "/badhost.jks";

zookeeper-server/src/test/java/org/apache/zookeeper/server/quorum/QuorumServerTest.java

@@ -23,6 +23,7 @@ import org.apache.zookeeper.KeeperException;
 import org.apache.zookeeper.ZKTestCase;
 import org.apache.zookeeper.server.quorum.QuorumPeer.QuorumServer;
 import org.apache.zookeeper.server.quorum.QuorumPeerConfig.ConfigException;
+import org.junit.After;
 import org.junit.Test;
 
 public class QuorumServerTest extends ZKTestCase {
@@ -31,6 +32,11 @@ public class QuorumServerTest extends ZKTestCase {
     private String ipv6n2 = "[2600:0:0:0:0:0:1:0]";
     private String ipv4config = "127.0.0.1:1234:1236";
 
+    @After
+    public void tearDown() {
+        System.clearProperty(QuorumPeer.CONFIG_KEY_MULTI_ADDRESS_ENABLED);
+    }
+
     @Test
     public void testToString() throws ConfigException {
         String provided = ipv4config + ":participant;0.0.0.0:1237";
@@ -88,6 +94,19 @@ public class QuorumServerTest extends ZKTestCase {
         new QuorumServer(0, ipv4config + ":participant;[::1:1237");
     }
 
+    @Test(expected = ConfigException.class)
+    public void shouldNotAllowMultipleAddressesWhenMultiAddressFeatureIsDisabled() throws ConfigException {
+        System.setProperty(QuorumPeer.CONFIG_KEY_MULTI_ADDRESS_ENABLED, "false");
+        new QuorumServer(0, "127.0.0.1:1234:1236|127.0.0.1:2234:2236");
+    }
+
+    @Test
+    public void shouldAllowMultipleAddressesWhenMultiAddressFeatureIsEnabled() throws ConfigException {
+        System.setProperty(QuorumPeer.CONFIG_KEY_MULTI_ADDRESS_ENABLED, "true");
+        QuorumServer qs = new QuorumServer(0, "127.0.0.1:1234:1236|127.0.0.1:2234:2236");
+        assertEquals("MultiAddress parse error", "127.0.0.1:1234:1236|127.0.0.1:2234:2236:participant", qs.toString());
+    }
+
     @Test
     public void testWildcard() throws KeeperException.BadArgumentsException {
         String[] addrs = new String[]{"127.0.0.1", "[0:0:0:0:0:0:0:1]", "0.0.0.0", "[::]"};

zookeeper-server/src/test/java/org/apache/zookeeper/server/quorum/auth/QuorumAuthTestBase.java

@@ -29,8 +29,10 @@ import javax.security.auth.login.Configuration;
 import org.apache.commons.io.FileUtils;
 import org.apache.zookeeper.PortAssignment;
 import org.apache.zookeeper.ZKTestCase;
+import org.apache.zookeeper.server.quorum.QuorumPeer;
 import org.apache.zookeeper.server.quorum.QuorumPeerTestBase.MainThread;
 import org.apache.zookeeper.test.ClientBase;
+import org.junit.After;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -69,6 +71,11 @@ public class QuorumAuthTestBase extends ZKTestCase {
         }
     }
 
+    @After
+    public void tearDown() throws Exception {
+        System.clearProperty(QuorumPeer.CONFIG_KEY_MULTI_ADDRESS_ENABLED);
+    }
+
     protected String startQuorum(final int serverCount, Map<String, String> authConfigs,
         int authServerCount) throws IOException {
         return this.startQuorum(serverCount, authConfigs, authServerCount, false);
@@ -76,6 +83,7 @@ public class QuorumAuthTestBase extends ZKTestCase {
 
     protected String startMultiAddressQuorum(final int serverCount, Map<String, String> authConfigs,
         int authServerCount) throws IOException {
+        System.setProperty(QuorumPeer.CONFIG_KEY_MULTI_ADDRESS_ENABLED, "true");
         return this.startQuorum(serverCount, authConfigs, authServerCount, true);
     }
 

zookeeper-server/src/test/java/org/apache/zookeeper/server/quorum/auth/QuorumAuthUpgradeTest.java

@@ -67,8 +67,10 @@ public class QuorumAuthUpgradeTest extends QuorumAuthTestBase {
     }
 
     @After
+    @Override
     public void tearDown() throws Exception {
         shutdownAll();
+        super.tearDown();
     }
 
     @AfterClass

zookeeper-server/src/test/java/org/apache/zookeeper/server/quorum/auth/QuorumDigestAuthTest.java

@@ -60,11 +60,13 @@ public class QuorumDigestAuthTest extends QuorumAuthTestBase {
     }
 
     @After
+    @Override
     public void tearDown() throws Exception {
         for (MainThread mainThread : mt) {
             mainThread.shutdown();
             mainThread.deleteBaseDir();
         }
+        super.tearDown();
     }
 
     @AfterClass

zookeeper-server/src/test/java/org/apache/zookeeper/server/quorum/auth/QuorumKerberosAuthTest.java

@@ -88,11 +88,13 @@ public class QuorumKerberosAuthTest extends KerberosSecurityTestcase {
     }
 
     @After
+    @Override
     public void tearDown() throws Exception {
         for (MainThread mainThread : mt) {
             mainThread.shutdown();
             mainThread.deleteBaseDir();
         }
+        super.tearDown();
     }
 
     @AfterClass

zookeeper-server/src/test/java/org/apache/zookeeper/server/quorum/auth/QuorumKerberosHostBasedAuthTest.java

@@ -113,11 +113,13 @@ public class QuorumKerberosHostBasedAuthTest extends KerberosSecurityTestcase {
     }
 
     @After
+    @Override
     public void tearDown() throws Exception {
         for (MainThread mainThread : mt) {
             mainThread.shutdown();
             mainThread.deleteBaseDir();
         }
+        super.tearDown();
     }
 
     @AfterClass