ZOOKEEPER-837: Eliminate cycle dependency between ClientCnxn and ZooKeeper

1. Extract ZKWatchManager to single file
2. Move ZKWatchManager instance to ClientCnxn in order to eliminate cycle dependency
3. let `ZooKeeper` syncs a copy of default watcher, in order to reduce dependencies to `getWatchManager()`

Author: tison <wander4096@gmail.com>

Reviewers: Enrico Olivelli <eolivelli@apache.org>, Andor Molnar <andor@apache.org>

Closes #1095 from TisonKun/ZOOKEEPER-837

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

@@ -165,13 +165,11 @@ public class ClientCnxn {
 
     private final int sessionTimeout;
 
-    private final ZooKeeper zooKeeper;
-
-    private final ClientWatchManager watcher;
+    private final ZKWatchManager watchManager;
 
     private long sessionId;
 
-    private byte[] sessionPasswd = new byte[16];
+    private byte[] sessionPasswd;
 
     /**
      * If true, the connection is allowed to go to r-o mode. This field's value
@@ -224,6 +222,10 @@ public class ClientCnxn {
      */
     private long requestTimeout;
 
+    ZKWatchManager getWatcherManager() {
+        return watchManager;
+    }
+
     public long getSessionId() {
         return sessionId;
     }
@@ -362,35 +364,29 @@ public class ClientCnxn {
      * established until needed. The start() instance method must be called
      * subsequent to construction.
      *
-     * @param chrootPath - the chroot of this client. Should be removed from this Class in ZOOKEEPER-838
-     * @param hostProvider
-     *                the list of ZooKeeper servers to connect to
-     * @param sessionTimeout
-     *                the timeout for connections.
-     * @param zooKeeper
-     *                the zookeeper object that this connection is related to.
-     * @param watcher watcher for this connection
-     * @param clientCnxnSocket
-     *                the socket implementation used (e.g. NIO/Netty)
-     * @param canBeReadOnly
-     *                whether the connection is allowed to go to read-only
-     *                mode in case of partitioning
-     * @throws IOException
+     * @param chrootPath the chroot of this client. Should be removed from this Class in ZOOKEEPER-838
+     * @param hostProvider the list of ZooKeeper servers to connect to
+     * @param sessionTimeout the timeout for connections.
+     * @param clientConfig the client configuration.
+     * @param defaultWatcher default watcher for this connection
+     * @param clientCnxnSocket the socket implementation used (e.g. NIO/Netty)
+     * @param canBeReadOnly whether the connection is allowed to go to read-only mode in case of partitioning
      */
     public ClientCnxn(
         String chrootPath,
         HostProvider hostProvider,
         int sessionTimeout,
-        ZooKeeper zooKeeper,
-        ClientWatchManager watcher,
+        ZKClientConfig clientConfig,
+        Watcher defaultWatcher,
         ClientCnxnSocket clientCnxnSocket,
-        boolean canBeReadOnly) throws IOException {
+        boolean canBeReadOnly
+    ) throws IOException {
         this(
             chrootPath,
             hostProvider,
             sessionTimeout,
-            zooKeeper,
-            watcher,
+            clientConfig,
+            defaultWatcher,
             clientCnxnSocket,
             0,
             new byte[16],
@@ -402,48 +398,45 @@ public class ClientCnxn {
      * established until needed. The start() instance method must be called
      * subsequent to construction.
      *
-     * @param chrootPath - the chroot of this client. Should be removed from this Class in ZOOKEEPER-838
-     * @param hostProvider
-     *                the list of ZooKeeper servers to connect to
-     * @param sessionTimeout
-     *                the timeout for connections.
-     * @param zooKeeper
-     *                the zookeeper object that this connection is related to.
-     * @param watcher watcher for this connection
-     * @param clientCnxnSocket
-     *                the socket implementation used (e.g. NIO/Netty)
+     * @param chrootPath the chroot of this client. Should be removed from this Class in ZOOKEEPER-838
+     * @param hostProvider the list of ZooKeeper servers to connect to
+     * @param sessionTimeout the timeout for connections.
+     * @param clientConfig the client configuration.
+     * @param defaultWatcher default watcher for this connection
+     * @param clientCnxnSocket the socket implementation used (e.g. NIO/Netty)
      * @param sessionId session id if re-establishing session
      * @param sessionPasswd session passwd if re-establishing session
-     * @param canBeReadOnly
-     *                whether the connection is allowed to go to read-only
-     *                mode in case of partitioning
+     * @param canBeReadOnly whether the connection is allowed to go to read-only mode in case of partitioning
      * @throws IOException in cases of broken network
      */
     public ClientCnxn(
         String chrootPath,
         HostProvider hostProvider,
         int sessionTimeout,
-        ZooKeeper zooKeeper,
-        ClientWatchManager watcher,
+        ZKClientConfig clientConfig,
+        Watcher defaultWatcher,
         ClientCnxnSocket clientCnxnSocket,
         long sessionId,
         byte[] sessionPasswd,
-        boolean canBeReadOnly) throws IOException {
-        this.zooKeeper = zooKeeper;
-        this.watcher = watcher;
+        boolean canBeReadOnly
+    ) throws IOException {
+        this.chrootPath = chrootPath;
+        this.hostProvider = hostProvider;
+        this.sessionTimeout = sessionTimeout;
+        this.clientConfig = clientConfig;
         this.sessionId = sessionId;
         this.sessionPasswd = sessionPasswd;
-        this.sessionTimeout = sessionTimeout;
-        this.hostProvider = hostProvider;
-        this.chrootPath = chrootPath;
+        this.readOnly = canBeReadOnly;
+
+        this.watchManager = new ZKWatchManager(
+                clientConfig.getBoolean(ZKClientConfig.DISABLE_AUTO_WATCH_RESET),
+                defaultWatcher);
 
-        connectTimeout = sessionTimeout / hostProvider.size();
-        readTimeout = sessionTimeout * 2 / 3;
-        readOnly = canBeReadOnly;
+        this.connectTimeout = sessionTimeout / hostProvider.size();
+        this.readTimeout = sessionTimeout * 2 / 3;
 
-        sendThread = new SendThread(clientCnxnSocket);
-        eventThread = new EventThread();
-        this.clientConfig = zooKeeper.getClientConfig();
+        this.sendThread = new SendThread(clientCnxnSocket);
+        this.eventThread = new EventThread();
         initRequestTimeout();
     }
 
@@ -506,10 +499,9 @@ public class ClientCnxn {
             final Set<Watcher> watchers;
             if (materializedWatchers == null) {
                 // materialize the watchers based on the event
-                watchers = watcher.materialize(event.getState(), event.getType(), event.getPath());
+                watchers = watchManager.materialize(event.getState(), event.getType(), event.getPath());
             } else {
-                watchers = new HashSet<Watcher>();
-                watchers.addAll(materializedWatchers);
+                watchers = new HashSet<>(materializedWatchers);
             }
             WatcherSetEventPair pair = new WatcherSetEventPair(watchers, event);
             // queue the pair (watch set & event) for later processing
@@ -1007,14 +999,12 @@ public class ClientCnxn {
             ConnectRequest conReq = new ConnectRequest(0, lastZxid, sessionTimeout, sessId, sessionPasswd);
             // We add backwards since we are pushing into the front
             // Only send if there's a pending watch
-            // TODO: here we have the only remaining use of zooKeeper in
-            // this class. It's to be eliminated!
             if (!clientConfig.getBoolean(ZKClientConfig.DISABLE_AUTO_WATCH_RESET)) {
-                List<String> dataWatches = zooKeeper.getDataWatches();
-                List<String> existWatches = zooKeeper.getExistWatches();
-                List<String> childWatches = zooKeeper.getChildWatches();
-                List<String> persistentWatches = zooKeeper.getPersistentWatches();
-                List<String> persistentRecursiveWatches = zooKeeper.getPersistentRecursiveWatches();
+                List<String> dataWatches = watchManager.getDataWatchList();
+                List<String> existWatches = watchManager.getExistWatchList();
+                List<String> childWatches = watchManager.getChildWatchList();
+                List<String> persistentWatches = watchManager.getPersistentWatchList();
+                List<String> persistentRecursiveWatches = watchManager.getPersistentRecursiveWatchList();
                 if (!dataWatches.isEmpty() || !existWatches.isEmpty() || !childWatches.isEmpty()
                         || !persistentWatches.isEmpty() || !persistentRecursiveWatches.isEmpty()) {
                     Iterator<String> dataWatchesIter = prependChroot(dataWatches).iterator();

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

@@ -22,7 +22,6 @@ import java.util.Map;
 import java.util.Set;
 import org.apache.zookeeper.Watcher.Event.EventType;
 import org.apache.zookeeper.Watcher.WatcherType;
-import org.apache.zookeeper.ZooKeeper.ZKWatchManager;
 
 /**
  * Handles the special case of removing watches which has registered for a

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

@@ -0,0 +1,455 @@
+/*
+ * 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;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import org.apache.zookeeper.server.watch.PathParentIterator;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Manage watchers and handle events generated by the {@link ClientCnxn} object.
+ *
+ * This class is intended to be packaged-private so that it doesn't serve
+ * as part of ZooKeeper client API.
+ */
+class ZKWatchManager implements ClientWatchManager {
+
+    private static final Logger LOG = LoggerFactory.getLogger(ZKWatchManager.class);
+
+    private final Map<String, Set<Watcher>> dataWatches = new HashMap<>();
+    private final Map<String, Set<Watcher>> existWatches = new HashMap<>();
+    private final Map<String, Set<Watcher>> childWatches = new HashMap<>();
+    private final Map<String, Set<Watcher>> persistentWatches = new HashMap<>();
+    private final Map<String, Set<Watcher>> persistentRecursiveWatches = new HashMap<>();
+    private final boolean disableAutoWatchReset;
+
+    private volatile Watcher defaultWatcher;
+
+    ZKWatchManager(boolean disableAutoWatchReset, Watcher defaultWatcher) {
+        this.disableAutoWatchReset = disableAutoWatchReset;
+        this.defaultWatcher = defaultWatcher;
+    }
+
+    void setDefaultWatcher(Watcher defaultWatcher) {
+        this.defaultWatcher = defaultWatcher;
+    }
+
+    Watcher getDefaultWatcher() {
+        return defaultWatcher;
+    }
+
+    List<String> getDataWatchList() {
+        synchronized (dataWatches) {
+            return new ArrayList<>(dataWatches.keySet());
+        }
+    }
+
+    List<String> getChildWatchList() {
+        synchronized (childWatches) {
+            return new ArrayList<>(childWatches.keySet());
+        }
+    }
+
+    List<String> getExistWatchList() {
+        synchronized (existWatches) {
+            return new ArrayList<>(existWatches.keySet());
+        }
+    }
+
+    List<String> getPersistentWatchList() {
+        synchronized (persistentWatches) {
+            return new ArrayList<>(persistentWatches.keySet());
+        }
+    }
+
+    List<String> getPersistentRecursiveWatchList() {
+        synchronized (persistentRecursiveWatches) {
+            return new ArrayList<>(persistentRecursiveWatches.keySet());
+        }
+    }
+
+    Map<String, Set<Watcher>> getDataWatches() {
+        return dataWatches;
+    }
+
+    Map<String, Set<Watcher>> getExistWatches() {
+        return existWatches;
+    }
+
+    Map<String, Set<Watcher>> getChildWatches() {
+        return childWatches;
+    }
+
+    Map<String, Set<Watcher>> getPersistentWatches() {
+        return persistentWatches;
+    }
+
+    Map<String, Set<Watcher>> getPersistentRecursiveWatches() {
+        return persistentRecursiveWatches;
+    }
+
+    private void addTo(Set<Watcher> from, Set<Watcher> to) {
+        if (from != null) {
+            to.addAll(from);
+        }
+    }
+
+    public Map<Watcher.Event.EventType, Set<Watcher>> removeWatcher(
+        String clientPath,
+        Watcher watcher,
+        Watcher.WatcherType watcherType,
+        boolean local,
+        int rc
+    ) throws KeeperException {
+        // Validate the provided znode path contains the given watcher of
+        // watcherType
+        containsWatcher(clientPath, watcher, watcherType);
+
+        Map<Watcher.Event.EventType, Set<Watcher>> removedWatchers = new HashMap<>();
+        HashSet<Watcher> childWatchersToRem = new HashSet<>();
+        removedWatchers.put(Watcher.Event.EventType.ChildWatchRemoved, childWatchersToRem);
+        HashSet<Watcher> dataWatchersToRem = new HashSet<>();
+        removedWatchers.put(Watcher.Event.EventType.DataWatchRemoved, dataWatchersToRem);
+        HashSet<Watcher> persistentWatchersToRem = new HashSet<>();
+        removedWatchers.put(Watcher.Event.EventType.PersistentWatchRemoved, persistentWatchersToRem);
+        boolean removedWatcher = false;
+        switch (watcherType) {
+        case Children: {
+            synchronized (childWatches) {
+                removedWatcher = removeWatches(childWatches, watcher, clientPath, local, rc, childWatchersToRem);
+            }
+            break;
+        }
+        case Data: {
+            synchronized (dataWatches) {
+                removedWatcher = removeWatches(dataWatches, watcher, clientPath, local, rc, dataWatchersToRem);
+            }
+
+            synchronized (existWatches) {
+                boolean removedDataWatcher = removeWatches(existWatches, watcher, clientPath, local, rc, dataWatchersToRem);
+                removedWatcher |= removedDataWatcher;
+            }
+            break;
+        }
+        case Any: {
+            synchronized (childWatches) {
+                removedWatcher = removeWatches(childWatches, watcher, clientPath, local, rc, childWatchersToRem);
+            }
+
+            synchronized (dataWatches) {
+                boolean removedDataWatcher = removeWatches(dataWatches, watcher, clientPath, local, rc, dataWatchersToRem);
+                removedWatcher |= removedDataWatcher;
+            }
+
+            synchronized (existWatches) {
+                boolean removedDataWatcher = removeWatches(existWatches, watcher, clientPath, local, rc, dataWatchersToRem);
+                removedWatcher |= removedDataWatcher;
+            }
+
+            synchronized (persistentWatches) {
+                boolean removedPersistentWatcher = removeWatches(persistentWatches,
+                        watcher, clientPath, local, rc, persistentWatchersToRem);
+                removedWatcher |= removedPersistentWatcher;
+            }
+
+            synchronized (persistentRecursiveWatches) {
+                boolean removedPersistentRecursiveWatcher = removeWatches(persistentRecursiveWatches,
+                        watcher, clientPath, local, rc, persistentWatchersToRem);
+                removedWatcher |= removedPersistentRecursiveWatcher;
+            }
+        }
+        }
+        // Watcher function doesn't exists for the specified params
+        if (!removedWatcher) {
+            throw new KeeperException.NoWatcherException(clientPath);
+        }
+        return removedWatchers;
+    }
+
+    private boolean contains(String path, Watcher watcherObj, Map<String, Set<Watcher>> pathVsWatchers) {
+        boolean watcherExists = true;
+        if (pathVsWatchers == null || pathVsWatchers.size() == 0) {
+            watcherExists = false;
+        } else {
+            Set<Watcher> watchers = pathVsWatchers.get(path);
+            if (watchers == null) {
+                watcherExists = false;
+            } else if (watcherObj == null) {
+                watcherExists = watchers.size() > 0;
+            } else {
+                watcherExists = watchers.contains(watcherObj);
+            }
+        }
+        return watcherExists;
+    }
+
+    /**
+     * Validate the provided znode path contains the given watcher and
+     * watcherType
+     *
+     * @param path
+     *            - client path
+     * @param watcher
+     *            - watcher object reference
+     * @param watcherType
+     *            - type of the watcher
+     * @throws KeeperException.NoWatcherException
+     */
+    void containsWatcher(String path, Watcher watcher, Watcher.WatcherType watcherType) throws
+            KeeperException.NoWatcherException {
+        boolean containsWatcher = false;
+        switch (watcherType) {
+        case Children: {
+            synchronized (childWatches) {
+                containsWatcher = contains(path, watcher, childWatches);
+            }
+
+            synchronized (persistentWatches) {
+                boolean contains_temp = contains(path, watcher,
+                        persistentWatches);
+                containsWatcher |= contains_temp;
+            }
+
+            synchronized (persistentRecursiveWatches) {
+                boolean contains_temp = contains(path, watcher,
+                        persistentRecursiveWatches);
+                containsWatcher |= contains_temp;
+            }
+            break;
+        }
+        case Data: {
+            synchronized (dataWatches) {
+                containsWatcher = contains(path, watcher, dataWatches);
+            }
+
+            synchronized (existWatches) {
+                boolean contains_temp = contains(path, watcher, existWatches);
+                containsWatcher |= contains_temp;
+            }
+
+            synchronized (persistentWatches) {
+                boolean contains_temp = contains(path, watcher,
+                        persistentWatches);
+                containsWatcher |= contains_temp;
+            }
+
+            synchronized (persistentRecursiveWatches) {
+                boolean contains_temp = contains(path, watcher,
+                        persistentRecursiveWatches);
+                containsWatcher |= contains_temp;
+            }
+            break;
+        }
+        case Any: {
+            synchronized (childWatches) {
+                containsWatcher = contains(path, watcher, childWatches);
+            }
+
+            synchronized (dataWatches) {
+                boolean contains_temp = contains(path, watcher, dataWatches);
+                containsWatcher |= contains_temp;
+            }
+
+            synchronized (existWatches) {
+                boolean contains_temp = contains(path, watcher, existWatches);
+                containsWatcher |= contains_temp;
+            }
+
+            synchronized (persistentWatches) {
+                boolean contains_temp = contains(path, watcher,
+                        persistentWatches);
+                containsWatcher |= contains_temp;
+            }
+
+            synchronized (persistentRecursiveWatches) {
+                boolean contains_temp = contains(path, watcher,
+                        persistentRecursiveWatches);
+                containsWatcher |= contains_temp;
+            }
+        }
+        }
+        // Watcher function doesn't exists for the specified params
+        if (!containsWatcher) {
+            throw new KeeperException.NoWatcherException(path);
+        }
+    }
+
+    protected boolean removeWatches(
+        Map<String, Set<Watcher>> pathVsWatcher,
+        Watcher watcher,
+        String path,
+        boolean local,
+        int rc,
+        Set<Watcher> removedWatchers) throws KeeperException {
+        if (!local && rc != KeeperException.Code.OK.intValue()) {
+            throw KeeperException.create(KeeperException.Code.get(rc), path);
+        }
+        boolean success = false;
+        // When local flag is true, remove watchers for the given path
+        // irrespective of rc. Otherwise shouldn't remove watchers locally
+        // when sees failure from server.
+        if (rc == KeeperException.Code.OK.intValue() || (local && rc != KeeperException.Code.OK.intValue())) {
+            // Remove all the watchers for the given path
+            if (watcher == null) {
+                Set<Watcher> pathWatchers = pathVsWatcher.remove(path);
+                if (pathWatchers != null) {
+                    // found path watchers
+                    removedWatchers.addAll(pathWatchers);
+                    success = true;
+                }
+            } else {
+                Set<Watcher> watchers = pathVsWatcher.get(path);
+                if (watchers != null) {
+                    if (watchers.remove(watcher)) {
+                        // found path watcher
+                        removedWatchers.add(watcher);
+                        // cleanup <path vs watchlist>
+                        if (watchers.size() <= 0) {
+                            pathVsWatcher.remove(path);
+                        }
+                        success = true;
+                    }
+                }
+            }
+        }
+        return success;
+    }
+
+    /* (non-Javadoc)
+     * @see org.apache.zookeeper.ClientWatchManager#materialize(Event.KeeperState,
+     *                                                        Event.EventType, java.lang.String)
+     */
+    @Override
+    public Set<Watcher> materialize(
+        Watcher.Event.KeeperState state,
+        Watcher.Event.EventType type,
+        String clientPath
+    ) {
+        final Set<Watcher> result = new HashSet<>();
+
+        switch (type) {
+        case None:
+            if (defaultWatcher != null) {
+                result.add(defaultWatcher);
+            }
+
+            boolean clear = disableAutoWatchReset && state != Watcher.Event.KeeperState.SyncConnected;
+            synchronized (dataWatches) {
+                for (Set<Watcher> ws : dataWatches.values()) {
+                    result.addAll(ws);
+                }
+                if (clear) {
+                    dataWatches.clear();
+                }
+            }
+
+            synchronized (existWatches) {
+                for (Set<Watcher> ws : existWatches.values()) {
+                    result.addAll(ws);
+                }
+                if (clear) {
+                    existWatches.clear();
+                }
+            }
+
+            synchronized (childWatches) {
+                for (Set<Watcher> ws : childWatches.values()) {
+                    result.addAll(ws);
+                }
+                if (clear) {
+                    childWatches.clear();
+                }
+            }
+
+            synchronized (persistentWatches) {
+                for (Set<Watcher> ws: persistentWatches.values()) {
+                    result.addAll(ws);
+                }
+            }
+
+            synchronized (persistentRecursiveWatches) {
+                for (Set<Watcher> ws: persistentRecursiveWatches.values()) {
+                    result.addAll(ws);
+                }
+            }
+
+            return result;
+        case NodeDataChanged:
+        case NodeCreated:
+            synchronized (dataWatches) {
+                addTo(dataWatches.remove(clientPath), result);
+            }
+            synchronized (existWatches) {
+                addTo(existWatches.remove(clientPath), result);
+            }
+            addPersistentWatches(clientPath, result);
+            break;
+        case NodeChildrenChanged:
+            synchronized (childWatches) {
+                addTo(childWatches.remove(clientPath), result);
+            }
+            addPersistentWatches(clientPath, result);
+            break;
+        case NodeDeleted:
+            synchronized (dataWatches) {
+                addTo(dataWatches.remove(clientPath), result);
+            }
+            // TODO This shouldn't be needed, but just in case
+            synchronized (existWatches) {
+                Set<Watcher> list = existWatches.remove(clientPath);
+                if (list != null) {
+                    addTo(list, result);
+                    LOG.warn("We are triggering an exists watch for delete! Shouldn't happen!");
+                }
+            }
+            synchronized (childWatches) {
+                addTo(childWatches.remove(clientPath), result);
+            }
+            addPersistentWatches(clientPath, result);
+            break;
+        default:
+            String errorMsg = String.format(
+                "Unhandled watch event type %s with state %s on path %s",
+                type,
+                state,
+                clientPath);
+            LOG.error(errorMsg);
+            throw new RuntimeException(errorMsg);
+        }
+
+        return result;
+    }
+
+    private void addPersistentWatches(String clientPath, Set<Watcher> result) {
+        synchronized (persistentWatches) {
+            addTo(persistentWatches.get(clientPath), result);
+        }
+        synchronized (persistentRecursiveWatches) {
+            for (String path : PathParentIterator.forAll(clientPath).asIterable()) {
+                addTo(persistentRecursiveWatches.get(path), result);
+            }
+        }
+    }
+}

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

@@ -24,7 +24,6 @@ import java.net.InetSocketAddress;
 import java.net.SocketAddress;
 import java.util.ArrayList;
 import java.util.Collection;
-import java.util.HashMap;
 import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
@@ -40,10 +39,7 @@ import org.apache.zookeeper.AsyncCallback.MultiCallback;
 import org.apache.zookeeper.AsyncCallback.StatCallback;
 import org.apache.zookeeper.AsyncCallback.StringCallback;
 import org.apache.zookeeper.AsyncCallback.VoidCallback;
-import org.apache.zookeeper.KeeperException.Code;
-import org.apache.zookeeper.KeeperException.NoWatcherException;
 import org.apache.zookeeper.OpResult.ErrorResult;
-import org.apache.zookeeper.Watcher.Event.EventType;
 import org.apache.zookeeper.Watcher.WatcherType;
 import org.apache.zookeeper.client.ConnectStringParser;
 import org.apache.zookeeper.client.HostProvider;
@@ -85,7 +81,6 @@ import org.apache.zookeeper.proto.SyncRequest;
 import org.apache.zookeeper.proto.SyncResponse;
 import org.apache.zookeeper.server.DataTree;
 import org.apache.zookeeper.server.EphemeralType;
-import org.apache.zookeeper.server.watch.PathParentIterator;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -231,8 +226,6 @@ public class ZooKeeper implements AutoCloseable {
         return cnxn.zooKeeperSaslClient;
     }
 
-    protected final ZKWatchManager watchManager;
-
     private final ZKClientConfig clientConfig;
 
     public ZKClientConfig getClientConfig() {
@@ -240,407 +233,37 @@ public class ZooKeeper implements AutoCloseable {
     }
 
     protected List<String> getDataWatches() {
-        synchronized (watchManager.dataWatches) {
-            List<String> rc = new ArrayList<String>(watchManager.dataWatches.keySet());
-            return rc;
-        }
+        return getWatchManager().getDataWatchList();
     }
+
     protected List<String> getExistWatches() {
-        synchronized (watchManager.existWatches) {
-            List<String> rc = new ArrayList<String>(watchManager.existWatches.keySet());
-            return rc;
-        }
+        return getWatchManager().getExistWatchList();
     }
+
     protected List<String> getChildWatches() {
-        synchronized (watchManager.childWatches) {
-            List<String> rc = new ArrayList<String>(watchManager.childWatches.keySet());
-            return rc;
-        }
+        return getWatchManager().getChildWatchList();
     }
+
     protected List<String> getPersistentWatches() {
-        synchronized (watchManager.persistentWatches) {
-            List<String> rc = new ArrayList<String>(watchManager.persistentWatches.keySet());
-            return rc;
-        }
+        return getWatchManager().getPersistentWatchList();
     }
+
     protected List<String> getPersistentRecursiveWatches() {
-        synchronized (watchManager.persistentRecursiveWatches) {
-            List<String> rc = new ArrayList<String>(watchManager.persistentRecursiveWatches.keySet());
-            return rc;
-        }
+        return getWatchManager().getPersistentRecursiveWatchList();
     }
 
-    /**
-     * Manage watchers and handle events generated by the ClientCnxn object.
-     *
-     * We are implementing this as a nested class of ZooKeeper so that
-     * the public methods will not be exposed as part of the ZooKeeper client
-     * API.
-     */
-    static class ZKWatchManager implements ClientWatchManager {
-
-        private final Map<String, Set<Watcher>> dataWatches = new HashMap<String, Set<Watcher>>();
-        private final Map<String, Set<Watcher>> existWatches = new HashMap<String, Set<Watcher>>();
-        private final Map<String, Set<Watcher>> childWatches = new HashMap<String, Set<Watcher>>();
-        private final Map<String, Set<Watcher>> persistentWatches = new HashMap<String, Set<Watcher>>();
-        private final Map<String, Set<Watcher>> persistentRecursiveWatches = new HashMap<String, Set<Watcher>>();
-        private boolean disableAutoWatchReset;
-
-        ZKWatchManager(boolean disableAutoWatchReset) {
-            this.disableAutoWatchReset = disableAutoWatchReset;
-        }
-
-        protected volatile Watcher defaultWatcher;
-
-        private void addTo(Set<Watcher> from, Set<Watcher> to) {
-            if (from != null) {
-                to.addAll(from);
-            }
-        }
-
-        public Map<EventType, Set<Watcher>> removeWatcher(
-            String clientPath,
-            Watcher watcher,
-            WatcherType watcherType,
-            boolean local,
-            int rc) throws KeeperException {
-            // Validate the provided znode path contains the given watcher of
-            // watcherType
-            containsWatcher(clientPath, watcher, watcherType);
-
-            Map<EventType, Set<Watcher>> removedWatchers = new HashMap<>();
-            HashSet<Watcher> childWatchersToRem = new HashSet<>();
-            removedWatchers.put(EventType.ChildWatchRemoved, childWatchersToRem);
-            HashSet<Watcher> dataWatchersToRem = new HashSet<>();
-            removedWatchers.put(EventType.DataWatchRemoved, dataWatchersToRem);
-            HashSet<Watcher> persistentWatchersToRem = new HashSet<>();
-            removedWatchers.put(EventType.PersistentWatchRemoved, persistentWatchersToRem);
-            boolean removedWatcher = false;
-            switch (watcherType) {
-            case Children: {
-                synchronized (childWatches) {
-                    removedWatcher = removeWatches(childWatches, watcher, clientPath, local, rc, childWatchersToRem);
-                }
-                break;
-            }
-            case Data: {
-                synchronized (dataWatches) {
-                    removedWatcher = removeWatches(dataWatches, watcher, clientPath, local, rc, dataWatchersToRem);
-                }
-
-                synchronized (existWatches) {
-                    boolean removedDataWatcher = removeWatches(existWatches, watcher, clientPath, local, rc, dataWatchersToRem);
-                    removedWatcher |= removedDataWatcher;
-                }
-                break;
-            }
-            case Any: {
-                synchronized (childWatches) {
-                    removedWatcher = removeWatches(childWatches, watcher, clientPath, local, rc, childWatchersToRem);
-                }
-
-                synchronized (dataWatches) {
-                    boolean removedDataWatcher = removeWatches(dataWatches, watcher, clientPath, local, rc, dataWatchersToRem);
-                    removedWatcher |= removedDataWatcher;
-                }
-
-                synchronized (existWatches) {
-                    boolean removedDataWatcher = removeWatches(existWatches, watcher, clientPath, local, rc, dataWatchersToRem);
-                    removedWatcher |= removedDataWatcher;
-                }
-
-                synchronized (persistentWatches) {
-                    boolean removedPersistentWatcher = removeWatches(persistentWatches,
-                            watcher, clientPath, local, rc, persistentWatchersToRem);
-                    removedWatcher |= removedPersistentWatcher;
-                }
-
-                synchronized (persistentRecursiveWatches) {
-                    boolean removedPersistentRecursiveWatcher = removeWatches(persistentRecursiveWatches,
-                            watcher, clientPath, local, rc, persistentWatchersToRem);
-                    removedWatcher |= removedPersistentRecursiveWatcher;
-                }
-            }
-            }
-            // Watcher function doesn't exists for the specified params
-            if (!removedWatcher) {
-                throw new KeeperException.NoWatcherException(clientPath);
-            }
-            return removedWatchers;
-        }
-
-        private boolean contains(String path, Watcher watcherObj, Map<String, Set<Watcher>> pathVsWatchers) {
-            boolean watcherExists = true;
-            if (pathVsWatchers == null || pathVsWatchers.size() == 0) {
-                watcherExists = false;
-            } else {
-                Set<Watcher> watchers = pathVsWatchers.get(path);
-                if (watchers == null) {
-                    watcherExists = false;
-                } else if (watcherObj == null) {
-                    watcherExists = watchers.size() > 0;
-                } else {
-                    watcherExists = watchers.contains(watcherObj);
-                }
-            }
-            return watcherExists;
-        }
-
-        /**
-         * Validate the provided znode path contains the given watcher and
-         * watcherType
-         *
-         * @param path
-         *            - client path
-         * @param watcher
-         *            - watcher object reference
-         * @param watcherType
-         *            - type of the watcher
-         * @throws NoWatcherException
-         */
-        void containsWatcher(String path, Watcher watcher, WatcherType watcherType) throws NoWatcherException {
-            boolean containsWatcher = false;
-            switch (watcherType) {
-            case Children: {
-                synchronized (childWatches) {
-                    containsWatcher = contains(path, watcher, childWatches);
-                }
-
-                synchronized (persistentWatches) {
-                    boolean contains_temp = contains(path, watcher,
-                            persistentWatches);
-                    containsWatcher |= contains_temp;
-                }
-
-                synchronized (persistentRecursiveWatches) {
-                    boolean contains_temp = contains(path, watcher,
-                            persistentRecursiveWatches);
-                    containsWatcher |= contains_temp;
-                }
-                break;
-            }
-            case Data: {
-                synchronized (dataWatches) {
-                    containsWatcher = contains(path, watcher, dataWatches);
-                }
-
-                synchronized (existWatches) {
-                    boolean contains_temp = contains(path, watcher, existWatches);
-                    containsWatcher |= contains_temp;
-                }
-
-                synchronized (persistentWatches) {
-                    boolean contains_temp = contains(path, watcher,
-                            persistentWatches);
-                    containsWatcher |= contains_temp;
-                }
-
-                synchronized (persistentRecursiveWatches) {
-                    boolean contains_temp = contains(path, watcher,
-                            persistentRecursiveWatches);
-                    containsWatcher |= contains_temp;
-                }
-                break;
-            }
-            case Any: {
-                synchronized (childWatches) {
-                    containsWatcher = contains(path, watcher, childWatches);
-                }
-
-                synchronized (dataWatches) {
-                    boolean contains_temp = contains(path, watcher, dataWatches);
-                    containsWatcher |= contains_temp;
-                }
-
-                synchronized (existWatches) {
-                    boolean contains_temp = contains(path, watcher, existWatches);
-                    containsWatcher |= contains_temp;
-                }
-
-                synchronized (persistentWatches) {
-                    boolean contains_temp = contains(path, watcher,
-                            persistentWatches);
-                    containsWatcher |= contains_temp;
-                }
-
-                synchronized (persistentRecursiveWatches) {
-                    boolean contains_temp = contains(path, watcher,
-                            persistentRecursiveWatches);
-                    containsWatcher |= contains_temp;
-                }
-            }
-            }
-            // Watcher function doesn't exists for the specified params
-            if (!containsWatcher) {
-                throw new KeeperException.NoWatcherException(path);
-            }
-        }
-
-        protected boolean removeWatches(
-            Map<String, Set<Watcher>> pathVsWatcher,
-            Watcher watcher,
-            String path,
-            boolean local,
-            int rc,
-            Set<Watcher> removedWatchers) throws KeeperException {
-            if (!local && rc != Code.OK.intValue()) {
-                throw KeeperException.create(KeeperException.Code.get(rc), path);
-            }
-            boolean success = false;
-            // When local flag is true, remove watchers for the given path
-            // irrespective of rc. Otherwise shouldn't remove watchers locally
-            // when sees failure from server.
-            if (rc == Code.OK.intValue() || (local && rc != Code.OK.intValue())) {
-                // Remove all the watchers for the given path
-                if (watcher == null) {
-                    Set<Watcher> pathWatchers = pathVsWatcher.remove(path);
-                    if (pathWatchers != null) {
-                        // found path watchers
-                        removedWatchers.addAll(pathWatchers);
-                        success = true;
-                    }
-                } else {
-                    Set<Watcher> watchers = pathVsWatcher.get(path);
-                    if (watchers != null) {
-                        if (watchers.remove(watcher)) {
-                            // found path watcher
-                            removedWatchers.add(watcher);
-                            // cleanup <path vs watchlist>
-                            if (watchers.size() <= 0) {
-                                pathVsWatcher.remove(path);
-                            }
-                            success = true;
-                        }
-                    }
-                }
-            }
-            return success;
-        }
-
-        /* (non-Javadoc)
-         * @see org.apache.zookeeper.ClientWatchManager#materialize(Event.KeeperState,
-         *                                                        Event.EventType, java.lang.String)
-         */
-        @Override
-        public Set<Watcher> materialize(
-            Watcher.Event.KeeperState state,
-            Watcher.Event.EventType type,
-            String clientPath
-        ) {
-            final Set<Watcher> result = new HashSet<>();
-
-            switch (type) {
-            case None:
-                if (defaultWatcher != null) {
-                    result.add(defaultWatcher);
-                }
-
-                boolean clear = disableAutoWatchReset && state != Watcher.Event.KeeperState.SyncConnected;
-                synchronized (dataWatches) {
-                    for (Set<Watcher> ws : dataWatches.values()) {
-                        result.addAll(ws);
-                    }
-                    if (clear) {
-                        dataWatches.clear();
-                    }
-                }
-
-                synchronized (existWatches) {
-                    for (Set<Watcher> ws : existWatches.values()) {
-                        result.addAll(ws);
-                    }
-                    if (clear) {
-                        existWatches.clear();
-                    }
-                }
-
-                synchronized (childWatches) {
-                    for (Set<Watcher> ws : childWatches.values()) {
-                        result.addAll(ws);
-                    }
-                    if (clear) {
-                        childWatches.clear();
-                    }
-                }
-
-                synchronized (persistentWatches) {
-                    for (Set<Watcher> ws: persistentWatches.values()) {
-                        result.addAll(ws);
-                    }
-                }
-
-                synchronized (persistentRecursiveWatches) {
-                    for (Set<Watcher> ws: persistentRecursiveWatches.values()) {
-                        result.addAll(ws);
-                    }
-                }
-
-                return result;
-            case NodeDataChanged:
-            case NodeCreated:
-                synchronized (dataWatches) {
-                    addTo(dataWatches.remove(clientPath), result);
-                }
-                synchronized (existWatches) {
-                    addTo(existWatches.remove(clientPath), result);
-                }
-                addPersistentWatches(clientPath, result);
-                break;
-            case NodeChildrenChanged:
-                synchronized (childWatches) {
-                    addTo(childWatches.remove(clientPath), result);
-                }
-                addPersistentWatches(clientPath, result);
-                break;
-            case NodeDeleted:
-                synchronized (dataWatches) {
-                    addTo(dataWatches.remove(clientPath), result);
-                }
-                // TODO This shouldn't be needed, but just in case
-                synchronized (existWatches) {
-                    Set<Watcher> list = existWatches.remove(clientPath);
-                    if (list != null) {
-                        addTo(list, result);
-                        LOG.warn("We are triggering an exists watch for delete! Shouldn't happen!");
-                    }
-                }
-                synchronized (childWatches) {
-                    addTo(childWatches.remove(clientPath), result);
-                }
-                addPersistentWatches(clientPath, result);
-                break;
-            default:
-                String errorMsg = String.format(
-                    "Unhandled watch event type %s with state %s on path %s",
-                    type,
-                    state,
-                    clientPath);
-                LOG.error(errorMsg);
-                throw new RuntimeException(errorMsg);
-            }
-
-            return result;
-        }
-
-        private void addPersistentWatches(String clientPath, Set<Watcher> result) {
-            synchronized (persistentWatches) {
-                addTo(persistentWatches.get(clientPath), result);
-            }
-            synchronized (persistentRecursiveWatches) {
-                for (String path : PathParentIterator.forAll(clientPath).asIterable()) {
-                    addTo(persistentRecursiveWatches.get(path), result);
-                }
-            }
-        }
+    ZKWatchManager getWatchManager() {
+        return cnxn.getWatcherManager();
     }
 
     /**
      * Register a watcher for a particular path.
      */
-    public abstract class WatchRegistration {
+    public abstract static class WatchRegistration {
 
         private Watcher watcher;
         private String clientPath;
+
         public WatchRegistration(Watcher watcher, String clientPath) {
             this.watcher = watcher;
             this.clientPath = clientPath;
@@ -689,7 +312,7 @@ public class ZooKeeper implements AutoCloseable {
 
         @Override
         protected Map<String, Set<Watcher>> getWatches(int rc) {
-            return rc == 0 ? watchManager.dataWatches : watchManager.existWatches;
+            return rc == 0 ? getWatchManager().getDataWatches() : getWatchManager().getExistWatches();
         }
 
         @Override
@@ -707,7 +330,7 @@ public class ZooKeeper implements AutoCloseable {
 
         @Override
         protected Map<String, Set<Watcher>> getWatches(int rc) {
-            return watchManager.dataWatches;
+            return getWatchManager().getDataWatches();
         }
 
     }
@@ -720,7 +343,7 @@ public class ZooKeeper implements AutoCloseable {
 
         @Override
         protected Map<String, Set<Watcher>> getWatches(int rc) {
-            return watchManager.childWatches;
+            return getWatchManager().getChildWatches();
         }
 
     }
@@ -737,9 +360,9 @@ public class ZooKeeper implements AutoCloseable {
         protected Map<String, Set<Watcher>> getWatches(int rc) {
             switch (mode) {
                 case PERSISTENT:
-                    return watchManager.persistentWatches;
+                    return getWatchManager().getPersistentWatches();
                 case PERSISTENT_RECURSIVE:
-                    return watchManager.persistentRecursiveWatches;
+                    return getWatchManager().getPersistentRecursiveWatches();
             }
             throw new IllegalArgumentException("Mode not supported: " + mode);
         }
@@ -989,7 +612,7 @@ public class ZooKeeper implements AutoCloseable {
      *            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.
-     * @param aHostProvider
+     * @param hostProvider
      *            use this as HostProvider to enable custom behaviour.
      * @param clientConfig
      *            (added in 3.5.2) passing this conf object gives each client the flexibility of
@@ -1004,49 +627,45 @@ public class ZooKeeper implements AutoCloseable {
         int sessionTimeout,
         Watcher watcher,
         boolean canBeReadOnly,
-        HostProvider aHostProvider,
-        ZKClientConfig clientConfig) throws IOException {
+        HostProvider hostProvider,
+        ZKClientConfig clientConfig
+    ) throws IOException {
         LOG.info(
             "Initiating client connection, connectString={} sessionTimeout={} watcher={}",
             connectString,
             sessionTimeout,
             watcher);
 
-        if (clientConfig == null) {
-            clientConfig = new ZKClientConfig();
-        }
-        this.clientConfig = clientConfig;
-        watchManager = defaultWatchManager();
-        watchManager.defaultWatcher = watcher;
+        this.clientConfig = clientConfig != null ? clientConfig : new ZKClientConfig();
+        this.hostProvider = hostProvider;
         ConnectStringParser connectStringParser = new ConnectStringParser(connectString);
-        hostProvider = aHostProvider;
 
         cnxn = createConnection(
             connectStringParser.getChrootPath(),
             hostProvider,
             sessionTimeout,
-            this,
-            watchManager,
+            this.clientConfig,
+            watcher,
             getClientCnxnSocket(),
             canBeReadOnly);
         cnxn.start();
     }
 
-    // @VisibleForTesting
-    protected ClientCnxn createConnection(
+    ClientCnxn createConnection(
         String chrootPath,
         HostProvider hostProvider,
         int sessionTimeout,
-        ZooKeeper zooKeeper,
-        ClientWatchManager watcher,
+        ZKClientConfig clientConfig,
+        Watcher defaultWatcher,
         ClientCnxnSocket clientCnxnSocket,
-        boolean canBeReadOnly) throws IOException {
+        boolean canBeReadOnly
+    ) throws IOException {
         return new ClientCnxn(
             chrootPath,
             hostProvider,
             sessionTimeout,
-            this,
-            watchManager,
+            clientConfig,
+            defaultWatcher,
             clientCnxnSocket,
             canBeReadOnly);
     }
@@ -1383,7 +1002,7 @@ public class ZooKeeper implements AutoCloseable {
      *            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.
-     * @param aHostProvider
+     * @param hostProvider
      *            use this as HostProvider to enable custom behaviour.
      * @param clientConfig
      *            (added in 3.5.2) passing this conf object gives each client the flexibility of
@@ -1400,7 +1019,7 @@ public class ZooKeeper implements AutoCloseable {
         long sessionId,
         byte[] sessionPasswd,
         boolean canBeReadOnly,
-        HostProvider aHostProvider,
+        HostProvider hostProvider,
         ZKClientConfig clientConfig) throws IOException {
         LOG.info(
             "Initiating client connection, connectString={} "
@@ -1411,22 +1030,16 @@ public class ZooKeeper implements AutoCloseable {
             Long.toHexString(sessionId),
             (sessionPasswd == null ? "<null>" : "<hidden>"));
 
-        if (clientConfig == null) {
-            clientConfig = new ZKClientConfig();
-        }
-        this.clientConfig = clientConfig;
-        watchManager = defaultWatchManager();
-        watchManager.defaultWatcher = watcher;
-
+        this.clientConfig = clientConfig != null ? clientConfig : new ZKClientConfig();
         ConnectStringParser connectStringParser = new ConnectStringParser(connectString);
-        hostProvider = aHostProvider;
+        this.hostProvider = hostProvider;
 
         cnxn = new ClientCnxn(
             connectStringParser.getChrootPath(),
             hostProvider,
             sessionTimeout,
-            this,
-            watchManager,
+            this.clientConfig,
+            watcher,
             getClientCnxnSocket(),
             sessionId,
             sessionPasswd,
@@ -1523,11 +1136,6 @@ public class ZooKeeper implements AutoCloseable {
         return new ZooKeeperTestable(cnxn);
     }
 
-    /* Useful for testing watch handling behavior */
-    protected ZKWatchManager defaultWatchManager() {
-        return new ZKWatchManager(getClientConfig().getBoolean(ZKClientConfig.DISABLE_AUTO_WATCH_RESET));
-    }
-
     /**
      * The session id for this ZooKeeper client instance. The value returned is
      * not valid until the client connects to a server and may change after a
@@ -1582,11 +1190,9 @@ public class ZooKeeper implements AutoCloseable {
     /**
      * Specify the default watcher for the connection (overrides the one
      * specified during construction).
-     *
-     * @param watcher
      */
     public synchronized void register(Watcher watcher) {
-        watchManager.defaultWatcher = watcher;
+        getWatchManager().setDefaultWatcher(watcher);
     }
 
     /**
@@ -3213,9 +2819,11 @@ public class ZooKeeper implements AutoCloseable {
      *  error code.
      * @since 3.6.0
      */
-    public void addWatch(String basePath, AddWatchMode mode)
-            throws KeeperException, InterruptedException {
-        addWatch(basePath, watchManager.defaultWatcher, mode);
+    public void addWatch(
+            String basePath,
+            AddWatchMode mode
+    ) throws KeeperException, InterruptedException {
+        addWatch(basePath, getWatchManager().getDefaultWatcher(), mode);
     }
 
     /**
@@ -3229,8 +2837,12 @@ public class ZooKeeper implements AutoCloseable {
      * @throws IllegalArgumentException if an invalid path is specified
      * @since 3.6.0
      */
-    public void addWatch(String basePath, Watcher watcher, AddWatchMode mode,
-                         VoidCallback cb, Object ctx) {
+    public void addWatch(
+            String basePath,
+            Watcher watcher, AddWatchMode mode,
+            VoidCallback cb,
+            Object ctx
+    ) {
         PathUtils.validatePath(basePath);
         String serverPath = prependChroot(basePath);
 
@@ -3251,9 +2863,8 @@ public class ZooKeeper implements AutoCloseable {
      * @throws IllegalArgumentException if an invalid path is specified
      * @since 3.6.0
      */
-    public void addWatch(String basePath, AddWatchMode mode,
-                         VoidCallback cb, Object ctx) {
-        addWatch(basePath, watchManager.defaultWatcher, mode, cb, ctx);
+    public void addWatch(String basePath, AddWatchMode mode, VoidCallback cb, Object ctx) {
+        addWatch(basePath, getWatchManager().getDefaultWatcher(), mode, cb, ctx);
     }
 
     private void validateWatcher(Watcher watcher) {
@@ -3271,7 +2882,7 @@ public class ZooKeeper implements AutoCloseable {
         PathUtils.validatePath(path);
         final String clientPath = path;
         final String serverPath = prependChroot(clientPath);
-        WatchDeregistration wcb = new WatchDeregistration(clientPath, watcher, watcherType, local, watchManager);
+        WatchDeregistration wcb = new WatchDeregistration(clientPath, watcher, watcherType, local, getWatchManager());
 
         RequestHeader h = new RequestHeader();
         h.setType(opCode);
@@ -3294,7 +2905,7 @@ public class ZooKeeper implements AutoCloseable {
         PathUtils.validatePath(path);
         final String clientPath = path;
         final String serverPath = prependChroot(clientPath);
-        WatchDeregistration wcb = new WatchDeregistration(clientPath, watcher, watcherType, local, watchManager);
+        WatchDeregistration wcb = new WatchDeregistration(clientPath, watcher, watcherType, local, getWatchManager());
 
         RequestHeader h = new RequestHeader();
         h.setType(opCode);
@@ -3424,8 +3035,9 @@ public class ZooKeeper implements AutoCloseable {
      */
     private Watcher getDefaultWatcher(boolean required) {
         if (required) {
-            if (watchManager.defaultWatcher != null) {
-                return watchManager.defaultWatcher;
+            final Watcher defaultWatcher = getWatchManager().getDefaultWatcher();
+            if (defaultWatcher != null) {
+                return defaultWatcher;
             } else {
                 throw new IllegalStateException("Default watcher is required, but it is null.");
             }

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

@@ -281,11 +281,19 @@ public class ClientCnxnSocketFragilityTest extends QuorumPeerTestBase {
             String chrootPath,
             HostProvider hostProvider,
             int sessionTimeout,
-            ZooKeeper zooKeeper,
-            ClientWatchManager watcher,
+            ZKClientConfig zkClientConfig,
+            Watcher defaultWatcher,
             ClientCnxnSocket clientCnxnSocket,
-            boolean canBeReadOnly) throws IOException {
-            super(chrootPath, hostProvider, sessionTimeout, zooKeeper, watcher, clientCnxnSocket, canBeReadOnly);
+            boolean canBeReadOnly
+        ) throws IOException {
+            super(
+                chrootPath,
+                hostProvider,
+                sessionTimeout,
+                zkClientConfig,
+                defaultWatcher,
+                clientCnxnSocket,
+                canBeReadOnly);
         }
 
         void attemptClose() {
@@ -344,18 +352,25 @@ public class ClientCnxnSocketFragilityTest extends QuorumPeerTestBase {
             return cnxn.getState().isAlive();
         }
 
-        @Override
-        protected ClientCnxn createConnection(
+        ClientCnxn createConnection(
             String chrootPath,
             HostProvider hostProvider,
             int sessionTimeout,
-            ZooKeeper zooKeeper,
-            ClientWatchManager watcher,
+            ZKClientConfig clientConfig,
+            Watcher defaultWatcher,
             ClientCnxnSocket clientCnxnSocket,
-            boolean canBeReadOnly) throws IOException {
+            boolean canBeReadOnly
+        ) throws IOException {
             Assert.assertTrue(clientCnxnSocket instanceof FragileClientCnxnSocketNIO);
             socket = (FragileClientCnxnSocketNIO) clientCnxnSocket;
-            ClientCnxnSocketFragilityTest.this.cnxn = new CustomClientCnxn(chrootPath, hostProvider, sessionTimeout, zooKeeper, watcher, clientCnxnSocket, canBeReadOnly);
+            ClientCnxnSocketFragilityTest.this.cnxn = new CustomClientCnxn(
+                chrootPath,
+                hostProvider,
+                sessionTimeout,
+                clientConfig,
+                defaultWatcher,
+                clientCnxnSocket,
+                canBeReadOnly);
             return ClientCnxnSocketFragilityTest.this.cnxn;
         }
     }

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

@@ -66,8 +66,14 @@ public class ClientReconnectTest extends ZKTestCase {
         sc = SocketChannel.open();
 
         ClientCnxnSocketNIO nioCnxn = new MockCnxn();
-        ClientWatchManager watcher = mock(ClientWatchManager.class);
-        ClientCnxn clientCnxn = new ClientCnxn("tmp", hostProvider, 5000, zk, watcher, nioCnxn, false);
+        ClientCnxn clientCnxn = new ClientCnxn(
+            "tmp",
+            hostProvider,
+            5000,
+            zk.getClientConfig(),
+            DummyWatcher.INSTANCE,
+            nioCnxn,
+            false);
         clientCnxn.start();
         countDownLatch.await(5000, TimeUnit.MILLISECONDS);
         assertTrue(countDownLatch.getCount() == 0);

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

@@ -25,6 +25,7 @@ import static org.junit.Assert.fail;
 import java.io.IOException;
 import org.apache.zookeeper.ZooDefs.Ids;
 import org.apache.zookeeper.client.HostProvider;
+import org.apache.zookeeper.client.ZKClientConfig;
 import org.apache.zookeeper.server.quorum.QuorumPeerTestBase;
 import org.apache.zookeeper.test.ClientBase;
 import org.apache.zookeeper.test.ClientBase.CountdownWatcher;
@@ -110,15 +111,23 @@ public class ClientRequestTimeoutTest extends QuorumPeerTestBase {
 
     class CustomClientCnxn extends ClientCnxn {
 
-        public CustomClientCnxn(
+        CustomClientCnxn(
             String chrootPath,
             HostProvider hostProvider,
             int sessionTimeout,
-            ZooKeeper zooKeeper,
-            ClientWatchManager watcher,
+            ZKClientConfig clientConfig,
+            Watcher defaultWatcher,
             ClientCnxnSocket clientCnxnSocket,
-            boolean canBeReadOnly) throws IOException {
-            super(chrootPath, hostProvider, sessionTimeout, zooKeeper, watcher, clientCnxnSocket, canBeReadOnly);
+            boolean canBeReadOnly
+        ) throws IOException {
+            super(
+                chrootPath,
+                hostProvider,
+                sessionTimeout,
+                clientConfig,
+                defaultWatcher,
+                clientCnxnSocket,
+                canBeReadOnly);
         }
 
         @Override
@@ -140,15 +149,23 @@ public class ClientRequestTimeoutTest extends QuorumPeerTestBase {
         }
 
         @Override
-        protected ClientCnxn createConnection(
+        ClientCnxn createConnection(
             String chrootPath,
             HostProvider hostProvider,
             int sessionTimeout,
-            ZooKeeper zooKeeper,
-            ClientWatchManager watcher,
+            ZKClientConfig clientConfig,
+            Watcher defaultWatcher,
             ClientCnxnSocket clientCnxnSocket,
-            boolean canBeReadOnly) throws IOException {
-            return new CustomClientCnxn(chrootPath, hostProvider, sessionTimeout, zooKeeper, watcher, clientCnxnSocket, canBeReadOnly);
+            boolean canBeReadOnly
+        ) throws IOException {
+            return new CustomClientCnxn(
+                chrootPath,
+                hostProvider,
+                sessionTimeout,
+                clientConfig,
+                defaultWatcher,
+                clientCnxnSocket,
+                canBeReadOnly);
         }
 
     }

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

@@ -18,12 +18,16 @@
 
 package org.apache.zookeeper;
 
+import static org.hamcrest.CoreMatchers.is;
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertFalse;
 import static org.junit.Assert.assertNotNull;
 import static org.junit.Assert.assertNull;
+import static org.junit.Assert.assertThat;
 import static org.junit.Assert.assertTrue;
 import static org.junit.Assert.fail;
+import static org.mockito.Mockito.doReturn;
+import static org.mockito.Mockito.spy;
 import java.io.IOException;
 import java.util.ArrayList;
 import java.util.Arrays;
@@ -37,11 +41,9 @@ import java.util.concurrent.TimeUnit;
 import java.util.concurrent.TimeoutException;
 import org.apache.commons.collections.CollectionUtils;
 import org.apache.zookeeper.KeeperException.Code;
-import org.apache.zookeeper.KeeperException.NoWatcherException;
 import org.apache.zookeeper.Watcher.Event.EventType;
 import org.apache.zookeeper.Watcher.WatcherType;
 import org.apache.zookeeper.ZooDefs.Ids;
-import org.apache.zookeeper.client.ZKClientConfig;
 import org.apache.zookeeper.server.ServerCnxn;
 import org.apache.zookeeper.test.ClientBase;
 import org.junit.Test;
@@ -657,7 +659,9 @@ public class RemoveWatchesTest extends ClientBase {
     @Test(timeout = 90000)
     public void testNoWatcherServerException() throws InterruptedException, IOException, TimeoutException {
         CountdownWatcher watcher = new CountdownWatcher();
-        MyZooKeeper zk = new MyZooKeeper(hostPort, CONNECTION_TIMEOUT, watcher);
+        ZooKeeper zk = spy(new ZooKeeper(hostPort, CONNECTION_TIMEOUT, watcher));
+        MyWatchManager watchManager = new MyWatchManager(false, watcher);
+        doReturn(watchManager).when(zk).getWatchManager();
         boolean nw = false;
 
         watcher.waitForConnected(CONNECTION_TIMEOUT);
@@ -670,8 +674,8 @@ public class RemoveWatchesTest extends ClientBase {
             }
         }
 
-        assertTrue("Server didn't return NOWATCHER", zk.getRemoveWatchesRC() == Code.NOWATCHER.intValue());
-        assertTrue("NoWatcherException didn't happen", nw);
+        assertThat("Server didn't return NOWATCHER", watchManager.lastReturnCode, is(Code.NOWATCHER.intValue()));
+        assertThat("NoWatcherException didn't happen", nw, is(true));
     }
 
     /**
@@ -920,54 +924,32 @@ public class RemoveWatchesTest extends ClientBase {
         assertEquals("Received watch notification after removal!", 2, watchCount.getCount());
     }
 
-    /* a mocked ZK class that doesn't do client-side verification
-     * before/after calling removeWatches */
-    private class MyZooKeeper extends ZooKeeper {
+    private static class MyWatchManager extends ZKWatchManager {
 
-        class MyWatchManager extends ZKWatchManager {
+        int lastReturnCode;
 
-            public MyWatchManager(boolean disableAutoWatchReset) {
-                super(disableAutoWatchReset);
-            }
-
-            public int lastrc;
-
-            /* Pretend that any watcher exists */
-            void containsWatcher(String path, Watcher watcher, WatcherType watcherType) throws NoWatcherException {
-            }
-
-            /* save the return error code by the server */
-            protected boolean removeWatches(
-                Map<String, Set<Watcher>> pathVsWatcher,
-                Watcher watcher,
-                String path,
-                boolean local,
-                int rc,
-                Set<Watcher> removedWatchers) throws KeeperException {
-                lastrc = rc;
-                return false;
-            }
-
-        }
-
-        public MyZooKeeper(String hp, int timeout, Watcher watcher) throws IOException {
-            super(hp, timeout, watcher, false);
+        MyWatchManager(boolean disableAutoWatchReset, Watcher defaultWatcher) {
+            super(disableAutoWatchReset, defaultWatcher);
         }
 
-        private MyWatchManager myWatchManager;
-
-        protected ZKWatchManager defaultWatchManager() {
-            myWatchManager = new MyWatchManager(getClientConfig().getBoolean(ZKClientConfig.DISABLE_AUTO_WATCH_RESET));
-            return myWatchManager;
+        void containsWatcher(String path, Watcher watcher, WatcherType watcherType) {
+            // prevent contains watcher
         }
 
-        public int getRemoveWatchesRC() {
-            return myWatchManager.lastrc;
+        @Override
+        protected boolean removeWatches(
+            Map<String, Set<Watcher>> pathVsWatcher,
+            Watcher watcher,
+            String path,
+            boolean local,
+            int rc,
+            Set<Watcher> removedWatchers) {
+            lastReturnCode = rc;
+            return false;
         }
-
     }
 
-    private class MyWatcher implements Watcher {
+    private static class MyWatcher implements Watcher {
 
         private final String path;
         private String eventPath;

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

@@ -20,12 +20,10 @@ package org.apache.zookeeper;
 
 import java.io.IOException;
 import java.net.SocketAddress;
-import java.util.List;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.TimeUnit;
 import org.apache.jute.Record;
 import org.apache.zookeeper.admin.ZooKeeperAdmin;
-import org.apache.zookeeper.client.HostProvider;
 import org.apache.zookeeper.proto.ReplyHeader;
 import org.apache.zookeeper.proto.RequestHeader;
 
@@ -35,76 +33,12 @@ public class TestableZooKeeper extends ZooKeeperAdmin {
         super(host, sessionTimeout, watcher);
     }
 
-    class TestableClientCnxn extends ClientCnxn {
-
-        TestableClientCnxn(
-            String chrootPath,
-            HostProvider hostProvider,
-            int sessionTimeout,
-            ZooKeeper zooKeeper,
-            ClientWatchManager watcher,
-            ClientCnxnSocket clientCnxnSocket,
-            boolean canBeReadOnly) throws IOException {
-            super(chrootPath,
-                  hostProvider,
-                  sessionTimeout,
-                  zooKeeper,
-                  watcher,
-                  clientCnxnSocket,
-                  0,
-                  new byte[16],
-                  canBeReadOnly);
-        }
-
-        void setXid(int newXid) {
-            xid = newXid;
-        }
-
-        int checkXid() {
-            return xid;
-        }
-
-    }
-
-    protected ClientCnxn createConnection(
-        String chrootPath,
-        HostProvider hostProvider,
-        int sessionTimeout,
-        ZooKeeper zooKeeper,
-        ClientWatchManager watcher,
-        ClientCnxnSocket clientCnxnSocket,
-        boolean canBeReadOnly) throws IOException {
-        return new TestableClientCnxn(
-            chrootPath,
-            hostProvider,
-            sessionTimeout,
-            this,
-            watcher,
-            clientCnxnSocket,
-            canBeReadOnly);
-    }
-
     public void setXid(int xid) {
-        ((TestableClientCnxn) cnxn).setXid(xid);
+        cnxn.xid = xid;
     }
 
     public int checkXid() {
-        return ((TestableClientCnxn) cnxn).checkXid();
-    }
-
-    @Override
-    public List<String> getChildWatches() {
-        return super.getChildWatches();
-    }
-
-    @Override
-    public List<String> getDataWatches() {
-        return super.getDataWatches();
-    }
-
-    @Override
-    public List<String> getExistWatches() {
-        return super.getExistWatches();
+        return cnxn.xid;
     }
 
     /**