ZOOKEEPER-4697: Add Builder to construct ZooKeeper and ZooKeeperAdmin (#2001)

* ZOOKEEPER-4697: Add Builder to construct ZooKeeper and its derivations

Currently, there are 10 constructor variants for `ZooKeeper` and 4 for
`ZooKeeperAdmin`. It is enough for us to resort to a builder.

The `build` method throws `IOException` to make it a drop-in replacement
of existing constructors of `ZooKeeper`.

This pr also unify body of `ZooKeeper` constructor to one. Previously,
there are diverged to two. One has `sessionId` and `sessionPasswd`, and
another doesn't have. This pr uses `sessionId == 0` to differentiate the
two as it is used in server side to differentiate session create and
reconnect.

* Restrict Builder to only ZooKeeper and ZooKeeperAdmin

zookeeper-server/src/main/java/org/apache/zookeeper/ZooKeeper.java

@@ -46,6 +46,8 @@ import org.apache.zookeeper.client.ConnectStringParser;
 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 class ZooKeeper implements AutoCloseable {
      *             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 class ZooKeeper implements AutoCloseable {
         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 class ZooKeeper implements AutoCloseable {
         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 class ZooKeeper implements AutoCloseable {
         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 @@ public class ZooKeeper implements AutoCloseable {
         ZKClientConfig clientConfig,
         Watcher defaultWatcher,
         ClientCnxnSocket clientCnxnSocket,
+        long sessionId,
+        byte[] sessionPasswd,
         boolean canBeReadOnly
     ) throws IOException {
         return new ClientCnxn(
@@ -671,6 +671,8 @@ public class ZooKeeper implements AutoCloseable {
             clientConfig,
             defaultWatcher,
             clientCnxnSocket,
+            sessionId,
+            sessionPasswd,
             canBeReadOnly);
     }
 
@@ -731,7 +733,10 @@ public class ZooKeeper implements AutoCloseable {
         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 class ZooKeeper implements AutoCloseable {
         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 class ZooKeeper implements AutoCloseable {
         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 class ZooKeeper implements AutoCloseable {
         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 class ZooKeeper implements AutoCloseable {
         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 ? "<null>" : "<hidden>"));
+        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.
+     *
+     * <p>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.
+     *
+     * <p>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 ? "<null>" : "<hidden>"));
+        }
+
+        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 class ZooKeeper implements AutoCloseable {
             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 class ZooKeeper implements AutoCloseable {
         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

zookeeper-server/src/main/java/org/apache/zookeeper/admin/ZooKeeperAdmin.java

@@ -27,6 +27,7 @@ import org.apache.zookeeper.Watcher;
 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.

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<Collection<InetSocketAddress>, 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<Collection<InetSocketAddress>, 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());
+    }
+}

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<Collection<InetSocketAddress>, 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<Collection<InetSocketAddress>, 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<Collection<InetSocketAddress>, 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;
+    }
+}

zookeeper-server/src/test/java/org/apache/zookeeper/ClientCnxnSocketFragilityTest.java

@@ -286,6 +286,8 @@ public class ClientCnxnSocketFragilityTest extends QuorumPeerTestBase {
             ZKClientConfig zkClientConfig,
             Watcher defaultWatcher,
             ClientCnxnSocket clientCnxnSocket,
+            long sessionId,
+            byte[] sessionPasswd,
             boolean canBeReadOnly
         ) throws IOException {
             super(
@@ -295,6 +297,8 @@ public class ClientCnxnSocketFragilityTest extends QuorumPeerTestBase {
                 zkClientConfig,
                 defaultWatcher,
                 clientCnxnSocket,
+                sessionId,
+                sessionPasswd,
                 canBeReadOnly);
         }
 
@@ -351,6 +355,7 @@ public class ClientCnxnSocketFragilityTest extends QuorumPeerTestBase {
             return cnxn.getState().isAlive();
         }
 
+        @Override
         ClientCnxn createConnection(
             String chrootPath,
             HostProvider hostProvider,
@@ -358,6 +363,8 @@ public class ClientCnxnSocketFragilityTest extends QuorumPeerTestBase {
             ZKClientConfig clientConfig,
             Watcher defaultWatcher,
             ClientCnxnSocket clientCnxnSocket,
+            long sessionId,
+            byte[] sessionPasswd,
             boolean canBeReadOnly
         ) throws IOException {
             assertTrue(clientCnxnSocket instanceof FragileClientCnxnSocketNIO);
@@ -369,6 +376,8 @@ public class ClientCnxnSocketFragilityTest extends QuorumPeerTestBase {
                 clientConfig,
                 defaultWatcher,
                 clientCnxnSocket,
+                sessionId,
+                sessionPasswd,
                 canBeReadOnly);
             return ClientCnxnSocketFragilityTest.this.cnxn;
         }

zookeeper-server/src/test/java/org/apache/zookeeper/ClientRequestTimeoutTest.java

@@ -119,6 +119,8 @@ public class ClientRequestTimeoutTest extends QuorumPeerTestBase {
             ZKClientConfig clientConfig,
             Watcher defaultWatcher,
             ClientCnxnSocket clientCnxnSocket,
+            long sessionId,
+            byte[] sessionPasswd,
             boolean canBeReadOnly
         ) throws IOException {
             super(
@@ -128,6 +130,8 @@ public class ClientRequestTimeoutTest extends QuorumPeerTestBase {
                 clientConfig,
                 defaultWatcher,
                 clientCnxnSocket,
+                sessionId,
+                sessionPasswd,
                 canBeReadOnly);
         }
 
@@ -157,6 +161,8 @@ public class ClientRequestTimeoutTest extends QuorumPeerTestBase {
             ZKClientConfig clientConfig,
             Watcher defaultWatcher,
             ClientCnxnSocket clientCnxnSocket,
+            long sessionId,
+            byte[] sessionPasswd,
             boolean canBeReadOnly
         ) throws IOException {
             return new CustomClientCnxn(
@@ -166,6 +172,8 @@ public class ClientRequestTimeoutTest extends QuorumPeerTestBase {
                 clientConfig,
                 defaultWatcher,
                 clientCnxnSocket,
+                sessionId,
+                sessionPasswd,
                 canBeReadOnly);
         }
 

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<WatchedEvent> 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<WatchedEvent> events = new LinkedBlockingQueue<>();
+        ZooKeeper zk = new ZooKeeperBuilder(hostPort, 1000)
+            .withDefaultWatcher(events::offer)
+            .build();
+        testClient(events, zk);
+    }
+
+    @Test
+    public void testBuildAdminClient() throws Exception {
+        BlockingQueue<WatchedEvent> events = new LinkedBlockingQueue<>();
+        ZooKeeper zk = new ZooKeeperBuilder(hostPort, 1000)
+            .withDefaultWatcher(events::offer)
+            .buildAdmin();
+        testClient(events, zk);
+    }
+}