Move ownership of QuorumPeer to Recoonfigurer

3 files changed, 49 insertions, 12 deletions

diff --git a/zookeeper-server/zookeeper-server-common/src/main/java/com/yahoo/vespa/zookeeper/QuorumPeer.java b/zookeeper-server/zookeeper-server-common/src/main/java/com/yahoo/vespa/zookeeper/QuorumPeer.java
new file mode 100644
index 00000000000..010b6546443
--- /dev/null
+++ b/zookeeper-server/zookeeper-server-common/src/main/java/com/yahoo/vespa/zookeeper/QuorumPeer.java
@@ -0,0 +1,20 @@
+// Copyright Verizon Media. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root.
+package com.yahoo.vespa.zookeeper;
+
+import java.nio.file.Path;
+import java.time.Duration;
+
+/**
+ * Wraps ZK-version-dependent VespaQuorumPeer
+ *
+ * @author jonmv
+ */
+public interface QuorumPeer {
+
+    /** Starts ZK with config from the given path. */
+    void start(Path path);
+
+    /** Shuts down this peer, with the given timeout, or kills the process. */
+    void shutdown(Duration timeout);
+
+}
diff --git a/zookeeper-server/zookeeper-server-common/src/main/java/com/yahoo/vespa/zookeeper/Reconfigurer.java b/zookeeper-server/zookeeper-server-common/src/main/java/com/yahoo/vespa/zookeeper/Reconfigurer.java
index 5296e3646c8..045538b2d7e 100644
--- a/zookeeper-server/zookeeper-server-common/src/main/java/com/yahoo/vespa/zookeeper/Reconfigurer.java
+++ b/zookeeper-server/zookeeper-server-common/src/main/java/com/yahoo/vespa/zookeeper/Reconfigurer.java
@@ -13,6 +13,8 @@ import java.time.Instant;
 import java.util.ArrayList;
 import java.util.List;
 import java.util.Objects;
+import java.util.function.Consumer;
+import java.util.function.Supplier;
 import java.util.logging.Level;
 import java.util.logging.Logger;
 import java.util.stream.Collectors;
@@ -34,6 +36,7 @@ public class Reconfigurer extends AbstractComponent {
     private final VespaZooKeeperAdmin vespaZooKeeperAdmin;
     private final Sleeper sleeper;
 
+    private QuorumPeer peer;
     private ZooKeeperRunner zooKeeperRunner;
     private ZookeeperServerConfig activeConfig;
 
@@ -48,12 +51,16 @@ public class Reconfigurer extends AbstractComponent {
         log.log(Level.FINE, "Created ZooKeeperReconfigurer");
     }
 
-    void startOrReconfigure(ZookeeperServerConfig newConfig, VespaZooKeeperServer server) {
-        if (zooKeeperRunner == null)
+    void startOrReconfigure(ZookeeperServerConfig newConfig, VespaZooKeeperServer server,
+                            Supplier<QuorumPeer> quorumPeerGetter, Consumer<QuorumPeer> quorumPeerSetter) {
+        if (zooKeeperRunner == null) {
+            peer = quorumPeerGetter.get(); // Obtain the peer from the server. This will be shared with laters servers.
             zooKeeperRunner = startServer(newConfig, server);
+        }
+        quorumPeerSetter.accept(peer);
 
         if (shouldReconfigure(newConfig)) {
-            reconfigure(newConfig, server);
+            reconfigure(newConfig);
         }
     }
 
@@ -79,11 +86,11 @@ public class Reconfigurer extends AbstractComponent {
         return runner;
     }
 
-    private void reconfigure(ZookeeperServerConfig newConfig, VespaZooKeeperServer server) {
+    private void reconfigure(ZookeeperServerConfig newConfig) {
         Instant reconfigTriggered = Instant.now();
         // No point in trying to reconfigure if there is only one server in the new ensemble,
         // the others will be shutdown or are about to be shutdown
-        if (newConfig.server().size() == 1) shutdownAndDie(server, Duration.ZERO);
+        if (newConfig.server().size() == 1) shutdownAndDie(Duration.ZERO);
 
         List<String> newServers = difference(servers(newConfig), servers(activeConfig));
         String leavingServerIds = String.join(",", serverIdsDifference(activeConfig, newConfig));
@@ -122,12 +129,12 @@ public class Reconfigurer extends AbstractComponent {
         }
 
         // Reconfiguration failed
-        shutdownAndDie(server, reconfigTimeout);
+        shutdownAndDie(reconfigTimeout);
     }
 
-    private void shutdownAndDie(VespaZooKeeperServer server, Duration reconfigTimeout) {
-        server.shutdown();
-        Process.logAndDie("Reconfiguration did not complete within timeout " + reconfigTimeout + ". Forcing shutdown");
+    private void shutdownAndDie(Duration reconfigTimeout) {
+        shutdown();
+        Process.logAndDie("Reconfiguration did not complete within timeout " + reconfigTimeout + ". Forcing container shutdown.");
     }
 
     /** Returns the timeout to use for the given joining server count */
diff --git a/zookeeper-server/zookeeper-server-common/src/test/java/com/yahoo/vespa/zookeeper/ReconfigurerTest.java b/zookeeper-server/zookeeper-server-common/src/test/java/com/yahoo/vespa/zookeeper/ReconfigurerTest.java
index 5cee0de2b6e..a2a3fc0a39f 100644
--- a/zookeeper-server/zookeeper-server-common/src/test/java/com/yahoo/vespa/zookeeper/ReconfigurerTest.java
+++ b/zookeeper-server/zookeeper-server-common/src/test/java/com/yahoo/vespa/zookeeper/ReconfigurerTest.java
@@ -14,10 +14,13 @@ import java.io.IOException;
 import java.nio.file.Path;
 import java.time.Duration;
 import java.util.Arrays;
+import java.util.concurrent.atomic.AtomicBoolean;
 
 import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
 import static org.junit.Assert.assertNull;
 import static org.junit.Assert.assertSame;
+import static org.junit.Assert.assertTrue;
 
 /**
  * Tests dynamic reconfiguration of zookeeper cluster.
@@ -136,9 +139,16 @@ public class ReconfigurerTest {
         return builder;
     }
 
+    private static class MockQuorumPeer implements QuorumPeer {
+        final AtomicBoolean on = new AtomicBoolean();
+        @Override public void start(Path path) { assertFalse(on.getAndSet(true)); }
+        @Override public void shutdown(Duration timeout) { assertTrue(on.getAndSet(false)); }
+    }
+
     private static class TestableReconfigurer extends Reconfigurer implements VespaZooKeeperServer {
 
         private final TestableVespaZooKeeperAdmin zooKeeperAdmin;
+        private QuorumPeer serverPeer;
 
         TestableReconfigurer(TestableVespaZooKeeperAdmin zooKeeperAdmin) {
             super(zooKeeperAdmin, new Sleeper() {
@@ -152,7 +162,7 @@ public class ReconfigurerTest {
         }
 
         void startOrReconfigure(ZookeeperServerConfig newConfig) {
-            startOrReconfigure(newConfig, this);
+            startOrReconfigure(newConfig, this, MockQuorumPeer::new, peer -> serverPeer = peer);
         }
 
         String connectionSpec() {
@@ -172,10 +182,10 @@ public class ReconfigurerTest {
         }
 
         @Override
-        public void shutdown() {}
+        public void shutdown() { serverPeer.shutdown(Duration.ofSeconds(1)); }
 
         @Override
-        public void start(Path configFilePath) { }
+        public void start(Path configFilePath) { serverPeer.start(configFilePath); }
 
         @Override
         public boolean reconfigurable() {