ZOOKEEPER-4262: Backport ZOOKEEPER-3911 to branch-3.5.9

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

@@ -467,6 +467,19 @@ public void startdata()
     }
 
     public synchronized void startup() {
+        startupWithServerState(State.RUNNING);
+    }
+
+    public synchronized void startupWithoutServing() {
+        startupWithServerState(State.INITIAL);
+    }
+
+    public synchronized void startServing() {
+        setState(State.RUNNING);
+        notifyAll();
+    }
+
+    private void startupWithServerState(State state) {
         if (sessionTracker == null) {
             createSessionTracker();
         }
@@ -475,7 +488,8 @@ public synchronized void startup() {
 
         registerJMX();
 
-        setState(State.RUNNING);
+        setState(state);
+
         notifyAll();
     }
 

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

@@ -94,11 +94,7 @@ void followLeader() throws InterruptedException {
                 }
             } catch (Exception e) {
                 LOG.warn("Exception when following the leader", e);
-                try {
-                    sock.close();
-                } catch (IOException e1) {
-                    e1.printStackTrace();
-                }
+                closeSocket();
 
                 // clear pending revalidations
                 pendingRevalidations.clear();

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

@@ -559,17 +559,29 @@ else if (qp.getType() == Leader.SNAP) {
                    }
 
                     self.setCurrentEpoch(newEpoch);
-                    writeToTxnLog = true; //Anything after this needs to go to the transaction log, not applied directly in memory
+                    writeToTxnLog = true;
+                    //Anything after this needs to go to the transaction log, not applied directly in memory
                     isPreZAB1_0 = false;
+
+                    // ZOOKEEPER-3911: make sure sync the uncommitted logs before commit them (ACK NEWLEADER).
+                    sock.setSoTimeout(self.tickTime * self.syncLimit);
+                    zk.startupWithoutServing();
+                    if (zk instanceof FollowerZooKeeperServer) {
+                        FollowerZooKeeperServer fzk = (FollowerZooKeeperServer) zk;
+                        for (PacketInFlight p : packetsNotCommitted) {
+                            fzk.logRequest(p.hdr, p.rec);
+                        }
+                        packetsNotCommitted.clear();
+                    }
+
                     writePacket(new QuorumPacket(Leader.ACK, newLeaderZxid, null, null), true);
                     break;
                 }
             }
         }
         ack.setZxid(ZxidUtils.makeZxid(newEpoch, 0));
         writePacket(ack, true);
-        sock.setSoTimeout(self.tickTime * self.syncLimit);
-        zk.startup();
+        zk.startServing();
         /*
          * Update the election vote here to ensure that all members of the
          * ensemble report the same vote to new servers that start up and
@@ -658,6 +670,7 @@ public void shutdown() {
         self.setZooKeeperServer(null);
         self.closeAllConnections();
         self.adminServer.setZooKeeperServer(null);
+        closeSocket();
         // shutdown previous zookeeper
         if (zk != null) {
             zk.shutdown();
@@ -667,4 +680,14 @@ public void shutdown() {
     boolean isRunning() {
         return self.isRunning() && zk.isRunning();
     }
+
+    void closeSocket() {
+        try {
+            if (sock != null) {
+                sock.close();
+            }
+        } catch (IOException e) {
+            LOG.warn("Ignoring error closing connection to leader", e);
+        }
+    }
 }

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

@@ -79,11 +79,7 @@ void observeLeader() throws Exception {
                 }
             } catch (Exception e) {
                 LOG.warn("Exception when observing the leader", e);
-                try {
-                    sock.close();
-                } catch (IOException e1) {
-                    e1.printStackTrace();
-                }
+                closeSocket();
 
                 // clear pending revalidations
                 pendingRevalidations.clear();