diff --git a/zookeeper-server/src/main/java/org/apache/zookeeper/ZooKeeper.java b/zookeeper-server/src/main/java/org/apache/zookeeper/ZooKeeper.java index 09006ed103e..475430b89ec 100644 --- a/zookeeper-server/src/main/java/org/apache/zookeeper/ZooKeeper.java +++ b/zookeeper-server/src/main/java/org/apache/zookeeper/ZooKeeper.java @@ -46,6 +46,8 @@ import org.apache.zookeeper.client.HostProvider; import org.apache.zookeeper.client.StaticHostProvider; import org.apache.zookeeper.client.ZKClientConfig; +import org.apache.zookeeper.client.ZooKeeperBuilder; +import org.apache.zookeeper.client.ZooKeeperOptions; import org.apache.zookeeper.client.ZooKeeperSaslClient; import org.apache.zookeeper.common.PathUtils; import org.apache.zookeeper.data.ACL; @@ -445,7 +447,9 @@ public boolean isConnected() { * if an invalid chroot path is specified */ public ZooKeeper(String connectString, int sessionTimeout, Watcher watcher) throws IOException { - this(connectString, sessionTimeout, watcher, false); + this(new ZooKeeperBuilder(connectString, sessionTimeout) + .withDefaultWatcher(watcher) + .toOptions()); } /** @@ -498,7 +502,10 @@ public ZooKeeper( int sessionTimeout, Watcher watcher, ZKClientConfig conf) throws IOException { - this(connectString, sessionTimeout, watcher, false, conf); + this(new ZooKeeperBuilder(connectString, sessionTimeout) + .withDefaultWatcher(watcher) + .withClientConfig(conf) + .toOptions()); } /** @@ -564,7 +571,11 @@ public ZooKeeper( Watcher watcher, boolean canBeReadOnly, HostProvider aHostProvider) throws IOException { - this(connectString, sessionTimeout, watcher, canBeReadOnly, aHostProvider, null); + this(new ZooKeeperBuilder(connectString, sessionTimeout) + .withDefaultWatcher(watcher) + .withCanBeReadOnly(canBeReadOnly) + .withHostProvider(ignored -> aHostProvider) + .toOptions()); } /** @@ -634,25 +645,12 @@ public ZooKeeper( HostProvider hostProvider, ZKClientConfig clientConfig ) throws IOException { - LOG.info( - "Initiating client connection, connectString={} sessionTimeout={} watcher={}", - connectString, - sessionTimeout, - watcher); - - this.clientConfig = clientConfig != null ? clientConfig : new ZKClientConfig(); - this.hostProvider = hostProvider; - ConnectStringParser connectStringParser = new ConnectStringParser(connectString); - - cnxn = createConnection( - connectStringParser.getChrootPath(), - hostProvider, - sessionTimeout, - this.clientConfig, - watcher, - getClientCnxnSocket(), - canBeReadOnly); - cnxn.start(); + this(new ZooKeeperBuilder(connectString, sessionTimeout) + .withDefaultWatcher(watcher) + .withCanBeReadOnly(canBeReadOnly) + .withHostProvider(ignored -> hostProvider) + .withClientConfig(clientConfig) + .toOptions()); } ClientCnxn createConnection( @@ -662,6 +660,8 @@ ClientCnxn createConnection( ZKClientConfig clientConfig, Watcher defaultWatcher, ClientCnxnSocket clientCnxnSocket, + long sessionId, + byte[] sessionPasswd, boolean canBeReadOnly ) throws IOException { return new ClientCnxn( @@ -671,6 +671,8 @@ ClientCnxn createConnection( clientConfig, defaultWatcher, clientCnxnSocket, + sessionId, + sessionPasswd, canBeReadOnly); } @@ -731,7 +733,10 @@ public ZooKeeper( int sessionTimeout, Watcher watcher, boolean canBeReadOnly) throws IOException { - this(connectString, sessionTimeout, watcher, canBeReadOnly, createDefaultHostProvider(connectString)); + this(new ZooKeeperBuilder(connectString, sessionTimeout) + .withDefaultWatcher(watcher) + .withCanBeReadOnly(canBeReadOnly) + .toOptions()); } /** @@ -794,13 +799,11 @@ public ZooKeeper( Watcher watcher, boolean canBeReadOnly, ZKClientConfig conf) throws IOException { - this( - connectString, - sessionTimeout, - watcher, - canBeReadOnly, - createDefaultHostProvider(connectString), - conf); + this(new ZooKeeperBuilder(connectString, sessionTimeout) + .withDefaultWatcher(watcher) + .withCanBeReadOnly(canBeReadOnly) + .withClientConfig(conf) + .toOptions()); } /** @@ -861,7 +864,10 @@ public ZooKeeper( Watcher watcher, long sessionId, byte[] sessionPasswd) throws IOException { - this(connectString, sessionTimeout, watcher, sessionId, sessionPasswd, false); + this(new ZooKeeperBuilder(connectString, sessionTimeout) + .withDefaultWatcher(watcher) + .withSession(sessionId, sessionPasswd) + .toOptions()); } /** @@ -936,15 +942,12 @@ public ZooKeeper( byte[] sessionPasswd, boolean canBeReadOnly, HostProvider aHostProvider) throws IOException { - this( - connectString, - sessionTimeout, - watcher, - sessionId, - sessionPasswd, - canBeReadOnly, - aHostProvider, - null); + this(new ZooKeeperBuilder(connectString, sessionTimeout) + .withDefaultWatcher(watcher) + .withSession(sessionId, sessionPasswd) + .withCanBeReadOnly(canBeReadOnly) + .withHostProvider(ignored -> aHostProvider) + .toOptions()); } /** @@ -1025,20 +1028,72 @@ public ZooKeeper( boolean canBeReadOnly, HostProvider hostProvider, ZKClientConfig clientConfig) throws IOException { - LOG.info( - "Initiating client connection, connectString={} " - + "sessionTimeout={} watcher={} sessionId=0x{} sessionPasswd={}", - connectString, - sessionTimeout, - watcher, - Long.toHexString(sessionId), - (sessionPasswd == null ? "" : "")); + this(new ZooKeeperBuilder(connectString, sessionTimeout) + .withSession(sessionId, sessionPasswd) + .withDefaultWatcher(watcher) + .withCanBeReadOnly(canBeReadOnly) + .withHostProvider(ignored -> hostProvider) + .withClientConfig(clientConfig) + .toOptions()); + } + /** + * Create a ZooKeeper client and establish session asynchronously. + * + *

This constructor will initiate connection to the server and return + * immediately - potentially (usually) before the session is fully established. + * The watcher from options will be notified of any changes in state. This + * notification can come at any point before or after the constructor call + * has returned. + * + *

The instantiated ZooKeeper client object will pick an arbitrary server + * from the connect string and attempt to connect to it. If establishment of + * the connection fails, another server in the connect string will be tried + * (the order is non-deterministic, as we random shuffle the list), until a + * connection is established. The client will continue attempts until the + * session is explicitly closed (or the session is expired by the server). + * + * @param options options for ZooKeeper client + * @throws IOException in cases of IO failure + */ + @InterfaceAudience.Private + public ZooKeeper(ZooKeeperOptions options) throws IOException { + String connectString = options.getConnectString(); + int sessionTimeout = options.getSessionTimeout(); + long sessionId = options.getSessionId(); + byte[] sessionPasswd = sessionId == 0 ? new byte[16] : options.getSessionPasswd(); + Watcher watcher = options.getDefaultWatcher(); + boolean canBeReadOnly = options.isCanBeReadOnly(); + + if (sessionId == 0) { + LOG.info( + "Initiating client connection, connectString={} sessionTimeout={} watcher={}", + connectString, + sessionTimeout, + watcher); + } else { + LOG.info( + "Initiating client connection, connectString={} " + + "sessionTimeout={} watcher={} sessionId=0x{} sessionPasswd={}", + connectString, + sessionTimeout, + watcher, + Long.toHexString(sessionId), + (sessionPasswd == null ? "" : "")); + } + + ZKClientConfig clientConfig = options.getClientConfig(); this.clientConfig = clientConfig != null ? clientConfig : new ZKClientConfig(); ConnectStringParser connectStringParser = new ConnectStringParser(connectString); + HostProvider hostProvider; + if (options.getHostProvider() != null) { + hostProvider = options.getHostProvider().apply(connectStringParser.getServerAddresses()); + } else { + hostProvider = new StaticHostProvider(connectStringParser.getServerAddresses()); + } this.hostProvider = hostProvider; - cnxn = new ClientCnxn( + cnxn = createConnection( connectStringParser.getChrootPath(), hostProvider, sessionTimeout, @@ -1048,7 +1103,7 @@ public ZooKeeper( sessionId, sessionPasswd, canBeReadOnly); - cnxn.seenRwServerBefore = true; // since user has provided sessionId + cnxn.seenRwServerBefore = sessionId != 0; // since user has provided sessionId cnxn.start(); } @@ -1120,19 +1175,11 @@ public ZooKeeper( long sessionId, byte[] sessionPasswd, boolean canBeReadOnly) throws IOException { - this( - connectString, - sessionTimeout, - watcher, - sessionId, - sessionPasswd, - canBeReadOnly, - createDefaultHostProvider(connectString)); - } - - // default hostprovider - private static HostProvider createDefaultHostProvider(String connectString) { - return new StaticHostProvider(new ConnectStringParser(connectString).getServerAddresses()); + this(new ZooKeeperBuilder(connectString, sessionTimeout) + .withDefaultWatcher(watcher) + .withSession(sessionId, sessionPasswd) + .withCanBeReadOnly(canBeReadOnly) + .toOptions()); } // VisibleForTesting diff --git a/zookeeper-server/src/main/java/org/apache/zookeeper/admin/ZooKeeperAdmin.java b/zookeeper-server/src/main/java/org/apache/zookeeper/admin/ZooKeeperAdmin.java index 8240526a513..4d5dc6fd263 100644 --- a/zookeeper-server/src/main/java/org/apache/zookeeper/admin/ZooKeeperAdmin.java +++ b/zookeeper-server/src/main/java/org/apache/zookeeper/admin/ZooKeeperAdmin.java @@ -27,6 +27,7 @@ import org.apache.zookeeper.ZooDefs; import org.apache.zookeeper.ZooKeeper; import org.apache.zookeeper.client.ZKClientConfig; +import org.apache.zookeeper.client.ZooKeeperOptions; import org.apache.zookeeper.common.StringUtils; import org.apache.zookeeper.data.Stat; import org.apache.zookeeper.proto.GetDataResponse; @@ -53,6 +54,11 @@ public class ZooKeeperAdmin extends ZooKeeper { private static final Logger LOG = LoggerFactory.getLogger(ZooKeeperAdmin.class); + @InterfaceAudience.Private + public ZooKeeperAdmin(ZooKeeperOptions options) throws IOException { + super(options); + } + /** * Create a ZooKeeperAdmin object which is used to perform dynamic reconfiguration * operations. diff --git a/zookeeper-server/src/main/java/org/apache/zookeeper/client/ZooKeeperBuilder.java b/zookeeper-server/src/main/java/org/apache/zookeeper/client/ZooKeeperBuilder.java new file mode 100644 index 00000000000..796ed24ea08 --- /dev/null +++ b/zookeeper-server/src/main/java/org/apache/zookeeper/client/ZooKeeperBuilder.java @@ -0,0 +1,181 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.zookeeper.client; + +import edu.umd.cs.findbugs.annotations.SuppressFBWarnings; +import java.io.IOException; +import java.net.InetSocketAddress; +import java.util.Collection; +import java.util.function.Function; +import org.apache.yetus.audience.InterfaceAudience; +import org.apache.yetus.audience.InterfaceStability; +import org.apache.zookeeper.Watcher; +import org.apache.zookeeper.ZooKeeper; +import org.apache.zookeeper.admin.ZooKeeperAdmin; + +/** + * Builder to construct {@link ZooKeeper} and {@link ZooKeeperAdmin}. + */ +@InterfaceAudience.Public +@InterfaceStability.Evolving +public class ZooKeeperBuilder { + private final String connectString; + private final int sessionTimeout; + private Function, HostProvider> hostProvider; + private Watcher defaultWatcher; + private boolean canBeReadOnly = false; + private long sessionId = 0; + private byte[] sessionPasswd; + private ZKClientConfig clientConfig; + + /** + * Creates a builder with given connect string and session timeout. + * + * @param connectString + * comma separated host:port pairs, each corresponding to a zk + * server. e.g. "127.0.0.1:3000,127.0.0.1:3001,127.0.0.1:3002" + * If the optional chroot suffix is used the example would look + * like: "127.0.0.1:3000,127.0.0.1:3001,127.0.0.1:3002/app/a" + * where the client would be rooted at "/app/a" and all paths + * would be relative to this root - ie getting/setting/etc... + * "/foo/bar" would result in operations being run on + * "/app/a/foo/bar" (from the server perspective). + * @param sessionTimeoutMs + * session timeout in milliseconds + */ + public ZooKeeperBuilder(String connectString, int sessionTimeoutMs) { + this.connectString = connectString; + this.sessionTimeout = sessionTimeoutMs; + } + + /** + * Specified watcher to receive state changes, and node events if attached later. + * + * @param watcher + * a watcher object which will be notified of state changes, may + * also be notified for node events + * @return this + */ + public ZooKeeperBuilder withDefaultWatcher(Watcher watcher) { + this.defaultWatcher = watcher; + return this; + } + + /** + * Specifies a function to construct a {@link HostProvider} with initial server addresses from connect string. + * + * @param hostProvider + * use this as HostProvider to enable custom behaviour. + * @return this + */ + public ZooKeeperBuilder withHostProvider(Function, HostProvider> hostProvider) { + this.hostProvider = hostProvider; + return this; + } + + /** + * Specifies whether the created client is allowed to go to read-only mode in case of partitioning. + * + * @param canBeReadOnly + * whether the created client is allowed to go to + * read-only mode in case of partitioning. Read-only mode + * basically means that if the client can't find any majority + * servers but there's partitioned server it could reach, it + * connects to one in read-only mode, i.e. read requests are + * allowed while write requests are not. It continues seeking for + * majority in the background. + * @return this + * @since 3.4 + */ + public ZooKeeperBuilder withCanBeReadOnly(boolean canBeReadOnly) { + this.canBeReadOnly = canBeReadOnly; + return this; + } + + /** + * Specifies session id and password in session reestablishment. + * + * @param sessionId + * session id to use if reconnecting, otherwise 0 to open new session + * @param sessionPasswd + * password for this session + * @return this + * @see ZooKeeper#getSessionId() + * @see ZooKeeper#getSessionPasswd() + */ + @SuppressFBWarnings({"EI_EXPOSE_REP", "EI_EXPOSE_REP2"}) + public ZooKeeperBuilder withSession(long sessionId, byte[] sessionPasswd) { + this.sessionId = sessionId; + this.sessionPasswd = sessionPasswd; + return this; + } + + /** + * Specifies the client config used to construct ZooKeeper instances. + * + * @param clientConfig + * passing this conf object gives each client the flexibility of + * configuring properties differently compared to other instances + * @return this + * @since 3.5.2 + */ + public ZooKeeperBuilder withClientConfig(ZKClientConfig clientConfig) { + this.clientConfig = clientConfig; + return this; + } + + /** + * Creates a {@link ZooKeeperOptions} with configured options. + * + * @apiNote helper to delegate existing constructors to {@link ZooKeeper#ZooKeeper(ZooKeeperOptions)} + */ + @InterfaceAudience.Private + public ZooKeeperOptions toOptions() { + return new ZooKeeperOptions( + connectString, + sessionTimeout, + defaultWatcher, + hostProvider, + canBeReadOnly, + sessionId, + sessionPasswd, + clientConfig + ); + } + + /** + * Constructs an instance of {@link ZooKeeper}. + * + * @return an instance of {@link ZooKeeper} + * @throws IOException from constructor of {@link ZooKeeper} + */ + public ZooKeeper build() throws IOException { + return new ZooKeeper(toOptions()); + } + + /** + * Constructs an instance of {@link ZooKeeperAdmin}. + * + * @return an instance of {@link ZooKeeperAdmin} + * @throws IOException from constructor of {@link ZooKeeperAdmin} + */ + public ZooKeeperAdmin buildAdmin() throws IOException { + return new ZooKeeperAdmin(toOptions()); + } +} diff --git a/zookeeper-server/src/main/java/org/apache/zookeeper/client/ZooKeeperOptions.java b/zookeeper-server/src/main/java/org/apache/zookeeper/client/ZooKeeperOptions.java new file mode 100644 index 00000000000..27e5244ba87 --- /dev/null +++ b/zookeeper-server/src/main/java/org/apache/zookeeper/client/ZooKeeperOptions.java @@ -0,0 +1,92 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.zookeeper.client; + +import edu.umd.cs.findbugs.annotations.SuppressFBWarnings; +import java.net.InetSocketAddress; +import java.util.Collection; +import java.util.function.Function; +import org.apache.yetus.audience.InterfaceAudience; +import org.apache.zookeeper.Watcher; + +/** + * Options to construct {@link org.apache.zookeeper.ZooKeeper} and {@link org.apache.zookeeper.admin.ZooKeeperAdmin}. + */ +@InterfaceAudience.Private +public class ZooKeeperOptions { + private final String connectString; + private final int sessionTimeout; + private final Watcher defaultWatcher; + private final Function, HostProvider> hostProvider; + private final boolean canBeReadOnly; + private final long sessionId; + private final byte[] sessionPasswd; + private final ZKClientConfig clientConfig; + + ZooKeeperOptions(String connectString, + int sessionTimeout, + Watcher defaultWatcher, + Function, HostProvider> hostProvider, + boolean canBeReadOnly, + long sessionId, + byte[] sessionPasswd, + ZKClientConfig clientConfig) { + this.connectString = connectString; + this.sessionTimeout = sessionTimeout; + this.hostProvider = hostProvider; + this.defaultWatcher = defaultWatcher; + this.canBeReadOnly = canBeReadOnly; + this.sessionId = sessionId; + this.sessionPasswd = sessionPasswd; + this.clientConfig = clientConfig; + } + + public String getConnectString() { + return connectString; + } + + public int getSessionTimeout() { + return sessionTimeout; + } + + public Watcher getDefaultWatcher() { + return defaultWatcher; + } + + public Function, HostProvider> getHostProvider() { + return hostProvider; + } + + public boolean isCanBeReadOnly() { + return canBeReadOnly; + } + + public long getSessionId() { + return sessionId; + } + + @SuppressFBWarnings({"EI_EXPOSE_REP", "EI_EXPOSE_REP2"}) + public byte[] getSessionPasswd() { + return sessionPasswd; + } + + public ZKClientConfig getClientConfig() { + return clientConfig; + } +} diff --git a/zookeeper-server/src/test/java/org/apache/zookeeper/ClientCnxnSocketFragilityTest.java b/zookeeper-server/src/test/java/org/apache/zookeeper/ClientCnxnSocketFragilityTest.java index 953414468f3..bd2f1c287a0 100644 --- a/zookeeper-server/src/test/java/org/apache/zookeeper/ClientCnxnSocketFragilityTest.java +++ b/zookeeper-server/src/test/java/org/apache/zookeeper/ClientCnxnSocketFragilityTest.java @@ -286,6 +286,8 @@ public CustomClientCnxn( ZKClientConfig zkClientConfig, Watcher defaultWatcher, ClientCnxnSocket clientCnxnSocket, + long sessionId, + byte[] sessionPasswd, boolean canBeReadOnly ) throws IOException { super( @@ -295,6 +297,8 @@ public CustomClientCnxn( zkClientConfig, defaultWatcher, clientCnxnSocket, + sessionId, + sessionPasswd, canBeReadOnly); } @@ -351,6 +355,7 @@ public boolean isAlive() { return cnxn.getState().isAlive(); } + @Override ClientCnxn createConnection( String chrootPath, HostProvider hostProvider, @@ -358,6 +363,8 @@ ClientCnxn createConnection( ZKClientConfig clientConfig, Watcher defaultWatcher, ClientCnxnSocket clientCnxnSocket, + long sessionId, + byte[] sessionPasswd, boolean canBeReadOnly ) throws IOException { assertTrue(clientCnxnSocket instanceof FragileClientCnxnSocketNIO); @@ -369,6 +376,8 @@ ClientCnxn createConnection( clientConfig, defaultWatcher, clientCnxnSocket, + sessionId, + sessionPasswd, canBeReadOnly); return ClientCnxnSocketFragilityTest.this.cnxn; } diff --git a/zookeeper-server/src/test/java/org/apache/zookeeper/ClientRequestTimeoutTest.java b/zookeeper-server/src/test/java/org/apache/zookeeper/ClientRequestTimeoutTest.java index ecf39273e65..d6b6da94b56 100644 --- a/zookeeper-server/src/test/java/org/apache/zookeeper/ClientRequestTimeoutTest.java +++ b/zookeeper-server/src/test/java/org/apache/zookeeper/ClientRequestTimeoutTest.java @@ -119,6 +119,8 @@ class CustomClientCnxn extends ClientCnxn { ZKClientConfig clientConfig, Watcher defaultWatcher, ClientCnxnSocket clientCnxnSocket, + long sessionId, + byte[] sessionPasswd, boolean canBeReadOnly ) throws IOException { super( @@ -128,6 +130,8 @@ class CustomClientCnxn extends ClientCnxn { clientConfig, defaultWatcher, clientCnxnSocket, + sessionId, + sessionPasswd, canBeReadOnly); } @@ -157,6 +161,8 @@ ClientCnxn createConnection( ZKClientConfig clientConfig, Watcher defaultWatcher, ClientCnxnSocket clientCnxnSocket, + long sessionId, + byte[] sessionPasswd, boolean canBeReadOnly ) throws IOException { return new CustomClientCnxn( @@ -166,6 +172,8 @@ ClientCnxn createConnection( clientConfig, defaultWatcher, clientCnxnSocket, + sessionId, + sessionPasswd, canBeReadOnly); } diff --git a/zookeeper-server/src/test/java/org/apache/zookeeper/client/ZooKeeperBuilderTest.java b/zookeeper-server/src/test/java/org/apache/zookeeper/client/ZooKeeperBuilderTest.java new file mode 100644 index 00000000000..32d30052a66 --- /dev/null +++ b/zookeeper-server/src/test/java/org/apache/zookeeper/client/ZooKeeperBuilderTest.java @@ -0,0 +1,86 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.zookeeper.client; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNotNull; +import java.util.concurrent.BlockingQueue; +import java.util.concurrent.LinkedBlockingQueue; +import java.util.concurrent.TimeUnit; +import org.apache.zookeeper.CreateMode; +import org.apache.zookeeper.WatchedEvent; +import org.apache.zookeeper.Watcher; +import org.apache.zookeeper.ZooDefs; +import org.apache.zookeeper.ZooKeeper; +import org.apache.zookeeper.common.Time; +import org.apache.zookeeper.test.ClientBase; +import org.junit.jupiter.api.Test; + +public class ZooKeeperBuilderTest extends ClientBase { + private void testClient(BlockingQueue events, ZooKeeper zk) throws Exception { + zk.exists("/test", true); + zk.create("/test", new byte[0], ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT); + Thread.sleep(100); + zk.close(); + + WatchedEvent connected = events.poll(10, TimeUnit.SECONDS); + assertNotNull(connected); + assertEquals(Watcher.Event.EventType.None, connected.getType()); + assertEquals(Watcher.Event.KeeperState.SyncConnected, connected.getState()); + + WatchedEvent created = events.poll(10, TimeUnit.SECONDS); + assertNotNull(created); + assertEquals(Watcher.Event.EventType.NodeCreated, created.getType()); + assertEquals("/test", created.getPath()); + + // A sleep(100) before disconnect approve that events receiving in closing is indeterminate, + // but the last should be closed. See ZOOKEEPER-4702. + WatchedEvent closed = null; + long timeoutMs = TimeUnit.SECONDS.toMillis(10); + long deadlineMs = Time.currentElapsedTime() + timeoutMs; + while (timeoutMs > 0 && (closed == null || closed.getState() != Watcher.Event.KeeperState.Closed)) { + WatchedEvent event = events.poll(10, TimeUnit.SECONDS); + if (event != null) { + closed = event; + } + timeoutMs = deadlineMs - Time.currentElapsedTime(); + } + assertNotNull(closed); + assertEquals(Watcher.Event.EventType.None, closed.getType()); + assertEquals(Watcher.Event.KeeperState.Closed, closed.getState()); + } + + @Test + public void testBuildClient() throws Exception { + BlockingQueue events = new LinkedBlockingQueue<>(); + ZooKeeper zk = new ZooKeeperBuilder(hostPort, 1000) + .withDefaultWatcher(events::offer) + .build(); + testClient(events, zk); + } + + @Test + public void testBuildAdminClient() throws Exception { + BlockingQueue events = new LinkedBlockingQueue<>(); + ZooKeeper zk = new ZooKeeperBuilder(hostPort, 1000) + .withDefaultWatcher(events::offer) + .buildAdmin(); + testClient(events, zk); + } +}