Wait for the in-flight replay at delegate.stop() instead of reordering doStop

Take the barrier onto delegate.stop() as suggested in review, and drop the
doStop reordering it replaces. retryPolicy = RETRY_POLICY_NEVER therefore stays
where it was, ahead of the leader selectors and the log watcher.

Log each step around the monitor so a shutdown says what it is waiting for and
where the replay progress stands.

Author: ikhoon

server/src/main/java/com/linecorp/centraldogma/server/internal/replication/ZooKeeperCommandExecutor.java

@@ -688,24 +688,13 @@ protected void doStop(@Nullable Runnable onReleaseLeadership,
         logger.info("Stopped the worker threads");
 
         try {
-            if (logWatcher != null) {
-                logger.info("Closing the log watcher");
-                // Wait for an in-flight replay: close() cancels its tasks with an interrupt.
-                // Releasing is safe because listenerInfo is null, so a new replay returns immediately.
-                synchronized (this) {
-                    logger.info("Drained the log watcher; last replayed revision: {}", lastReplayedRevision);
-                }
-                logWatcher.close();
-                interrupted |= shutdown(logWatcherExecutor);
-                logger.info("Closed the log watcher");
+            // A replay holds this monitor until it records its progress; close() would interrupt it.
+            logger.info("Waiting for an in-flight replay to finish");
+            synchronized (this) {
+                logger.info("No replay in flight; last replayed revision: {}", lastReplayedRevision);
+                logger.info("Stopping the delegate command executor");
+                delegate.stop();
             }
-        } catch (Exception e) {
-            logger.warn("Failed to close the log watcher: {}", e.getMessage(), e);
-        }
-
-        try {
-            logger.info("Stopping the delegate command executor");
-            delegate.stop();
             logger.info("Stopped the delegate command executor");
         } catch (Exception e) {
             logger.warn("Failed to stop the delegate command executor {}: {}", delegate, e.getMessage(), e);
@@ -732,26 +721,37 @@ protected void doStop(@Nullable Runnable onReleaseLeadership,
                     logger.warn("Failed to close the zone {} leader selector: {}", zone, e.getMessage(), e);
                 } finally {
                     try {
-                        if (curator != null) {
-                            logger.info("Closing the Curator framework");
-                            curator.close();
-                            logger.info("Closed the Curator framework");
+                        if (logWatcher != null) {
+                            logger.info("Closing the log watcher");
+                            logWatcher.close();
+                            interrupted |= shutdown(logWatcherExecutor);
+                            logger.info("Closed the log watcher");
                         }
                     } catch (Exception e) {
-                        logger.warn("Failed to close the Curator framework: {}", e.getMessage(), e);
+                        logger.warn("Failed to close the log watcher: {}", e.getMessage(), e);
                     } finally {
                         try {
-                            if (quorumPeer != null) {
-                                final long peerId = quorumPeer.getId();
-                                logger.info("Shutting down the ZooKeeper peer ({})", peerId);
-                                quorumPeer.shutdown();
-                                logger.info("Shut down the ZooKeeper peer ({})", peerId);
+                            if (curator != null) {
+                                logger.info("Closing the Curator framework");
+                                curator.close();
+                                logger.info("Closed the Curator framework");
                             }
                         } catch (Exception e) {
-                            logger.warn("Failed to shut down the ZooKeeper peer: {}", e.getMessage(), e);
+                            logger.warn("Failed to close the Curator framework: {}", e.getMessage(), e);
                         } finally {
-                            if (interrupted) {
-                                Thread.currentThread().interrupt();
+                            try {
+                                if (quorumPeer != null) {
+                                    final long peerId = quorumPeer.getId();
+                                    logger.info("Shutting down the ZooKeeper peer ({})", peerId);
+                                    quorumPeer.shutdown();
+                                    logger.info("Shut down the ZooKeeper peer ({})", peerId);
+                                }
+                            } catch (Exception e) {
+                                logger.warn("Failed to shut down the ZooKeeper peer: {}", e.getMessage(), e);
+                            } finally {
+                                if (interrupted) {
+                                    Thread.currentThread().interrupt();
+                                }
                             }
                         }
                     }