ZOOKEEPER-3188: support for dynamic reconfig + add more unit tests

Mate Szalay-Beko · Mate Szalay-Beko · commit a5d6bcb9722b · 2019-10-08T15:45:59.000+02:00
diff --git a/zookeeper-server/src/main/java/org/apache/zookeeper/server/quorum/Leader.java b/zookeeper-server/src/main/java/org/apache/zookeeper/server/quorum/Leader.java
@@ -434,7 +434,7 @@ class LearnerCnxAcceptor extends ZooKeeperCriticalThread {
             super("LearnerCnxAcceptor-" + serverSockets.stream()
                       .map(ServerSocket::getLocalSocketAddress)
                       .map(Objects::toString)
-                      .collect(Collectors.joining(",")),
+                      .collect(Collectors.joining("|")),
                   zk.getZooKeeperServerListener());
         }
 
diff --git a/zookeeper-server/src/main/java/org/apache/zookeeper/server/quorum/Learner.java b/zookeeper-server/src/main/java/org/apache/zookeeper/server/quorum/Learner.java
@@ -381,7 +381,11 @@ private Socket connectToLeader() throws IOException, X509Exception, InterruptedE
         }
     }
 
-    private Socket createSocket() throws X509Exception, IOException {
+    /**
+     * Creating a simple or and SSL socket.
+     * This can be overridden in tests to fake already connected sockets for connectToLeader.
+     */
+    protected Socket createSocket() throws X509Exception, IOException {
         Socket sock;
         if (self.isSslQuorum()) {
             sock = self.getX509Util().createSSLSocket();
diff --git a/zookeeper-server/src/main/java/org/apache/zookeeper/server/quorum/LocalPeerBean.java b/zookeeper-server/src/main/java/org/apache/zookeeper/server/quorum/LocalPeerBean.java
@@ -83,7 +83,7 @@ public String getState() {
 
     public String getQuorumAddress() {
         return peer.getQuorumAddress().getAllAddresses().stream().map(NetUtils::formatInetAddr)
-                .collect(Collectors.joining(","));
+                .collect(Collectors.joining("|"));
     }
 
     public int getElectionType() {
@@ -92,7 +92,7 @@ public int getElectionType() {
 
     public String getElectionAddress() {
         return peer.getElectionAddress().getAllAddresses().stream().map(NetUtils::formatInetAddr)
-                .collect(Collectors.joining(","));
+                .collect(Collectors.joining("|"));
     }
 
     public String getClientAddress() {
diff --git a/zookeeper-server/src/main/java/org/apache/zookeeper/server/quorum/MultipleAddresses.java b/zookeeper-server/src/main/java/org/apache/zookeeper/server/quorum/MultipleAddresses.java
@@ -208,6 +208,6 @@ public int hashCode() {
 
     @Override
     public String toString() {
-        return addresses.stream().map(InetSocketAddress::toString).collect(Collectors.joining(","));
+        return addresses.stream().map(InetSocketAddress::toString).collect(Collectors.joining("|"));
     }
 }
diff --git a/zookeeper-server/src/main/java/org/apache/zookeeper/server/quorum/QuorumCnxManager.java b/zookeeper-server/src/main/java/org/apache/zookeeper/server/quorum/QuorumCnxManager.java
@@ -238,7 +238,7 @@ public static InitialMessage parse(Long protocolVersion, DataInputStream din) th
                 throw new InitialMessageException("Read only %s bytes out of %s sent by server %s", num_read, remaining, sid);
             }
 
-            String[] addressStrings = new String(b).split(",");
+            String[] addressStrings = new String(b).split("\\|");
             List<InetSocketAddress> addresses = new ArrayList<>(addressStrings.length);
             for (String addr : addressStrings) {
 
@@ -420,7 +420,7 @@ private boolean startConnection(Socket sock, Long sid) throws IOException {
             dout.writeLong(PROTOCOL_VERSION);
             dout.writeLong(self.getId());
             String addr = self.getElectionAddress().getAllAddresses().stream()
-                    .map(NetUtils::formatInetAddr).collect(Collectors.joining(","));
+                    .map(NetUtils::formatInetAddr).collect(Collectors.joining("|"));
             byte[] addr_bytes = addr.getBytes();
             dout.writeInt(addr_bytes.length);
             dout.write(addr_bytes);
@@ -935,7 +935,7 @@ public void run() {
                         + "I won't be able to participate in leader "
                         + "election any longer: {}"
                         , self.getElectionAddress().getAllAddresses().stream().map(NetUtils::formatInetAddr)
-                                .collect(Collectors.joining(",")));
+                                .collect(Collectors.joining("|")));
                 if (socketException.get()) {
                     // After leaving listener thread, the host cannot join the quorum anymore,
                     // this is a severe error that we cannot recover from, so we need to exit
diff --git a/zookeeper-server/src/main/java/org/apache/zookeeper/server/quorum/QuorumPeer.java b/zookeeper-server/src/main/java/org/apache/zookeeper/server/quorum/QuorumPeer.java
@@ -228,13 +228,14 @@ private LearnerType getType(String s) throws ConfigException {
 
         private static final String wrongFormat =
             " does not have the form server_config or server_config;client_config"
-            + " where server_config is host:port:port or host:port:port:type and client_config is port or host:port";
+            + " where server_config is the pipe separated list of host:port:port or host:port:port:type"
+            + " and client_config is port or host:port";
 
         public QuorumServer(long sid, String addressStr) throws ConfigException {
             this.id = sid;
             LearnerType newType = null;
             String[] serverClientParts = addressStr.split(";");
-            String[] serverAddresses = serverClientParts[0].split(",");
+            String[] serverAddresses = serverClientParts[0].split("\\|");
 
             if (serverClientParts.length == 2) {
                 String[] clientParts = ConfigUtils.getHostAndPort(serverClientParts[1]);
@@ -346,7 +347,7 @@ public String toString() {
                 electionAddrList.sort(Comparator.comparing(InetSocketAddress::getHostString));
                 sw.append(IntStream.range(0, addrList.size()).mapToObj(i -> String.format("%s:%d:%d",
                         delimitedHostString(addrList.get(i)), addrList.get(i).getPort(), electionAddrList.get(i).getPort()))
-                        .collect(Collectors.joining(",")));
+                        .collect(Collectors.joining("|")));
             }
 
             if (type == LearnerType.OBSERVER) {
@@ -833,7 +834,7 @@ public SyncMode getSyncMode() {
 
     public void setLeaderAddressAndId(MultipleAddresses addr, long newId) {
         if (addr != null) {
-            leaderAddress.set(String.join(",", addr.getAllHostStrings()));
+            leaderAddress.set(String.join("|", addr.getAllHostStrings()));
         } else {
             leaderAddress.set(null);
         }
diff --git a/zookeeper-server/src/main/java/org/apache/zookeeper/server/quorum/QuorumZooKeeperServer.java b/zookeeper-server/src/main/java/org/apache/zookeeper/server/quorum/QuorumZooKeeperServer.java
@@ -176,10 +176,10 @@ public void dumpConf(PrintWriter pwriter) {
         pwriter.println(self.getElectionType());
         pwriter.print("electionPort=");
         pwriter.println(self.getElectionAddress().getAllPorts()
-                .stream().map(Objects::toString).collect(Collectors.joining(",")));
+                .stream().map(Objects::toString).collect(Collectors.joining("|")));
         pwriter.print("quorumPort=");
         pwriter.println(self.getQuorumAddress().getAllPorts()
-                        .stream().map(Objects::toString).collect(Collectors.joining(",")));
+                        .stream().map(Objects::toString).collect(Collectors.joining("|")));
         pwriter.print("peerType=");
         pwriter.println(self.getLearnerType().ordinal());
         pwriter.println("membership: ");
diff --git a/zookeeper-server/src/main/java/org/apache/zookeeper/server/quorum/ReadOnlyZooKeeperServer.java b/zookeeper-server/src/main/java/org/apache/zookeeper/server/quorum/ReadOnlyZooKeeperServer.java
@@ -173,10 +173,10 @@ public void dumpConf(PrintWriter pwriter) {
         pwriter.println(self.getElectionType());
         pwriter.print("electionPort=");
         pwriter.println(self.getElectionAddress().getAllPorts()
-                .stream().map(Objects::toString).collect(Collectors.joining(",")));
+                .stream().map(Objects::toString).collect(Collectors.joining("|")));
         pwriter.print("quorumPort=");
         pwriter.println(self.getQuorumAddress().getAllPorts()
-                .stream().map(Objects::toString).collect(Collectors.joining(",")));
+                .stream().map(Objects::toString).collect(Collectors.joining("|")));
         pwriter.print("peerType=");
         pwriter.println(self.getLearnerType().ordinal());
     }
diff --git a/zookeeper-server/src/main/java/org/apache/zookeeper/server/quorum/RemotePeerBean.java b/zookeeper-server/src/main/java/org/apache/zookeeper/server/quorum/RemotePeerBean.java
@@ -49,13 +49,13 @@ public boolean isHidden() {
     public String getQuorumAddress() {
         return peer.addr.getAllAddresses().stream()
                 .map(address -> String.format("%s:%d", address.getHostString(), address.getPort()))
-                .collect(Collectors.joining(","));
+                .collect(Collectors.joining("|"));
     }
 
     public String getElectionAddress() {
         return peer.electionAddr.getAllAddresses().stream()
                 .map(address -> String.format("%s:%d", address.getHostString(), address.getPort()))
-                .collect(Collectors.joining(","));
+                .collect(Collectors.joining("|"));
     }
 
     public String getClientAddress() {
diff --git a/zookeeper-server/src/test/java/org/apache/zookeeper/server/quorum/CnxManagerTest.java b/zookeeper-server/src/test/java/org/apache/zookeeper/server/quorum/CnxManagerTest.java
@@ -647,7 +647,7 @@ public void testInitialMessage() throws Exception {
         // good message, multiple election addresses (ZOOKEEPER-3188)
         try {
 
-            hostport = "1.1.1.1:9999,2.2.2.2:8888,3.3.3.3:7777";
+            hostport = "1.1.1.1:9999|2.2.2.2:8888|3.3.3.3:7777";
             bos = new ByteArrayOutputStream();
             dout = new DataOutputStream(bos);
             dout.writeLong(5L); // sid
diff --git a/zookeeper-server/src/test/java/org/apache/zookeeper/server/quorum/LearnerTest.java b/zookeeper-server/src/test/java/org/apache/zookeeper/server/quorum/LearnerTest.java
@@ -18,9 +18,13 @@
 
 package org.apache.zookeeper.server.quorum;
 
+import static java.util.Arrays.asList;
+import static java.util.Collections.emptySet;
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertTrue;
 import static org.junit.Assert.fail;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
 import java.io.BufferedOutputStream;
 import java.io.ByteArrayInputStream;
 import java.io.ByteArrayOutputStream;
@@ -30,10 +34,13 @@
 import java.net.InetSocketAddress;
 import java.net.Socket;
 import java.util.ArrayList;
+import java.util.HashSet;
+import java.util.Set;
 import org.apache.jute.BinaryInputArchive;
 import org.apache.jute.BinaryOutputArchive;
 import org.apache.zookeeper.ZKTestCase;
 import org.apache.zookeeper.ZooDefs;
+import org.apache.zookeeper.common.X509Exception;
 import org.apache.zookeeper.data.ACL;
 import org.apache.zookeeper.server.ZKDatabase;
 import org.apache.zookeeper.server.persistence.FileTxnSnapLog;
@@ -71,40 +78,59 @@ static class SimpleLearner extends Learner {
 
     }
 
-    static class TimeoutLearner extends Learner {
+    static class TestLearner extends Learner {
 
-        int passSocketConnectOnAttempt = 10;
-        int socketConnectAttempt = 0;
-        long timeMultiplier = 0;
+        private int passSocketConnectOnAttempt = 10;
+        private int socketConnectAttempt = 0;
+        private long timeMultiplier = 0;
+        private Socket socketToBeCreated = null;
+        private Set<InetSocketAddress> unreachableAddresses = emptySet();
 
-        public void setTimeMultiplier(long multiplier) {
+        private void setTimeMultiplier(long multiplier) {
             timeMultiplier = multiplier;
         }
 
-        public void setPassConnectAttempt(int num) {
+        private void setPassConnectAttempt(int num) {
             passSocketConnectOnAttempt = num;
         }
 
         protected long nanoTime() {
             return socketConnectAttempt * timeMultiplier;
         }
 
-        protected int getSockConnectAttempt() {
+        private int getSockConnectAttempt() {
             return socketConnectAttempt;
         }
 
+        private void setSocketToBeCreated(Socket socketToBeCreated) {
+            this.socketToBeCreated = socketToBeCreated;
+        }
+
+        private void setUnreachableAddresses(Set<InetSocketAddress> unreachableAddresses) {
+            this.unreachableAddresses = unreachableAddresses;
+        }
+
         @Override
         protected void sockConnect(Socket sock, InetSocketAddress addr, int timeout) throws IOException {
-            if (++socketConnectAttempt < passSocketConnectOnAttempt) {
-                throw new IOException("Test injected Socket.connect() error.");
+            synchronized (this) {
+                if (++socketConnectAttempt < passSocketConnectOnAttempt || unreachableAddresses.contains(addr)) {
+                    throw new IOException("Test injected Socket.connect() error.");
+                }
             }
         }
 
+        @Override
+        protected Socket createSocket() throws X509Exception, IOException {
+            if (socketToBeCreated != null) {
+                return socketToBeCreated;
+            }
+            return super.createSocket();
+        }
     }
 
     @Test(expected = IOException.class)
     public void connectionRetryTimeoutTest() throws Exception {
-        Learner learner = new TimeoutLearner();
+        Learner learner = new TestLearner();
         learner.self = new QuorumPeer();
         learner.self.setTickTime(2000);
         learner.self.setInitLimit(5);
@@ -119,7 +145,7 @@ public void connectionRetryTimeoutTest() throws Exception {
 
     @Test
     public void connectionInitLimitTimeoutTest() throws Exception {
-        TimeoutLearner learner = new TimeoutLearner();
+        TestLearner learner = new TestLearner();
         learner.self = new QuorumPeer();
         learner.self.setTickTime(2000);
         learner.self.setInitLimit(5);
@@ -144,9 +170,68 @@ public void connectionInitLimitTimeoutTest() throws Exception {
         }
     }
 
+    @Test
+    public void shouldTryMultipleAddresses() throws Exception {
+        TestLearner learner = new TestLearner();
+        learner.self = new QuorumPeer();
+        learner.self.setTickTime(2000);
+        learner.self.setInitLimit(5);
+        learner.self.setSyncLimit(2);
+
+        // this addr won't even be used since we fake the Socket.connect
+        InetSocketAddress addrA = new InetSocketAddress(1111);
+        InetSocketAddress addrB = new InetSocketAddress(2222);
+        InetSocketAddress addrC = new InetSocketAddress(3333);
+        InetSocketAddress addrD = new InetSocketAddress(4444);
+
+        // we will never pass (don't allow successful socker.connect) during this test
+        learner.setPassConnectAttempt(100);
+
+        // we expect this to throw an IOException since we're faking socket connect errors every time
+        try {
+            learner.connectToLeader(new MultipleAddresses(asList(addrA, addrB, addrC, addrD)), "");
+            fail("should have thrown IOException!");
+        } catch (IOException e) {
+            //good, wanted to see the IOException, let's make sure we tried each address 5 times
+            assertEquals(4 * 5, learner.getSockConnectAttempt());
+        }
+    }
+
+    @Test
+    public void multipleAddressesSomeAreFailing() throws Exception {
+        TestLearner learner = new TestLearner();
+        learner.self = new QuorumPeer();
+        learner.self.setTickTime(2000);
+        learner.self.setInitLimit(5);
+        learner.self.setSyncLimit(2);
+
+        // these addresses won't even be used since we fake the Socket.connect
+        InetSocketAddress addrWorking = new InetSocketAddress(1111);
+        InetSocketAddress addrBadA = new InetSocketAddress(2222);
+        InetSocketAddress addrBadB = new InetSocketAddress(3333);
+        InetSocketAddress addrBadC = new InetSocketAddress(4444);
+
+        // we will emulate socket connection error for each 'bad' address
+        learner.setUnreachableAddresses(new HashSet<>(asList(addrBadA, addrBadB, addrBadC)));
+
+        // all connection attempts should succeed (if it is not an unreachable address)
+        learner.setPassConnectAttempt(0);
+
+        // initialize a mock socket, created by the Learner
+        Socket mockSocket = mock(Socket.class);
+        when(mockSocket.isConnected()).thenReturn(true);
+        learner.setSocketToBeCreated(mockSocket);
+
+
+        // we expect this to not throw an IOException since there is a single working address
+        learner.connectToLeader(new MultipleAddresses(asList(addrBadA, addrBadB, addrBadC, addrWorking)), "");
+
+        assertEquals("Learner connected to the wrong address", learner.getSocket(), mockSocket);
+    }
+
     @Test
     public void connectToLearnerMasterLimitTest() throws Exception {
-        TimeoutLearner learner = new TimeoutLearner();
+        TestLearner learner = new TestLearner();
         learner.self = new QuorumPeer();
         learner.self.setTickTime(2000);
         learner.self.setInitLimit(2);
diff --git a/zookeeper-server/src/test/java/org/apache/zookeeper/server/quorum/QuorumPeerMainMultiAddressTest.java b/zookeeper-server/src/test/java/org/apache/zookeeper/server/quorum/QuorumPeerMainMultiAddressTest.java
diff --git a/zookeeper-server/src/test/java/org/apache/zookeeper/test/ReconfigTest.java b/zookeeper-server/src/test/java/org/apache/zookeeper/test/ReconfigTest.java

Original file line number	Diff line number	Diff line change
`@@ -434,7 +434,7 @@ class LearnerCnxAcceptor extends ZooKeeperCriticalThread {`
`434`	`434`	`super("LearnerCnxAcceptor-" + serverSockets.stream()`
`435`	`435`	`.map(ServerSocket::getLocalSocketAddress)`
`436`	`436`	`.map(Objects::toString)`
`437`		`- .collect(Collectors.joining(",")),`
	`437`	`+ .collect(Collectors.joining("\|")),`
`438`	`438`	`zk.getZooKeeperServerListener());`
`439`	`439`	`}`
`440`	`440`
Original file line number	Diff line number	Diff line change
`@@ -381,7 +381,11 @@ private Socket connectToLeader() throws IOException, X509Exception, InterruptedE`
`381`	`381`	`}`
`382`	`382`	`}`
`383`	`383`
`384`		`- private Socket createSocket() throws X509Exception, IOException {`
	`384`	`+ /**`
	`385`	`+ * Creating a simple or and SSL socket.`
	`386`	`+ * This can be overridden in tests to fake already connected sockets for connectToLeader.`
	`387`	`+ */`
	`388`	`+ protected Socket createSocket() throws X509Exception, IOException {`
`385`	`389`	`Socket sock;`
`386`	`390`	`if (self.isSslQuorum()) {`
`387`	`391`	`sock = self.getX509Util().createSSLSocket();`
Original file line number	Diff line number	Diff line change
`@@ -83,7 +83,7 @@ public String getState() {`
`83`	`83`
`84`	`84`	`public String getQuorumAddress() {`
`85`	`85`	`return peer.getQuorumAddress().getAllAddresses().stream().map(NetUtils::formatInetAddr)`
`86`		`- .collect(Collectors.joining(","));`
	`86`	`+ .collect(Collectors.joining("\|"));`
`87`	`87`	`}`
`88`	`88`
`89`	`89`	`public int getElectionType() {`
`@@ -92,7 +92,7 @@ public int getElectionType() {`
`92`	`92`
`93`	`93`	`public String getElectionAddress() {`
`94`	`94`	`return peer.getElectionAddress().getAllAddresses().stream().map(NetUtils::formatInetAddr)`
`95`		`- .collect(Collectors.joining(","));`
	`95`	`+ .collect(Collectors.joining("\|"));`
`96`	`96`	`}`
`97`	`97`
`98`	`98`	`public String getClientAddress() {`
Original file line number	Diff line number	Diff line change
`@@ -208,6 +208,6 @@ public int hashCode() {`
`208`	`208`
`209`	`209`	`@Override`
`210`	`210`	`public String toString() {`
`211`		`- return addresses.stream().map(InetSocketAddress::toString).collect(Collectors.joining(","));`
	`211`	`+ return addresses.stream().map(InetSocketAddress::toString).collect(Collectors.joining("\|"));`
`212`	`212`	`}`
`213`	`213`	`}`
Original file line number	Diff line number	Diff line change
`@@ -49,13 +49,13 @@ public boolean isHidden() {`
`49`	`49`	`public String getQuorumAddress() {`
`50`	`50`	`return peer.addr.getAllAddresses().stream()`
`51`	`51`	`.map(address -> String.format("%s:%d", address.getHostString(), address.getPort()))`
`52`		`- .collect(Collectors.joining(","));`
	`52`	`+ .collect(Collectors.joining("\|"));`
`53`	`53`	`}`
`54`	`54`
`55`	`55`	`public String getElectionAddress() {`
`56`	`56`	`return peer.electionAddr.getAllAddresses().stream()`
`57`	`57`	`.map(address -> String.format("%s:%d", address.getHostString(), address.getPort()))`
`58`		`- .collect(Collectors.joining(","));`
	`58`	`+ .collect(Collectors.joining("\|"));`
`59`	`59`	`}`
`60`	`60`
`61`	`61`	`public String getClientAddress() {`