ZookeeperScaler implements AutoCloseable { public Future<Void> scale(int scaleTo) { return connect() .compose(zkAdmin -> { Promise<Void> scalePromise = Promise.promise(); getCurrentConfig(zkAdmin) .compose(servers -> scaleTo(zkAdmin, servers, scaleTo)) .onComplete(res -> { closeConnection(zkAdmin); if (res.succeeded()) { scalePromise.complete(); } else { scalePromise.fail(res.cause()); } }); return scalePromise.future(); }); } protected  ZookeeperScaler(Vertx vertx, ZooKeeperAdminProvider zooAdminProvider, String zookeeperConnectionString, Function<Integer, String> zkNodeAddress, Secret clusterCaCertSecret, Secret coKeySecret, long operationTimeoutMs); Future<Void> scale(int scaleTo); @Override void close();  }
@Test public void testConnectionTimeout(VertxTestContext context) { ZooKeeperAdmin mockZooAdmin = mock(ZooKeeperAdmin.class); when(mockZooAdmin.getState()).thenReturn(ZooKeeper.States.NOT_CONNECTED); ZooKeeperAdminProvider zooKeeperAdminProvider = new ZooKeeperAdminProvider() { @Override public ZooKeeperAdmin createZookeeperAdmin(String connectString, int sessionTimeout, Watcher watcher, ZKClientConfig conf) throws IOException { return mockZooAdmin; } }; ZookeeperScaler scaler = new ZookeeperScaler(vertx, zooKeeperAdminProvider, "zookeeper:2181", null, dummyCaSecret, dummyCoSecret, 1_000); Checkpoint check = context.checkpoint(); scaler.scale(5).onComplete(context.failing(cause -> context.verify(() -> { assertThat(cause.getMessage(), is("Failed to connect to Zookeeper zookeeper:2181. Connection was not ready in 1000 ms.")); check.flag(); }))); }
@Test public void testNoChange(VertxTestContext context) throws KeeperException, InterruptedException { String config = "server.1=my-cluster-zookeeper-0.my-cluster-zookeeper-nodes.myproject.svc:2888:3888:participant;127.0.0.1:12181\n" + "version=100000000b"; ZooKeeperAdmin mockZooAdmin = mock(ZooKeeperAdmin.class); when(mockZooAdmin.getConfig(false, null)).thenReturn(config.getBytes(StandardCharsets.US_ASCII)); when(mockZooAdmin.getState()).thenReturn(ZooKeeper.States.CONNECTED); ZooKeeperAdminProvider zooKeeperAdminProvider = new ZooKeeperAdminProvider() { @Override public ZooKeeperAdmin createZookeeperAdmin(String connectString, int sessionTimeout, Watcher watcher, ZKClientConfig conf) throws IOException { watcher.process(new WatchedEvent(null, Watcher.Event.KeeperState.SyncConnected, null)); return mockZooAdmin; } }; ZookeeperScaler scaler = new ZookeeperScaler(vertx, zooKeeperAdminProvider, "zookeeper:2181", zkNodeAddress, dummyCaSecret, dummyCoSecret, 1_000); Checkpoint check = context.checkpoint(); scaler.scale(1).onComplete(context.succeeding(res -> context.verify(() -> { verify(mockZooAdmin, never()).reconfigure(isNull(), isNull(), anyList(), anyLong(), isNull()); check.flag(); }))); }
@Test public void testWithChange(VertxTestContext context) throws KeeperException, InterruptedException { String config = "server.1=my-cluster-zookeeper-0.my-cluster-zookeeper-nodes.myproject.svc:2888:3888:participant;127.0.0.1:12181\n" + "server.2=my-cluster-zookeeper-1.my-cluster-zookeeper-nodes.myproject.svc:2888:3888:participant;127.0.0.1:12181\n" + "version=100000000b"; String updated = "server.1=my-cluster-zookeeper-0.my-cluster-zookeeper-nodes.myproject.svc:2888:3888:participant;127.0.0.1:12181\n" + "version=100000000b"; ZooKeeperAdmin mockZooAdmin = mock(ZooKeeperAdmin.class); when(mockZooAdmin.getConfig(false, null)).thenReturn(config.getBytes(StandardCharsets.US_ASCII)); when(mockZooAdmin.reconfigure(isNull(), isNull(), anyList(), anyLong(), isNull())).thenReturn(updated.getBytes(StandardCharsets.US_ASCII)); when(mockZooAdmin.getState()).thenReturn(ZooKeeper.States.CONNECTED); ZooKeeperAdminProvider zooKeeperAdminProvider = new ZooKeeperAdminProvider() { @Override public ZooKeeperAdmin createZookeeperAdmin(String connectString, int sessionTimeout, Watcher watcher, ZKClientConfig conf) throws IOException { watcher.process(new WatchedEvent(null, Watcher.Event.KeeperState.SyncConnected, null)); return mockZooAdmin; } }; ZookeeperScaler scaler = new ZookeeperScaler(vertx, zooKeeperAdminProvider, "zookeeper:2181", zkNodeAddress, dummyCaSecret, dummyCoSecret, 1_000); Checkpoint check = context.checkpoint(); scaler.scale(1).onComplete(context.succeeding(res -> context.verify(() -> { verify(mockZooAdmin, times(1)).reconfigure(isNull(), isNull(), anyList(), anyLong(), isNull()); check.flag(); }))); }
@Test public void testWhenThrows(VertxTestContext context) throws KeeperException, InterruptedException { String config = "server.1=my-cluster-zookeeper-0.my-cluster-zookeeper-nodes.myproject.svc:2888:3888:participant;127.0.0.1:12181\n" + "server.2=my-cluster-zookeeper-1.my-cluster-zookeeper-nodes.myproject.svc:2888:3888:participant;127.0.0.1:12181\n" + "version=100000000b"; ZooKeeperAdmin mockZooAdmin = mock(ZooKeeperAdmin.class); when(mockZooAdmin.getConfig(false, null)).thenReturn(config.getBytes(StandardCharsets.US_ASCII)); when(mockZooAdmin.reconfigure(isNull(), isNull(), anyList(), anyLong(), isNull())).thenThrow(new KeeperException.NewConfigNoQuorum()); when(mockZooAdmin.getState()).thenReturn(ZooKeeper.States.CONNECTED); ZooKeeperAdminProvider zooKeeperAdminProvider = new ZooKeeperAdminProvider() { @Override public ZooKeeperAdmin createZookeeperAdmin(String connectString, int sessionTimeout, Watcher watcher, ZKClientConfig conf) throws IOException { watcher.process(new WatchedEvent(null, Watcher.Event.KeeperState.SyncConnected, null)); return mockZooAdmin; } }; ZookeeperScaler scaler = new ZookeeperScaler(vertx, zooKeeperAdminProvider, "zookeeper:2181", zkNodeAddress, dummyCaSecret, dummyCoSecret, 1_000); Checkpoint check = context.checkpoint(); scaler.scale(1).onComplete(context.failing(cause -> context.verify(() -> { assertThat(cause.getCause(), instanceOf(KeeperException.class)); check.flag(); }))); }
@Test public void testConnectionToNonExistingHost(VertxTestContext context) { ZookeeperScaler scaler = new ZookeeperScaler(vertx, new DefaultZooKeeperAdminProvider(), "i-do-not-exist.com:2181", null, dummyCaSecret, dummyCoSecret, 2_000); Checkpoint check = context.checkpoint(); scaler.scale(5).onComplete(context.failing(cause -> context.verify(() -> { assertThat(cause.getMessage(), is("Failed to connect to Zookeeper i-do-not-exist.com:2181. Connection was not ready in 2000 ms.")); check.flag(); }))); }