ZOOKEEPER-2164: Quorum members can not rejoin after restart

Ever since ZOOKEEPER-107 (released in 3.5.0) the servers are sending
their addresses during initial connection requests. The receiving
server can potentially use these addresses to send back a new
connection request if the challenge is won by the receiver.

If the server config contains wildcard address (e.g. 0.0.0.0 in case
of IPv4) then the first connection request sent by A to B will contain
this address. If the ID of A is smaller than the ID of B, then A will
lose the challenge and the second connection request sent back by B
will never reach A, as B will send the initial message to 0.0.0.0.

So in any 3.5+ ZooKeeper, if wildcard addresses are used in the configs,
then there might be some servers never able to rejoin to the quorum
after they got restarted.

In 3.5+ for backward compatibility reasons (needed during rolling
upgrade) there is a version of the QuorumCnxManager.connectOne()
method that needs no election address but use the last known address
to initiate the connection. In this commit, we simply call this method
if the address is a wildcard address.

I also added a few restart realted tests, to make sure that restart
still works when we don't use wildcard addresses. We can not test
the original error with unit tests, as it would require to start
the quorum on multiple hosts.

I also tested the patch for rolling restart manually both with and
without wildcard addresses in the config.

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

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

Closes #1254 from symat/ZOOKEEPER-2164

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

@@ -32,6 +32,7 @@ import java.net.ServerSocket;
 import java.net.Socket;
 import java.net.SocketException;
 import java.net.SocketTimeoutException;
+import java.net.UnknownHostException;
 import java.nio.BufferUnderflowException;
 import java.nio.ByteBuffer;
 import java.nio.channels.UnresolvedAddressException;
@@ -266,12 +267,33 @@ public class QuorumCnxManager {
                 } catch (ArrayIndexOutOfBoundsException e) {
                     throw new InitialMessageException("No port number in: %s", addr);
                 }
-                addresses.add(new InetSocketAddress(host_port[0], port));
+                if (!isWildcardAddress(host_port[0])) {
+                    addresses.add(new InetSocketAddress(host_port[0], port));
+                }
             }
 
             return new InitialMessage(sid, addresses);
         }
 
