ZOOKEEPER-4541 Ephemeral znode owned by closed session visible in 1 of 3 servers

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

@@ -883,7 +883,7 @@ public synchronized void shutdown(boolean fullyShutDown) {
                 //  * If we fetch a new snapshot from leader, the zkDb will be
                 //    cleared anyway before loading the snapshot
                 try {
-                    //This will fast forward the database to the latest recorded transactions
+                    // This will fast-forward the database to the latest recorded transactions
                     zkDb.fastForwardDataBase();
                 } catch (IOException e) {
                     LOG.error("Error updating DB", e);

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

@@ -155,11 +155,11 @@ protected void unregisterMetrics() {
     }
 
     @Override
-    public synchronized void shutdown() {
+    public synchronized void shutdown(boolean fullyShutDown) {
         if (containerManager != null) {
             containerManager.stop();
         }
-        super.shutdown();
+        super.shutdown(fullyShutDown);
     }
 
     @Override

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

@@ -152,23 +152,31 @@ protected void unregisterJMX(Learner peer) {
     }
 
     @Override
-    public synchronized void shutdown() {
+    public synchronized void shutdown(boolean fullyShutDown) {
         if (!canShutdown()) {
             LOG.debug("ZooKeeper server is not running, so not proceeding to shutdown!");
-            return;
         }
-        LOG.info("Shutting down");
-        try {
-            super.shutdown();
-        } catch (Exception e) {
-            LOG.warn("Ignoring unexpected exception during shutdown", e);
+        else {
+            LOG.info("Shutting down");
+            try {
+                if (syncProcessor != null) {
+                    // Shutting down the syncProcessor here, first, ensures queued transactions here are written to
+                    // permanent storage, which ensures that crash recovery data is consistent with what is used for a
+                    // leader election immediately following shutdown, because of the old leader going down; and also
+                    // that any state on its way to being written is also loaded in the potential call to
+                    // fast-forward-from-edits, in super.shutdown(...), so we avoid getting a DIFF from the new leader
+                    // that contains entries we have already written to our transaction log.
+                    syncProcessor.shutdown();
+                }
+            }
+            catch (Exception e) {
+                LOG.warn("Ignoring unexpected exception in syncprocessor shutdown", e);
+            }
         }
         try {
-            if (syncProcessor != null) {
-                syncProcessor.shutdown();
-            }
+            super.shutdown(fullyShutDown);
         } catch (Exception e) {
-            LOG.warn("Ignoring unexpected exception in syncprocessor shutdown", e);
+            LOG.warn("Ignoring unexpected exception during shutdown", e);
         }
     }
 

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

@@ -73,7 +73,7 @@ public Learner getLearner() {
      * @param request
      */
     public void commitRequest(Request request) {
-        if (syncRequestProcessorEnabled) {
+        if (syncProcessor != null) {
             // Write to txnlog and take periodic snapshot
             syncProcessor.processRequest(request);
         }
@@ -107,6 +107,9 @@ protected void setupRequestProcessors() {
             syncProcessor = new SyncRequestProcessor(this, null);
             syncProcessor.start();
         }
+        else {
+            syncProcessor = null;
+        }
     }
 
     /*
@@ -127,18 +130,6 @@ public String getState() {
         return "observer";
     }
 
-    @Override
-    public synchronized void shutdown() {
-        if (!canShutdown()) {
-            LOG.debug("ZooKeeper server is not running, so not proceeding to shutdown!");
-            return;
-        }
-        super.shutdown();
-        if (syncRequestProcessorEnabled && syncProcessor != null) {
-            syncProcessor.shutdown();
-        }
-    }
-
     @Override
     public void dumpMonitorValues(BiConsumer<String, Object> response) {
         super.dumpMonitorValues(response);

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

@@ -190,23 +190,24 @@ public long getServerId() {
     }
 
     @Override
-    public synchronized void shutdown() {
+    public synchronized void shutdown(boolean fullyShutDown) {
         if (!canShutdown()) {
+            super.shutdown(fullyShutDown);
             LOG.debug("ZooKeeper server is not running, so not proceeding to shutdown!");
-            return;
         }
-        shutdown = true;
-        unregisterJMX(this);
-
-        // set peer's server to null
-        self.setZooKeeperServer(null);
-        // clear all the connections
-        self.closeAllConnections();
+        else {
+            shutdown = true;
+            unregisterJMX(this);
 
-        self.adminServer.setZooKeeperServer(null);
+            // set peer's server to null
+            self.setZooKeeperServer(null);
+            // clear all the connections
+            self.closeAllConnections();
 
+            self.adminServer.setZooKeeperServer(null);
+        }
         // shutdown the server itself
-        super.shutdown();
+        super.shutdown(fullyShutDown);
     }
 
     @Override

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

@@ -20,6 +20,8 @@
 
 import java.io.Flushable;
 import java.io.IOException;
+import java.net.Socket;
+
 import org.apache.zookeeper.ZooDefs.OpCode;
 import org.apache.zookeeper.server.Request;
 import org.apache.zookeeper.server.RequestProcessor;
@@ -64,7 +66,8 @@ public void flush() throws IOException {
         } catch (IOException e) {
             LOG.warn("Closing connection to leader, exception during packet send", e);
             try {
-                if (!learner.sock.isClosed()) {
+                Socket socket = learner.sock;
+                if ( socket != null && ! learner.sock.isClosed()) {
                     learner.sock.close();
                 }
             } catch (IOException e1) {