Wait for an in-flight replay before closing the log watcher (#1347)

## Motivation

When a replica shuts down, `PathChildrenCache.close()` cancels its
in-flight tasks with an interrupt.
If a replay was waiting on `delegate.execute(...)` at that moment, the
command had already been applied
to the local data but `lastReplayedRevision` was never advanced, because
it is updated only on the
success path. The replica is then left with local data ahead of
`<dataDir>/last_revision`, and the next
start-up replays the same revision again. Re-applying a command whose
effect is already in the local
data fails, so the replica enters read-only mode and needs a manual
re-sync.

```
[INFO ] [command-executor-shutdown] Closing the log watcher
[ERROR] [zookeeper-log-watcher-1-1] Failed to replay a log at revision N; entering read-only mode.
java.lang.InterruptedException
    at java.base/java.util.concurrent.CompletableFuture.reportGet(CompletableFuture.java:386)
    at java.base/java.util.concurrent.CompletableFuture.get(CompletableFuture.java:2073)
    at ...ZooKeeperCommandExecutor.replayLogs(ZooKeeperCommandExecutor.java:820)
    at ...ZooKeeperCommandExecutor.childEvent(ZooKeeperCommandExecutor.java:885)
    at ...PathChildrenCache.lambda$callListeners$1(PathChildrenCache.java:529)
```

## Modifications

- Wait for an in-flight replay before closing the log watcher, by
acquiring the same monitor that
`replayLogs()` holds. Releasing it is safe because `listenerInfo` is
already null, so a replay that
  starts afterwards returns before executing anything.
- Move the log watcher shutdown ahead of `delegate.stop()`. The barrier
waits for a replay that is
itself waiting on the delegate, so the delegate must still be running.
This also matches the drain
that `shutdown(executor)` already performs for the command executor
threads.
`logWatcher.close()` still runs before `shutdown(logWatcherExecutor)`:
reversing the two would let
`PathChildrenCache` keep submitting to an already shut-down executor,
because `submitToExecutor()`
  only guards on its own state, which flips in `close()`.
- Wait uninterruptibly for the replay result, as a safeguard on the line
where the failure occurred.
- Log the last replayed revision once the replay is drained.

Note this covers a graceful shutdown only. A `SIGKILL` between the local
apply and the progress update
leaves the same divergence; making the replay idempotent is a separate
topic.

## Result

- A replica that shuts down while replaying no longer leaves its local
data ahead of its recorded
replication progress, so it no longer enters read-only mode on the next
start-up.

Author: ikhoon

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

@@ -680,14 +680,21 @@ private void stopLater() {
     protected void doStop(@Nullable Runnable onReleaseLeadership,
                           @Nullable Runnable onReleaseZoneLeadership) throws Exception {
         canReplicate = false;
+        // Stop accepting new replay logs.
         listenerInfo = null;
+
         logger.info("Stopping the worker threads");
         boolean interrupted = shutdown(executor);
         logger.info("Stopped the worker threads");
 
         try {
-            logger.info("Stopping the delegate command executor");
-            delegate.stop();
+            // 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();
+            }
             logger.info("Stopped the delegate command executor");
         } catch (Exception e) {
             logger.warn("Failed to stop the delegate command executor {}: {}", delegate, e.getMessage(), e);
@@ -817,7 +824,9 @@ private synchronized void replayLogs(long targetRevision, boolean force) {
                 l = loadLog(nextRevision);
                 final Command<?> command = l.command();
                 final Object expectedResult = l.result();
-                final Object actualResult = delegate.execute(REPLAY_CONTEXT, command).get();
+                // An interrupt here would split the local apply from updateLastReplayedRevision() below.
+                final Object actualResult =
+                        Uninterruptibles.getUninterruptibly(delegate.execute(REPLAY_CONTEXT, command));
 
                 if (!Objects.equals(expectedResult, actualResult)) {
                     throw new ReplicationException(

server/src/test/java/com/linecorp/centraldogma/server/internal/replication/ZooKeeperCommandExecutorTest.java

@@ -748,6 +748,68 @@ void inFlightCommandOnStopIsRecordedAndNotReplayed() throws Exception {
         }
     }
 
+    /**
+     * A replay still in flight when the executor stops must finish and record its progress durably,
+     * because closing the log watcher cancels its tasks with an interrupt. Otherwise the command is
+     * applied to the local data while last_revision is not advanced, and the next start-up applies it
+     * a second time.
+     */
+    @Test
+    @Timeout(120)
+    void inFlightReplayOnStopIsRecordedAndNotReplayed() throws Exception {
+        final CountDownLatch replayEntered = new CountDownLatch(1);
+        final CountDownLatch proceed = new CountDownLatch(1);
+        final AtomicInteger replayCount = new AtomicInteger();
+        final AtomicInteger replicaIndex = new AtomicInteger();
+        final Supplier<Function<Command<?>, CompletableFuture<?>>> delegateSupplier = () -> {
+            final boolean replaying = replicaIndex.getAndIncrement() == 1;
+            final Function<Command<?>, CompletableFuture<?>> base = newMockDelegate();
+            return command -> {
+                if (replaying && command != null && command.type() == CommandType.CREATE_REPOSITORY) {
+                    // Park inside replayLogs() so we can stop while the replay is in flight.
+                    replayCount.incrementAndGet();
+                    replayEntered.countDown();
+                    return CompletableFuture.supplyAsync(() -> {
+                        try {
+                            proceed.await();
+                        } catch (InterruptedException e) {
+                            throw new RuntimeException(e);
+                        }
+                        return null;
+                    }, CommonPools.blockingTaskExecutor());
+                }
+                return base.apply(command);
+            };
+        };
+
+        try (Cluster cluster = Cluster.builder().numReplicas(3).build(delegateSupplier)) {
+            final Replica origin = cluster.get(0);
+            final Replica replaying = cluster.get(1);
+            origin.commandExecutor().execute(Command.createProject(Author.SYSTEM, "p")).join();
+            await().untilAsserted(() -> assertThat(replaying.localRevision()).isEqualTo(0L));
+
+            // Park the replay of revision 1, then stop while it is in flight.
+            origin.commandExecutor().execute(Command.createRepository(Author.SYSTEM, "p", "r")).join();
+            assertThat(replayEntered.await(10, TimeUnit.SECONDS)).isTrue();
+            final CompletableFuture<Void> stopFuture = replaying.commandExecutor().stop();
+            // The shutdown must not finish while the replay is parked; that is the barrier doing its job.
+            assertThatThrownBy(() -> stopFuture.get(3, TimeUnit.SECONDS))
+                    .isInstanceOf(TimeoutException.class);
+            proceed.countDown();
+            stopFuture.join();
+
+            // The replay finished before the log watcher was closed, so its progress is durable.
+            assertThat(replaying.localRevision()).isEqualTo(1L);
+
+            // Catch up on a later revision to prove revision 1 was not applied a second time.
+            replaying.commandExecutor().start().join();
+            origin.commandExecutor().execute(Command.createProject(Author.SYSTEM, "p2")).join();
+            await().untilAsserted(() -> assertThat(replaying.localRevision()).isEqualTo(2L));
+            assertThat(replayCount).hasValue(1);
+            assertThat(replaying.commandExecutor().isWritable()).isTrue();
+        }
+    }
+
     private static <T> void awaitUntilReplicated(Cluster cluster, Command<T> command) {
         for (int i = 0; i < cluster.size(); i++) {
             final Replica replica = cluster.get(i);