+        /**
+         * Returns true if the specified hostname is a wildcard address,
+         * like 0.0.0.0 for IPv4 or :: for IPv6
+         *
+         * (the function is package-private to be visible for testing)
+         */
+        static boolean isWildcardAddress(final String hostname) {
+            try {
+                return InetAddress.getByName(hostname).isAnyLocalAddress();
+            } catch (UnknownHostException e) {
+                // if we can not resolve, it can not be a wildcard address
+                return false;
+            }
+        }
+
+        @Override
+        public String toString() {
+            return "InitialMessage{sid=" + sid + ", electionAddr=" + electionAddr + '}';
+        }
     }
 
     public QuorumCnxManager(QuorumPeer self, final long mySid, Map<Long, QuorumPeer.QuorumServer> view,
@@ -415,6 +437,7 @@ public class QuorumCnxManager {
     private boolean startConnection(Socket sock, Long sid) throws IOException {
         DataOutputStream dout = null;
         DataInputStream din = null;
+        LOG.debug("startConnection (myId:{} --> sid:{})", self.getId(), sid);
         try {
             // Use BufferedOutputStream to reduce the number of IP packets. This is
             // important for x-DC scenarios.
@@ -459,13 +482,11 @@ public class QuorumCnxManager {
 
         // If lost the challenge, then drop the new connection
         if (sid > self.getId()) {
-            LOG.info(
-                "Have smaller server identifier, so dropping the connection: ({}, {})",
-                sid,
-                self.getId());
+            LOG.info("Have smaller server identifier, so dropping the connection: (myId:{} --> sid:{})", self.getId(), sid);
             closeSocket(sock);
             // Otherwise proceed with the connection
         } else {
+            LOG.debug("Have larger server identifier, so keeping the connection: (myId:{} --> sid:{})", self.getId(), sid);
             SendWorker sw = new SendWorker(sock, sid);
             RecvWorker rw = new RecvWorker(sock, din, sid, sw);
             sw.setRecv(rw);
@@ -501,9 +522,11 @@ public class QuorumCnxManager {
         try {
             din = new DataInputStream(new BufferedInputStream(sock.getInputStream()));
 
+            LOG.debug("Sync handling of connection request received from: {}", sock.getRemoteSocketAddress());
             handleConnection(sock, din);
         } catch (IOException e) {
             LOG.error("Exception handling connection, addr: {}, closing server connection", sock.getRemoteSocketAddress());
+            LOG.debug("Exception details: ", e);
             closeSocket(sock);
         }
     }
@@ -514,10 +537,12 @@ public class QuorumCnxManager {
      */
     public void receiveConnectionAsync(final Socket sock) {
         try {
+            LOG.debug("Async handling of connection request received from: {}", sock.getRemoteSocketAddress());
             connectionExecutor.execute(new QuorumConnectionReceiverThread(sock));
             connectionThreadCnt.incrementAndGet();
         } catch (Throwable e) {
             LOG.error("Exception handling connection, addr: {}, closing server connection", sock.getRemoteSocketAddress());
+            LOG.debug("Exception details: ", e);
             closeSocket(sock);
         }
     }
@@ -552,10 +577,13 @@ public class QuorumCnxManager {
                 try {
                     InitialMessage init = InitialMessage.parse(protocolVersion, din);
                     sid = init.sid;
-                    electionAddr = new MultipleAddresses(init.electionAddr,
-                        Duration.ofMillis(self.getMultiAddressReachabilityCheckTimeoutMs()));
+                    if (!init.electionAddr.isEmpty()) {
+                        electionAddr = new MultipleAddresses(init.electionAddr,
+                                Duration.ofMillis(self.getMultiAddressReachabilityCheckTimeoutMs()));
+                    }
+                    LOG.debug("Initial message parsed by {}: {}", self.getId(), init.toString());
                 } catch (InitialMessage.InitialMessageException ex) {
-                    LOG.error(ex.toString());
+                    LOG.error("Initial message parsing error!", ex);
                     closeSocket(sock);
                     return;
                 }
@@ -601,6 +629,10 @@ public class QuorumCnxManager {
                 connectOne(sid);
             }
 
+        } else if (sid == self.getId()) {
+            // we saw this case in ZOOKEEPER-2164
+            LOG.warn("We got a connection request from a server with our own ID. "
+                     + "This should be either a configuration error, or a bug.");
         } else { // Otherwise start worker threads to receive data.
             SendWorker sw = new SendWorker(sock, sid);
             RecvWorker rw = new RecvWorker(sock, din, sid, sw);
@@ -745,6 +777,7 @@ public class QuorumCnxManager {
             Map<Long, QuorumPeer.QuorumServer> lastProposedView = lastSeenQV.getAllMembers();
             if (lastCommittedView.containsKey(sid)) {
                 knownId = true;
+                LOG.debug("Server {} knows {} already, it is in the lastCommittedView", self.getId(), sid);
                 if (connectOne(sid, lastCommittedView.get(sid).electionAddr)) {
                     return;
                 }
@@ -754,6 +787,8 @@ public class QuorumCnxManager {
                 && (!knownId
                     || (lastProposedView.get(sid).electionAddr != lastCommittedView.get(sid).electionAddr))) {
                 knownId = true;
+                LOG.debug("Server {} knows {} already, it is in the lastProposedView", self.getId(), sid);
+
                 if (connectOne(sid, lastProposedView.get(sid).electionAddr)) {
                     return;
                 }
@@ -821,7 +856,7 @@ public class QuorumCnxManager {
      */
     public void softHalt() {
         for (SendWorker sw : senderWorkerMap.values()) {
-            LOG.debug("Halting sender: {}", sw);
+            LOG.debug("Server {} is soft-halting sender towards: {}", self.getId(), sw);
             sw.finish();
         }
     }
@@ -925,6 +960,7 @@ public class QuorumCnxManager {
         @Override
         public void run() {
             if (!shutdown) {
+                LOG.debug("Listener thread started, myId: {}", self.getId());
                 Set<InetSocketAddress> addresses;
 
                 if (self.getQuorumListenOnAllIPs()) {
@@ -977,7 +1013,7 @@ public class QuorumCnxManager {
          * Halts this listener thread.
          */
         void halt() {
-            LOG.debug("Trying to close listeners");
+            LOG.debug("Halt called: Trying to close listeners");
             if (listenerHandlers != null) {
                 LOG.debug("Closing listener: {}", QuorumCnxManager.this.mySid);
                 for (ListenerHandler handler : listenerHandlers) {
@@ -1044,12 +1080,12 @@ public class QuorumCnxManager {
                 while ((!shutdown) && (portBindMaxRetry == 0 || numRetries < portBindMaxRetry)) {
                     try {
                         serverSocket = createNewServerSocket();
-                        LOG.info("My election bind port: {}", address.toString());
+                        LOG.info("{} is accepting connections now, my election bind port: {}", QuorumCnxManager.this.mySid, address.toString());
                         while (!shutdown) {
                             try {
                                 client = serverSocket.accept();
                                 setSockOpts(client);
-                                LOG.info("Received connection request {}", client.getRemoteSocketAddress());
+                                LOG.info("Received connection request from {}", client.getRemoteSocketAddress());
                                 // Receive and handle the connection request
                                 // asynchronously if the quorum sasl authentication is
                                 // enabled. This is required because sasl server
@@ -1173,7 +1209,7 @@ public class QuorumCnxManager {
         }
 
         synchronized boolean finish() {
-            LOG.debug("Calling finish for {}", sid);
+            LOG.debug("Calling SendWorker.finish for {}", sid);
 
             if (!running) {
                 /*
@@ -1240,6 +1276,7 @@ public class QuorumCnxManager {
                 LOG.error("Failed to send last message. Shutting down thread.", e);
                 this.finish();
             }
+            LOG.debug("SendWorker thread started towards {}. myId: {}", sid, QuorumCnxManager.this.mySid);
 
             try {
                 while (running && !shutdown && sock != null) {
@@ -1337,6 +1374,7 @@ public class QuorumCnxManager {
          * @return boolean  Value of variable running
          */
         synchronized boolean finish() {
+            LOG.debug("RecvWorker.finish called. sid: {}. myId: {}", sid, QuorumCnxManager.this.mySid);
             if (!running) {
                 /*
                  * Avoids running finish() twice.
@@ -1354,6 +1392,7 @@ public class QuorumCnxManager {
         public void run() {
             threadCnt.incrementAndGet();
             try {
+                LOG.debug("RecvWorker thread towards {} started. myId: {}", sid, QuorumCnxManager.this.mySid);
                 while (running && !shutdown && sock != null) {
                     /**
                      * Reads the first int to determine the length of the
@@ -1377,7 +1416,7 @@ public class QuorumCnxManager {
                     QuorumCnxManager.this.mySid,
                     e);
             } finally {
-                LOG.warn("Interrupting SendWorker");
+                LOG.warn("Interrupting SendWorker thread from RecvWorker. sid: {}. myId: {}", sid, QuorumCnxManager.this.mySid);
                 sw.finish();
                 closeSocket(sock);
             }

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

@@ -667,6 +667,17 @@ public class CnxManagerTest extends ZKTestCase {
         }
     }
 
+    @Test
+    public void testWildcardAddressRecognition() {
+        assertTrue(QuorumCnxManager.InitialMessage.isWildcardAddress("0.0.0.0"));
+        assertTrue(QuorumCnxManager.InitialMessage.isWildcardAddress("::"));
+        assertFalse(QuorumCnxManager.InitialMessage.isWildcardAddress("some.unresolvable.host.com"));
+        assertFalse(QuorumCnxManager.InitialMessage.isWildcardAddress("127.0.0.1"));
+        assertFalse(QuorumCnxManager.InitialMessage.isWildcardAddress("255.255.255.255"));
+        assertFalse(QuorumCnxManager.InitialMessage.isWildcardAddress("1.2.3.4"));
+        assertFalse(QuorumCnxManager.InitialMessage.isWildcardAddress("www.google.com"));
+    }
+
     private String createLongString(int size) {
         StringBuilder sb = new StringBuilder(size);
         for (int i = 0; i < size; i++) {

zookeeper-server/src/test/java/org/apache/zookeeper/test/QuorumRestartTest.java

@@ -0,0 +1,133 @@
+/*
+ * 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 static org.apache.zookeeper.client.ZKClientConfig.ZOOKEEPER_CLIENT_CNXN_SOCKET;
+import static org.junit.Assert.assertTrue;
+import org.apache.zookeeper.ZKTestCase;
+import org.apache.zookeeper.server.ServerCnxnFactory;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class QuorumRestartTest extends ZKTestCase {
+
+    private static final Logger LOG = LoggerFactory.getLogger(QuorumRestartTest.class);
+    private QuorumUtil qu;
+
+    @Before
+    public void setUp() throws Exception {
+        System.setProperty(ZOOKEEPER_CLIENT_CNXN_SOCKET, "org.apache.zookeeper.ClientCnxnSocketNetty");
+        System.setProperty(ServerCnxnFactory.ZOOKEEPER_SERVER_CNXN_FACTORY, "org.apache.zookeeper.server.NettyServerCnxnFactory");
+
+        // starting a 3 node ensemble without observers
+        qu = new QuorumUtil(1, 2);
+        qu.startAll();
+    }
+
+
+    /**
+     * A basic test for rolling restart. We are restarting the ZooKeeper servers one by one,
+     * starting from the first server. We always make sure that all the nodes joined to the
+     * Quorum before moving forward.
+     * @throws Exception
+     */
+    @Test
+    public void testRollingRestart() throws Exception {
+        for (int serverToRestart = 1; serverToRestart <= 3; serverToRestart++) {
+            LOG.info("***** restarting: " + serverToRestart);
+            qu.shutdown(serverToRestart);
+
+            assertTrue(String.format("Timeout during waiting for server %d to go down", serverToRestart),
+                    ClientBase.waitForServerDown("127.0.0.1:" + qu.getPeer(serverToRestart).clientPort, ClientBase.CONNECTION_TIMEOUT));
+
+            qu.restart(serverToRestart);
+
+            final String errorMessage = "Not all the quorum members are connected after restarting server " + serverToRestart;
+            waitFor(errorMessage, () -> qu.allPeersAreConnected(), 30);
+
+            LOG.info("***** Restart {} succeeded", serverToRestart);
+        }
+    }
+
+    /**
+     * Testing one of the errors reported in ZOOKEEPER-2164, when some servers can not
+     * rejoin to the Quorum after restarting the servers backwards
+     * @throws Exception
+     */
+    @Test
+    public void testRollingRestartBackwards() throws Exception {
+        for (int serverToRestart = 3; serverToRestart >= 1; serverToRestart--) {
+            LOG.info("***** restarting: " + serverToRestart);
+            qu.shutdown(serverToRestart);
+
+            assertTrue(String.format("Timeout during waiting for server %d to go down", serverToRestart),
+                    ClientBase.waitForServerDown("127.0.0.1:" + qu.getPeer(serverToRestart).clientPort, ClientBase.CONNECTION_TIMEOUT));
+
+            qu.restart(serverToRestart);
+
+            final String errorMessage = "Not all the quorum members are connected after restarting server " + serverToRestart;
+            waitFor(errorMessage, () -> qu.allPeersAreConnected(), 30);
+
+            LOG.info("***** Restart {} succeeded", serverToRestart);
+        }
+    }
+
+
+    /**
+     * Testing one of the errors reported in ZOOKEEPER-2164, when some servers can not
+     * rejoin to the Quorum after restarting the current leader multiple times
+     * @throws Exception
+     */
+    @Test
+    public void testRestartingLeaderMultipleTimes() throws Exception {
+        for (int restartCount = 1; restartCount <= 3; restartCount++) {
+            int leaderId = qu.getLeaderServer();
+            LOG.info("***** new leader: " + leaderId);
+            qu.shutdown(leaderId);
+
+            assertTrue("Timeout during waiting for current leader to go down",
+                    ClientBase.waitForServerDown("127.0.0.1:" + qu.getPeer(leaderId).clientPort, ClientBase.CONNECTION_TIMEOUT));
+
+            String errorMessage = "No new leader was elected";
+            waitFor(errorMessage, () -> qu.leaderExists() && qu.getLeaderServer() != leaderId, 30);
+
+            qu.restart(leaderId);
+
+            errorMessage = "Not all the quorum members are connected after restarting the old leader";
+            waitFor(errorMessage, () -> qu.allPeersAreConnected(), 30);
+
+            LOG.info("***** Leader Restart {} succeeded", restartCount);
+        }
+    }
+
+
+    @After
+    public void tearDown() throws Exception {
+        qu.shutdownAll();
+        System.clearProperty(ZOOKEEPER_CLIENT_CNXN_SOCKET);
+        System.clearProperty(ServerCnxnFactory.ZOOKEEPER_SERVER_CNXN_FACTORY);
+    }
+
+
+
+
+}

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

@@ -25,12 +25,14 @@ import java.io.File;
 import java.io.IOException;
 import java.net.InetSocketAddress;
 import java.util.ArrayList;
+import java.util.Arrays;
 import java.util.Collections;
 import java.util.HashMap;
 import java.util.LinkedHashSet;
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
+import java.util.TreeSet;
 import org.apache.zookeeper.PortAssignment;
 import org.apache.zookeeper.server.quorum.Election;
 import org.apache.zookeeper.server.quorum.QuorumPeer;
@@ -51,6 +53,8 @@ public class QuorumUtil {
     // TODO refactor QuorumBase to be special case of this
 
     private static final Logger LOG = LoggerFactory.getLogger(QuorumUtil.class);
+    private static final Set<QuorumPeer.ServerState> CONNECTED_STATES = new TreeSet<>(
+        Arrays.asList(QuorumPeer.ServerState.LEADING, QuorumPeer.ServerState.FOLLOWING, QuorumPeer.ServerState.OBSERVING));
 
     public static class PeerStruct {
 
@@ -274,6 +278,12 @@ public class QuorumUtil {
         return "127.0.0.1:" + peer.getClientPort();
     }
 
+    public boolean allPeersAreConnected() {
+        return peers.values().stream()
+          .map(ps -> ps.peer)
+          .allMatch(peer -> CONNECTED_STATES.contains(peer.getPeerState()));
+    }
+
     public QuorumPeer getLeaderQuorumPeer() {
         for (PeerStruct ps : peers.values()) {
             if (ps.peer.leader != null) {
@@ -320,6 +330,15 @@ public class QuorumUtil {
         return index;
     }
 
+    public boolean leaderExists() {
+        for (int i = 1; i <= ALL; i++) {
+            if (getPeer(i).peer.leader != null) {
+                return true;
+            }
+        }
+        return false;
+    }
+
     public String getConnectionStringForServer(final int index) {
         return "127.0.0.1:" + getPeer(index).clientPort;
     }