Zookeeper-3188: Improve resilience to network

zookeeper-server/src/main/java/org/apache/zookeeper/server/ObserverBean.java

@@ -21,6 +21,7 @@
 import org.apache.zookeeper.server.quorum.Observer;
 import org.apache.zookeeper.server.quorum.ObserverMXBean;
 import org.apache.zookeeper.server.quorum.QuorumPeer;
+import java.net.InetSocketAddress;
 
 /**
  * ObserverBean
@@ -52,7 +53,8 @@ public String getLearnerMaster() {
         if (learnerMaster == null || learnerMaster.addr == null) {
             return "Unknown";
         }
-        return learnerMaster.addr.getAddress().getHostAddress() + ":" + learnerMaster.addr.getPort();
+        InetSocketAddress address = learnerMaster.addr.getReachableOrOne();
+        return address.getAddress().getHostAddress() + ":" + address.getPort();
     }
 
     public void setLearnerMaster(String learnerMaster) {

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

@@ -1186,7 +1186,7 @@ public void processPacket(ServerCnxn cnxn, ByteBuffer incomingBuffer) throws IOE
         // there might be a race condition that it enabled recv after
         // processing request and then disabled when check throttling.
         //
-        // Be aware that we're actually checking the global outstanding 
+        // Be aware that we're actually checking the global outstanding
         // request before this request.
         //
         // It's fine if the IOException thrown before we decrease the count

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

@@ -22,6 +22,7 @@
 import java.io.IOException;
 import java.net.DatagramPacket;
 import java.net.DatagramSocket;
+import java.net.InetAddress;
 import java.net.InetSocketAddress;
 import java.net.SocketException;
 import java.nio.ByteBuffer;
@@ -44,8 +45,6 @@
 
 import org.apache.zookeeper.jmx.MBeanRegistry;
 import org.apache.zookeeper.server.ZooKeeperThread;
-import org.apache.zookeeper.server.quorum.Election;
-import org.apache.zookeeper.server.quorum.Vote;
 import org.apache.zookeeper.server.quorum.QuorumPeer.QuorumServer;
 import org.apache.zookeeper.server.quorum.QuorumPeer.ServerState;
 
@@ -732,8 +731,8 @@ private void process(ToSend m) {
             }
 
             for (QuorumServer server : self.getVotingView().values()) {
-                InetSocketAddress saddr = new InetSocketAddress(server.addr
-                        .getAddress(), port);
+                InetAddress address = server.addr.getReachableOrOne().getAddress();
+                InetSocketAddress saddr = new InetSocketAddress(address, port);
                 addrChallengeMap.put(saddr, new ConcurrentHashMap<Long, Long>());
             }
 
@@ -763,7 +762,7 @@ public AuthFastLeaderElection(QuorumPeer self) {
 
     private void starter(QuorumPeer self) {
         this.self = self;
-        port = self.getVotingView().get(self.getId()).electionAddr.getPort();
+        port = self.getVotingView().get(self.getId()).electionAddr.getAllPorts().get(0);
         proposedLeader = -1;
         proposedZxid = -1;
 
@@ -786,11 +785,10 @@ private void leaveInstance() {
 
     private void sendNotifications() {
         for (QuorumServer server : self.getView().values()) {
-
+            InetSocketAddress address = self.getView().get(server.id).electionAddr.getReachableOrOne();
             ToSend notmsg = new ToSend(ToSend.mType.notification,
                     AuthFastLeaderElection.sequencer++, proposedLeader,
-                    proposedZxid, logicalclock.get(), QuorumPeer.ServerState.LOOKING,
-                    self.getView().get(server.id).electionAddr);
+                    proposedZxid, logicalclock.get(), QuorumPeer.ServerState.LOOKING, address);
 
             sendqueue.offer(notmsg);
         }

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

@@ -19,7 +19,6 @@
 package org.apache.zookeeper.server.quorum;
 
 import java.io.IOException;
-import java.net.InetSocketAddress;
 import java.nio.ByteBuffer;
 
 import org.apache.jute.Record;
@@ -76,13 +75,14 @@ void followLeader() throws InterruptedException {
         self.start_fle = 0;
         self.end_fle = 0;
         fzk.registerJMX(new FollowerBean(this, zk), self.jmxLocalPeerBean);
+
         try {
             QuorumServer leaderServer = findLeader();
             try {
                 connectToLeader(leaderServer.addr, leaderServer.hostname);
                 long newEpochZxid = registerWithLeader(Leader.FOLLOWERINFO);
                 if (self.isReconfigStateChange())
-                   throw new Exception("learned about role change");
+                    throw new Exception("learned about role change");
                 //check to see if the leader zxid is lower than ours
                 //this should never happen but is just a safety check
                 long newEpoch = ZxidUtils.getEpochFromZxid(newEpochZxid);
@@ -119,15 +119,14 @@ void followLeader() throws InterruptedException {
                 } catch (IOException e1) {
                     e1.printStackTrace();
                 }
-
                 // clear pending revalidations
                 pendingRevalidations.clear();
             }
         } finally {
             if (om != null) {
                 om.stop();
             }
-            zk.unregisterJMX((Learner)this);
+            zk.unregisterJMX((Learner) this);
         }
     }