Split out a ZKAdmin interface and implement it

6 files changed, 157 insertions, 70 deletions

diff --git a/zookeeper-server/zookeeper-server-3.5.6/src/main/java/com/yahoo/vespa/zookeeper/LoggingWatcher.java b/zookeeper-server/zookeeper-server-3.5.6/src/main/java/com/yahoo/vespa/zookeeper/LoggingWatcher.java
new file mode 100644
index 00000000000..dfd99d38505
--- /dev/null
+++ b/zookeeper-server/zookeeper-server-3.5.6/src/main/java/com/yahoo/vespa/zookeeper/LoggingWatcher.java
@@ -0,0 +1,19 @@
+// 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 org.apache.zookeeper.WatchedEvent;
+import org.apache.zookeeper.Watcher;
+
+import java.util.logging.Level;
+import java.util.logging.Logger;
+
+class LoggingWatcher implements Watcher {
+
+    private static final Logger log = java.util.logging.Logger.getLogger(LoggingWatcher.class.getName());
+
+    @Override
+    public void process(WatchedEvent event) {
+        log.log(Level.INFO, event.toString());
+    }
+
+}
diff --git a/zookeeper-server/zookeeper-server-3.5.6/src/main/java/com/yahoo/vespa/zookeeper/Reconfigurer.java b/zookeeper-server/zookeeper-server-3.5.6/src/main/java/com/yahoo/vespa/zookeeper/Reconfigurer.java
index be0ec535740..89c29a642d4 100644
--- a/zookeeper-server/zookeeper-server-3.5.6/src/main/java/com/yahoo/vespa/zookeeper/Reconfigurer.java
+++ b/zookeeper-server/zookeeper-server-3.5.6/src/main/java/com/yahoo/vespa/zookeeper/Reconfigurer.java
@@ -5,13 +5,7 @@ import com.google.inject.Inject;
 import com.yahoo.cloud.config.ZookeeperServerConfig;
 import com.yahoo.component.AbstractComponent;
 import com.yahoo.yolean.Exceptions;
-import org.apache.zookeeper.KeeperException;
-import org.apache.zookeeper.WatchedEvent;
-import org.apache.zookeeper.Watcher;
-import org.apache.zookeeper.admin.ZooKeeperAdmin;
 
-import java.io.IOException;
-import java.nio.charset.StandardCharsets;
 import java.time.Duration;
 import java.time.Instant;
 import java.util.ArrayList;
