ZkImpl implements Zk { @Override public Zk create(String path, byte[] data, List<ACL> acls, CreateMode createMode, Handler<AsyncResult<Void>> handler) { workerPool().executeBlocking( future -> { try { zookeeper.create(path, data == null ? new byte[0] : data, acls, createMode); future.complete(); } catch (Throwable t) { future.fail(t); } }, handler); return this; } ZkImpl(Vertx vertx, ZkClient zkClient); @Override Zk create(String path, byte[] data, List<ACL> acls, CreateMode createMode, Handler<AsyncResult<Void>> handler); @Override Zk setData(String path, byte[] data, int version, Handler<AsyncResult<Void>> handler); @Override Zk disconnect(Handler<AsyncResult<Void>> handler); @Override Zk getData(String path, Handler<AsyncResult<byte[]>> handler); @Override Future<Zk> watchData(String path, Handler<AsyncResult<byte[]>> watcher); @Override Zk unwatchData(String path); @Override Zk delete(String path, int version, Handler<AsyncResult<Void>> handler); @Override Zk children(String path, Handler<AsyncResult<List<String>>> handler); @Override Future<Zk> watchChildren(String path, Handler<AsyncResult<List<String>>> watcher); @Override Zk unwatchChildren(String path);  }
@Disabled @Test public void testReconnectOnBounce(VertxTestContext context) throws IOException, InterruptedException { Checkpoint async = context.checkpoint(); Zk zkImpl = Zk.createSync(vertx, zkServer.getZkConnectString(), 60_000, 10_000); zkServer.restart(); Promise fooCreated = Promise.promise(); zkImpl.create("/foo", null, AclBuilder.PUBLIC, CreateMode.PERSISTENT, context.succeeding(v -> fooCreated.complete())); fooCreated.future().compose(v -> { try { zkServer.restart(); Thread.sleep(2000); zkImpl.create("/bar", null, AclBuilder.PUBLIC, CreateMode.PERSISTENT, context.succeeding(vv -> async.flag())); } catch (Exception e) { context.failNow(e); } return Future.succeededFuture(); }); }