@@ -31,9 +25,6 @@ public class Reconfigurer extends AbstractComponent {
 
     private static final Logger log = java.util.logging.Logger.getLogger(Reconfigurer.class.getName());
 
-    // Timeout for connecting to ZooKeeper to reconfigure
-    private static final Duration sessionTimeout = Duration.ofSeconds(30);
-
     // How long to wait before triggering reconfig. This is multiplied by the node ID
     private static final Duration reconfigInterval = Duration.ofSeconds(5);
 
@@ -46,8 +37,11 @@ public class Reconfigurer extends AbstractComponent {
     private ZooKeeperRunner zooKeeperRunner;
     private ZookeeperServerConfig activeConfig;
 
+    protected final ZkAdmin zkAdmin;
+
     @Inject
-    public Reconfigurer() {
+    public Reconfigurer(ZkAdmin zkAdmin) {
+        this.zkAdmin = zkAdmin;
         log.log(Level.FINE, "Created ZooKeeperReconfigurer");
     }
 
@@ -67,20 +61,6 @@ public class Reconfigurer extends AbstractComponent {
         return activeConfig;
     }
 
-    void zooKeeperReconfigure(String connectionSpec, String joiningServers, String leavingServers) throws KeeperException {
-        try {
-            ZooKeeperAdmin zooKeeperAdmin = new ZooKeeperAdmin(connectionSpec,
-                                                               (int) sessionTimeout.toMillis(),
-                                                               new LoggingWatcher());
-            long fromConfig = -1;
-            // Using string parameters because the List variant of reconfigure fails to join empty lists (observed on 3.5.6, fixed in 3.7.0)
-            byte[] appliedConfig = zooKeeperAdmin.reconfigure(joiningServers, leavingServers, null, fromConfig, null);
-            log.log(Level.INFO, "Applied ZooKeeper config: " + new String(appliedConfig, StandardCharsets.UTF_8));
-        } catch (IOException | InterruptedException e) {
-            throw new RuntimeException(e);
-        }
-    }
-
     void shutdown() {
         if (zooKeeperRunner != null) {
             zooKeeperRunner.shutdown();
@@ -115,16 +95,14 @@ public class Reconfigurer extends AbstractComponent {
         for (int attempts = 1; ! reconfigured && Instant.now().isBefore(end); attempts++) {
             try {
                 Instant reconfigStarted = Instant.now();
-                zooKeeperReconfigure(connectionSpec, joiningServers, leavingServers);
+                zkAdmin.reconfigure(connectionSpec, joiningServers, leavingServers);
                 Instant reconfigEnded = Instant.now();
                 log.log(Level.INFO, "Reconfiguration completed in " +
                                     Duration.between(reconfigTriggered, reconfigEnded) +
                                     ", after " + attempts + " attempt(s). ZooKeeper reconfig call took " +
                                     Duration.between(reconfigStarted, reconfigEnded));
                 reconfigured = true;
-            } catch (KeeperException e) {
-                if (!retryOn(e))
-                    throw new RuntimeException(e);
+            } catch (ReconfigException e) {
                 log.log(Level.INFO, "Reconfiguration failed. Retrying in " + retryWait + ": " +
                                     Exceptions.toMessageString(e));
                 sleeper.accept(retryWait);
@@ -139,11 +117,6 @@ public class Reconfigurer extends AbstractComponent {
         return reconfigInterval.multipliedBy(activeConfig.myid());
     }
 
-    private static boolean retryOn(KeeperException e) {
-        return e instanceof KeeperException.ReconfigInProgress ||
-               e instanceof  KeeperException.ConnectionLossException;
-    }
-
     private static String connectionSpec(ZookeeperServerConfig config) {
         return config.server().stream()
                      .map(server -> server.hostname() + ":" + config.clientPort())
@@ -179,13 +152,4 @@ public class Reconfigurer extends AbstractComponent {
         }
     }
 
-    private static class LoggingWatcher implements Watcher {
-
-        @Override
-        public void process(WatchedEvent event) {
-            log.log(Level.INFO, event.toString());
-        }
-
-    }
-
 }
diff --git a/zookeeper-server/zookeeper-server-3.5.6/src/main/java/com/yahoo/vespa/zookeeper/ZkAdminImpl.java b/zookeeper-server/zookeeper-server-3.5.6/src/main/java/com/yahoo/vespa/zookeeper/ZkAdminImpl.java
new file mode 100644
index 00000000000..f7d0abee2d0
--- /dev/null
+++ b/zookeeper-server/zookeeper-server-3.5.6/src/main/java/com/yahoo/vespa/zookeeper/ZkAdminImpl.java
@@ -0,0 +1,42 @@
+// 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 org.apache.zookeeper.KeeperException;
+import org.apache.zookeeper.admin.ZooKeeperAdmin;
+
+import java.io.IOException;
+import java.nio.charset.StandardCharsets;
+import java.util.logging.Level;
+import java.util.logging.Logger;
+
+public class ZkAdminImpl implements ZkAdmin {
+
+    private static final Logger log = java.util.logging.Logger.getLogger(ZkAdminImpl.class.getName());
+
+    @Override
+    public void reconfigure(String connectionSpec, String joiningServers, String leavingServers) throws ReconfigException {
+        try {
+            ZooKeeperAdmin zooKeeperAdmin = new ZooKeeperAdmin(connectionSpec,
+                                                               (int) sessionTimeout().toMillis(),
+                                                               new LoggingWatcher());
+            long fromConfig = -1;
+            // Using string parameters because the List variant of reconfigure fails to join empty lists (observed on 3.5.6, fixed in 3.7.0)
+            byte[] appliedConfig = zooKeeperAdmin.reconfigure(joiningServers, leavingServers, null, fromConfig, null);
+            log.log(Level.INFO, "Applied ZooKeeper config: " + new String(appliedConfig, StandardCharsets.UTF_8));
+        } catch (KeeperException e) {
+            if (retryOn(e))
+                throw new ReconfigException(e);
+            else
+                throw new RuntimeException(e);
+        } catch (IOException | InterruptedException e) {
+            throw new RuntimeException(e);
+        }
+    }
+
+    private static boolean retryOn(KeeperException e) {
+        return e instanceof KeeperException.ReconfigInProgress ||
+               e instanceof KeeperException.ConnectionLossException;
+    }
+
+}
+
diff --git a/zookeeper-server/zookeeper-server-3.5.6/src/test/java/com/yahoo/vespa/zookeeper/ReconfigurerTest.java b/zookeeper-server/zookeeper-server-3.5.6/src/test/java/com/yahoo/vespa/zookeeper/ReconfigurerTest.java
index bf1f9ca5611..e8ee17192ee 100644
--- a/zookeeper-server/zookeeper-server-3.5.6/src/test/java/com/yahoo/vespa/zookeeper/ReconfigurerTest.java
+++ b/zookeeper-server/zookeeper-server-3.5.6/src/test/java/com/yahoo/vespa/zookeeper/ReconfigurerTest.java
@@ -2,7 +2,6 @@
 package com.yahoo.vespa.zookeeper;
 
 import com.yahoo.cloud.config.ZookeeperServerConfig;
-import org.apache.zookeeper.KeeperException;
 import org.junit.After;
 import org.junit.Before;
 import org.junit.Rule;
@@ -35,7 +34,7 @@ public class ReconfigurerTest {
     public void setup() throws IOException {
         cfgFile = folder.newFile();
         idFile = folder.newFile("myid");
-        reconfigurer = new TestableReconfigurer();
+        reconfigurer = new TestableReconfigurer(new TestableZkAdmin());
     }
 
     @Test
@@ -47,40 +46,40 @@ public class ReconfigurerTest {
         // Cluster grows
         ZookeeperServerConfig nextConfig = createConfig(5, true);
         reconfigurer.startOrReconfigure(nextConfig);
-        assertEquals("node0:2181,node1:2181,node2:2181", reconfigurer.connectionSpec);
-        assertEquals("3=node3:2182:2183;2181,4=node4:2182:2183;2181", reconfigurer.joiningServers);
-        assertNull("No servers are leaving", reconfigurer.leavingServers);
-        assertEquals(1, reconfigurer.reconfigurations);
+        assertEquals("node0:2181,node1:2181,node2:2181", reconfigurer.connectionSpec());
+        assertEquals("3=node3:2182:2183;2181,4=node4:2182:2183;2181", reconfigurer.joiningServers());
+        assertNull("No servers are leaving", reconfigurer.leavingServers());
+        assertEquals(1, reconfigurer.reconfigurations());
         assertSame(nextConfig, reconfigurer.activeConfig());
 
         // No reconfiguration happens with same config
         reconfigurer.startOrReconfigure(nextConfig);
-        assertEquals(1, reconfigurer.reconfigurations);
+        assertEquals(1, reconfigurer.reconfigurations());
         assertSame(nextConfig, reconfigurer.activeConfig());
 
         // Cluster shrinks
         nextConfig = createConfig(3, true);
         reconfigurer.startOrReconfigure(nextConfig);
-        assertEquals(2, reconfigurer.reconfigurations);
-        assertEquals("node0:2181,node1:2181,node2:2181,node3:2181,node4:2181", reconfigurer.connectionSpec);
-        assertNull("No servers are joining", reconfigurer.joiningServers);
-        assertEquals("3,4", reconfigurer.leavingServers);
+        assertEquals(2, reconfigurer.reconfigurations());
+        assertEquals("node0:2181,node1:2181,node2:2181,node3:2181,node4:2181", reconfigurer.connectionSpec());
+        assertNull("No servers are joining", reconfigurer.joiningServers());
+        assertEquals("3,4", reconfigurer.leavingServers());
         assertSame(nextConfig, reconfigurer.activeConfig());
     }
 
     @Test
     public void testReconfigureFailsWithReconfigInProgressThenSucceeds() {
-        reconfigurer = new TemporarilyFailWithReconfigInProgressReconfigurer();
+        reconfigurer = new TestableReconfigurer(new TemporarilyFailZkAdmin());
         ZookeeperServerConfig initialConfig = createConfig(3, true);
         reconfigurer.startOrReconfigure(initialConfig);
         assertSame(initialConfig, reconfigurer.activeConfig());
 
         ZookeeperServerConfig nextConfig = createConfig(5, true);
         reconfigurer.startOrReconfigure(nextConfig);
-        assertEquals("node0:2181,node1:2181,node2:2181", reconfigurer.connectionSpec);
-        assertEquals("3=node3:2182:2183;2181,4=node4:2182:2183;2181", reconfigurer.joiningServers);
-        assertNull("No servers are leaving", reconfigurer.leavingServers);
-        assertEquals(1, reconfigurer.reconfigurations);
+        assertEquals("node0:2181,node1:2181,node2:2181", reconfigurer.connectionSpec());
+        assertEquals("3=node3:2182:2183;2181,4=node4:2182:2183;2181", reconfigurer.joiningServers());
+        assertNull("No servers are leaving", reconfigurer.leavingServers());
+        assertEquals(1, reconfigurer.reconfigurations());
         assertSame(nextConfig, reconfigurer.activeConfig());
     }
 
@@ -93,7 +92,7 @@ public class ReconfigurerTest {
         ZookeeperServerConfig nextConfig = createConfig(5, false);
         reconfigurer.startOrReconfigure(nextConfig);
         assertSame(initialConfig, reconfigurer.activeConfig());
-        assertEquals(0, reconfigurer.reconfigurations);
+        assertEquals(0, reconfigurer.reconfigurations());
     }
 
     @After
@@ -120,19 +119,45 @@ public class ReconfigurerTest {
 
     private static class TestableReconfigurer extends Reconfigurer {
 
-        private int reconfigurations = 0;
+        private final TestableZkAdmin zkReconfigurer;
 
-        private String connectionSpec;
-        private String joiningServers;
-        private String leavingServers;
+        TestableReconfigurer(TestableZkAdmin zkReconfigurer) {
+            super(zkReconfigurer);
+            this.zkReconfigurer = zkReconfigurer;
+        }
 
         @Override
         void startOrReconfigure(ZookeeperServerConfig newConfig) {
-            super.startOrReconfigure(newConfig, l->{});
+            super.startOrReconfigure(newConfig, l -> { });
+        }
+
+        String connectionSpec() {
+            return zkReconfigurer.connectionSpec;
+        }
+
+        String joiningServers() {
+            return zkReconfigurer.joiningServers;
+        }
+
+        String leavingServers() {
+            return zkReconfigurer.leavingServers;
         }
 
+        int reconfigurations() {
+            return zkReconfigurer.reconfigurations;
+        }
+
+    }
+
+    private static class TestableZkAdmin implements ZkAdmin {
+
+        String connectionSpec;
+        String joiningServers;
+        String leavingServers;
+        int reconfigurations = 0;
+
         @Override
-        void zooKeeperReconfigure(String connectionSpec, String joiningServers, String leavingServers) throws KeeperException {
+        public void reconfigure(String connectionSpec, String joiningServers, String leavingServers) throws ReconfigException {
             this.connectionSpec = connectionSpec;
             this.joiningServers = joiningServers;
             this.leavingServers = leavingServers;
@@ -142,18 +167,18 @@ public class ReconfigurerTest {
     }
 
     // Fails 3 times with KeeperException.ReconfigInProgress(), then succeeds
-    private static class TemporarilyFailWithReconfigInProgressReconfigurer extends TestableReconfigurer {
+    private static class TemporarilyFailZkAdmin extends TestableZkAdmin {
 
         private int attempts = 0;
 
-        @Override
-        void zooKeeperReconfigure(String connectionSpec, String joiningServers, String leavingServers) throws KeeperException {
+        public void reconfigure(String connectionSpec, String joiningServers, String leavingServers) throws ReconfigException {
             if (++attempts < 3)
-                throw new KeeperException.ReconfigInProgress();
+                throw new ReconfigException("Reconfig failed");
             else
-                super.zooKeeperReconfigure(connectionSpec, joiningServers, leavingServers);
+                super.reconfigure(connectionSpec, joiningServers, leavingServers);
         }
 
     }
 
+
 }
diff --git a/zookeeper-server/zookeeper-server-common/src/main/java/com/yahoo/vespa/zookeeper/ReconfigException.java b/zookeeper-server/zookeeper-server-common/src/main/java/com/yahoo/vespa/zookeeper/ReconfigException.java
new file mode 100644
index 00000000000..ade9245615b
--- /dev/null
+++ b/zookeeper-server/zookeeper-server-common/src/main/java/com/yahoo/vespa/zookeeper/ReconfigException.java
@@ -0,0 +1,19 @@
+// Copyright Verizon Media. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root.
+package com.yahoo.vespa.zookeeper;
+
+/**
+ * Interface for reconfiguring a zookeeper cluster.
+ *
+ * @author hmusum
+ */
+@SuppressWarnings("serial")
+public class ReconfigException extends RuntimeException {
+
+    public ReconfigException(Throwable cause) {
+        super(cause);
+    }
+
+    public ReconfigException(String message) {
+        super(message);
+    }
+}
diff --git a/zookeeper-server/zookeeper-server-common/src/main/java/com/yahoo/vespa/zookeeper/ZkAdmin.java b/zookeeper-server/zookeeper-server-common/src/main/java/com/yahoo/vespa/zookeeper/ZkAdmin.java
new file mode 100644
index 00000000000..fb62f662566
--- /dev/null
+++ b/zookeeper-server/zookeeper-server-common/src/main/java/com/yahoo/vespa/zookeeper/ZkAdmin.java
@@ -0,0 +1,18 @@
+// 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.time.Duration;
+
+/**
+ * Interface for administering a zookeeper cluster. Currently only supports reconfiguring a zookeeper cluster.
+ *
+ * @author hmusum
+ */
+public interface ZkAdmin {
+
+    void reconfigure(String connectionSpec, String joiningServers, String leavingServers) throws ReconfigException;
+
+    /* Timeout for connecting to ZooKeeper */
+    default Duration sessionTimeout() { return Duration.ofSeconds(30); }
+
+}