diff --git a/curator-client/src/main/java/org/apache/curator/utils/DebugUtils.java b/curator-client/src/main/java/org/apache/curator/utils/DebugUtils.java
index b098989..383bc13 100644
--- a/curator-client/src/main/java/org/apache/curator/utils/DebugUtils.java
+++ b/curator-client/src/main/java/org/apache/curator/utils/DebugUtils.java
@@ -23,6 +23,7 @@
     public static final String          PROPERTY_LOG_EVENTS = "curator-log-events";
     public static final String          PROPERTY_DONT_LOG_CONNECTION_ISSUES = "curator-dont-log-connection-problems";
     public static final String          PROPERTY_LOG_ONLY_FIRST_CONNECTION_ISSUE_AS_ERROR_LEVEL = "curator-log-only-first-connection-issue-as-error-level";
+    public static final String          PROPERTY_REMOVE_WATCHERS_IN_FOREGROUND = "curator-remove-watchers-in-foreground";
     public static final String          PROPERTY_RETRY_FAILED_TESTS = "curator-retry-failed-tests";
 
     private DebugUtils()
diff --git a/curator-framework/src/main/java/org/apache/curator/framework/CuratorFramework.java b/curator-framework/src/main/java/org/apache/curator/framework/CuratorFramework.java
index 9239ac4..58c5bf5 100644
--- a/curator-framework/src/main/java/org/apache/curator/framework/CuratorFramework.java
+++ b/curator-framework/src/main/java/org/apache/curator/framework/CuratorFramework.java
@@ -190,6 +190,12 @@
     public SyncBuilder sync();
 
     /**
+     * Start a remove watches builder.
+     * @return builder object
+     */
+    public RemoveWatchesBuilder watches();
+
+    /**
      * Returns the listenable interface for the Connect State
      *
      * @return listenable
@@ -259,7 +265,11 @@
      * Call this method on watchers you are no longer interested in.
      *
      * @param watcher the watcher
+     * 
+     * @deprecated As of ZooKeeper 3.5 Curators recipes will handle removing watcher references
+     * when they are no longer used.
      */
+    @Deprecated
     public void clearWatcherReferences(Watcher watcher);
         
     /**
@@ -278,4 +288,13 @@
      * @throws InterruptedException If interrupted while waiting
      */
     public void blockUntilConnected() throws InterruptedException;
+
+    /**
+     * Returns a facade of the current instance that tracks
+     * watchers created and allows a one-shot removal of all watchers
+     * via {@link WatcherRemoveCuratorFramework#removeWatchers()}
+     *
+     * @return facade
+     */
+    public WatcherRemoveCuratorFramework newWatcherRemoveCuratorFramework();
 }
diff --git a/curator-framework/src/main/java/org/apache/curator/framework/WatcherRemoveCuratorFramework.java b/curator-framework/src/main/java/org/apache/curator/framework/WatcherRemoveCuratorFramework.java
new file mode 100644
index 0000000..871b53c
--- /dev/null
+++ b/curator-framework/src/main/java/org/apache/curator/framework/WatcherRemoveCuratorFramework.java
@@ -0,0 +1,30 @@
+/**
+ * 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.curator.framework;
+
+/**
+ * A CuratorFramework facade that tracks watchers created and allows a one-shot removal of all watchers
+ */
+public interface WatcherRemoveCuratorFramework extends CuratorFramework
+{
+    /**
+     * Remove all outstanding watchers that have been set
+     */
+    void removeWatchers();
+}
diff --git a/curator-framework/src/main/java/org/apache/curator/framework/api/BackgroundPathableQuietly.java b/curator-framework/src/main/java/org/apache/curator/framework/api/BackgroundPathableQuietly.java
new file mode 100644
index 0000000..13202aa
--- /dev/null
+++ b/curator-framework/src/main/java/org/apache/curator/framework/api/BackgroundPathableQuietly.java
@@ -0,0 +1,23 @@
+/**
+ * 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.curator.framework.api;
+
+public interface BackgroundPathableQuietly<T> extends BackgroundPathable<T>, Quietly<BackgroundPathable<T>>
+{
+}
diff --git a/curator-framework/src/main/java/org/apache/curator/framework/api/BackgroundPathableQuietlyable.java b/curator-framework/src/main/java/org/apache/curator/framework/api/BackgroundPathableQuietlyable.java
new file mode 100644
index 0000000..8ed73fa
--- /dev/null
+++ b/curator-framework/src/main/java/org/apache/curator/framework/api/BackgroundPathableQuietlyable.java
@@ -0,0 +1,5 @@
+package org.apache.curator.framework.api;
+
+public interface BackgroundPathableQuietlyable<T> extends BackgroundPathable<T>, Quietly<BackgroundPathable<T>>
+{
+}
diff --git a/curator-framework/src/main/java/org/apache/curator/framework/api/CuratorEventType.java b/curator-framework/src/main/java/org/apache/curator/framework/api/CuratorEventType.java
index 5a2dc56..5dea211 100644
--- a/curator-framework/src/main/java/org/apache/curator/framework/api/CuratorEventType.java
+++ b/curator-framework/src/main/java/org/apache/curator/framework/api/CuratorEventType.java
@@ -89,6 +89,11 @@
     WATCHED,
 
     /**
+     * Corresponds to {@link CuratorFramework#watches()} ()}
+     */
+    REMOVE_WATCHES,
+
+    /**
      * Event sent when client is being closed
      */
     CLOSING
diff --git a/curator-framework/src/main/java/org/apache/curator/framework/api/DeleteBuilder.java b/curator-framework/src/main/java/org/apache/curator/framework/api/DeleteBuilder.java
index 3a3faf7..2da1843 100644
--- a/curator-framework/src/main/java/org/apache/curator/framework/api/DeleteBuilder.java
+++ b/curator-framework/src/main/java/org/apache/curator/framework/api/DeleteBuilder.java
@@ -18,6 +18,6 @@
  */
 package org.apache.curator.framework.api;
 
-public interface DeleteBuilder extends Guaranteeable, ChildrenDeletable
+public interface DeleteBuilder extends GuaranteeableDeletable, ChildrenDeletable
 {
 }
diff --git a/curator-framework/src/main/java/org/apache/curator/framework/api/Guaranteeable.java b/curator-framework/src/main/java/org/apache/curator/framework/api/Guaranteeable.java
index 481911b..bc033ac 100644
--- a/curator-framework/src/main/java/org/apache/curator/framework/api/Guaranteeable.java
+++ b/curator-framework/src/main/java/org/apache/curator/framework/api/Guaranteeable.java
@@ -18,23 +18,15 @@
  */
 package org.apache.curator.framework.api;
 
-public interface Guaranteeable extends BackgroundVersionable
+public interface Guaranteeable<T>
 {
     /**
-     * <p>
-     *     Solves this edge case: deleting a node can fail due to connection issues. Further,
-     *     if the node was ephemeral, the node will not get auto-deleted as the session is still valid.
-     *     This can wreak havoc with lock implementations.
-     * </p>
-     *
-     * <p>
-     *     When <code>guaranteed</code> is set, Curator will record failed node deletions and
-     *     attempt to delete them in the background until successful. NOTE: you will still get an
-     *     exception when the deletion fails. But, you can be assured that as long as the
-     *     {@link org.apache.curator.framework.CuratorFramework} instance is open attempts will be made to delete the node.
-     * </p>
+     * Solves edge cases where an operation may succeed on the server but connection failure occurs before a
+     * response can be successfully returned to the client.
+     * 
+     * @see org.apache.curator.framework.api.GuaranteeableDeletable 
      *  
      * @return this
      */
-    public ChildrenDeletable guaranteed();
+    public T guaranteed();
 }
diff --git a/curator-framework/src/main/java/org/apache/curator/framework/api/GuaranteeableDeletable.java b/curator-framework/src/main/java/org/apache/curator/framework/api/GuaranteeableDeletable.java
new file mode 100644
index 0000000..7f8139c
--- /dev/null
+++ b/curator-framework/src/main/java/org/apache/curator/framework/api/GuaranteeableDeletable.java
@@ -0,0 +1,39 @@
+/**
+ * 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.curator.framework.api;
+
+/**
+ * <p>
+ *     Solves this edge case: deleting a node can fail due to connection issues. Further,
+ *     if the node was ephemeral, the node will not get auto-deleted as the session is still valid.
+ *     This can wreak havoc with lock implementations.
+ * </p>
+ *
+ * <p>
+ *     When <code>guaranteed</code> is set, Curator will record failed node deletions and
+ *     attempt to delete them in the background until successful. NOTE: you will still get an
+ *     exception when the deletion fails. But, you can be assured that as long as the
+ *     {@link org.apache.curator.framework.CuratorFramework} instance is open attempts will be made to delete the node.
+ * </p>
+ *  
+ * @return this
+ */
+public interface GuaranteeableDeletable extends Guaranteeable<ChildrenDeletable>, BackgroundVersionable
+{
+}
diff --git a/curator-framework/src/main/java/org/apache/curator/framework/api/Quietly.java b/curator-framework/src/main/java/org/apache/curator/framework/api/Quietly.java
new file mode 100644
index 0000000..ad3762f
--- /dev/null
+++ b/curator-framework/src/main/java/org/apache/curator/framework/api/Quietly.java
@@ -0,0 +1,24 @@
+/**
+ * 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.curator.framework.api;
+
+public interface Quietly<T>
+{
+    public T quietly();
+}
diff --git a/curator-framework/src/main/java/org/apache/curator/framework/api/RemoveWatchesBuilder.java b/curator-framework/src/main/java/org/apache/curator/framework/api/RemoveWatchesBuilder.java
new file mode 100644
index 0000000..6cc0b05
--- /dev/null
+++ b/curator-framework/src/main/java/org/apache/curator/framework/api/RemoveWatchesBuilder.java
@@ -0,0 +1,47 @@
+/**
+ * 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.curator.framework.api;
+
+import org.apache.zookeeper.Watcher;
+
+/**
+ * Builder to allow watches to be removed 
+ */
+public interface RemoveWatchesBuilder
+{
+    /**
+     * Specify the watcher to be removed
+     * @param watcher
+     * @return
+     */
+    public RemoveWatchesType remove(Watcher watcher);
+    
+    /**
+     * Specify the watcher to be removed
+     * @param watcher
+     * @return
+     */
+    public RemoveWatchesType remove(CuratorWatcher watcher);
+    
+    /**
+     * Specify that all watches should be removed
+     * @return
+     */
+    public RemoveWatchesType removeAll();
+}
diff --git a/curator-framework/src/main/java/org/apache/curator/framework/api/RemoveWatchesLocal.java b/curator-framework/src/main/java/org/apache/curator/framework/api/RemoveWatchesLocal.java
new file mode 100644
index 0000000..4a67470
--- /dev/null
+++ b/curator-framework/src/main/java/org/apache/curator/framework/api/RemoveWatchesLocal.java
@@ -0,0 +1,35 @@
+/**
+ * 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.curator.framework.api;
+
+/**
+ * Builder to allow the specification of whether it is acceptable to remove client side watch information
+ * in the case where ZK cannot be contacted. 
+ */
+public interface RemoveWatchesLocal extends BackgroundPathableQuietlyable<Void>
+{
+   
+    /**
+     * Specify if the client should just remove client side watches if a connection to ZK
+     * is not available. Note that the standard Curator retry loop will not be used in t
+     * @return
+     */
+    public BackgroundPathableQuietlyable<Void> locally();
+    
+}
diff --git a/curator-framework/src/main/java/org/apache/curator/framework/api/RemoveWatchesType.java b/curator-framework/src/main/java/org/apache/curator/framework/api/RemoveWatchesType.java
new file mode 100644
index 0000000..84d8093
--- /dev/null
+++ b/curator-framework/src/main/java/org/apache/curator/framework/api/RemoveWatchesType.java
@@ -0,0 +1,37 @@
+/**
+ * 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.curator.framework.api;
+
+import org.apache.zookeeper.Watcher.WatcherType;
+
+/**
+ * Builder to allow the specification of whether it is acceptable to remove client side watch information
+ * in the case where ZK cannot be contacted. 
+ */
+public interface RemoveWatchesType extends RemoveWatchesLocal, Guaranteeable<BackgroundPathableQuietlyable<Void>>
+{
+   
+    /**
+     * Specify the type of watcher to be removed.
+     * @param watcherType
+     * @return
+     */
+    public RemoveWatchesLocal ofType(WatcherType watcherType);
+    
+}
diff --git a/curator-framework/src/main/java/org/apache/curator/framework/imps/CuratorFrameworkImpl.java b/curator-framework/src/main/java/org/apache/curator/framework/imps/CuratorFrameworkImpl.java
index 900374b..41bb7cd 100644
--- a/curator-framework/src/main/java/org/apache/curator/framework/imps/CuratorFrameworkImpl.java
+++ b/curator-framework/src/main/java/org/apache/curator/framework/imps/CuratorFrameworkImpl.java
@@ -30,6 +30,7 @@
 import org.apache.curator.framework.AuthInfo;
 import org.apache.curator.framework.CuratorFramework;
 import org.apache.curator.framework.CuratorFrameworkFactory;
+import org.apache.curator.framework.WatcherRemoveCuratorFramework;
 import org.apache.curator.framework.api.*;
 import org.apache.curator.framework.api.transaction.CuratorMultiTransaction;
 import org.apache.curator.framework.api.transaction.CuratorTransaction;
@@ -76,6 +77,7 @@
     private final List<AuthInfo> authInfos;
     private final byte[] defaultData;
     private final FailedDeleteManager failedDeleteManager;
+    private final FailedRemoveWatchManager failedRemoveWatcherManager;
     private final CompressionProvider compressionProvider;
     private final ACLProvider aclProvider;
     private final NamespaceFacadeCache namespaceFacadeCache;
@@ -128,6 +130,7 @@
         authInfos = buildAuths(builder);
 
         failedDeleteManager = new FailedDeleteManager(this);
+        failedRemoveWatcherManager = new FailedRemoveWatchManager(this);
         namespaceFacadeCache = new NamespaceFacadeCache(this);
     }
 
@@ -141,6 +144,12 @@
         return builder1.build();
     }
 
+    @Override
+    public WatcherRemoveCuratorFramework newWatcherRemoveCuratorFramework()
+    {
+        return new WatcherRemovalFacade(this);
+    }
+
     private ZookeeperFactory makeZookeeperFactory(final ZookeeperFactory actualZookeeperFactory)
     {
         return new ZookeeperFactory()
@@ -180,6 +189,7 @@
         connectionStateManager = parent.connectionStateManager;
         defaultData = parent.defaultData;
         failedDeleteManager = parent.failedDeleteManager;
+        failedRemoveWatcherManager = parent.failedRemoveWatcherManager;
         compressionProvider = parent.compressionProvider;
         aclProvider = parent.aclProvider;
         namespaceFacadeCache = parent.namespaceFacadeCache;
@@ -479,6 +489,12 @@
         return new SyncBuilderImpl(this);
     }
 
+    @Override
+    public RemoveWatchesBuilder watches()
+    {
+        return new RemoveWatchesBuilderImpl(this);
+    }
+
     protected void internalSync(CuratorFrameworkImpl impl, String path, Object context)
     {
         BackgroundOperation<String> operation = new BackgroundSyncImpl(impl, context);
@@ -507,6 +523,11 @@
         return failedDeleteManager;
     }
 
+    FailedRemoveWatchManager getFailedRemoveWatcherManager()
+    {
+        return failedRemoveWatcherManager;
+    }
+
     RetryLoop newRetryLoop()
     {
         return client.newRetryLoop();
@@ -689,6 +710,11 @@
         return Watcher.Event.KeeperState.fromInt(-1);
     }
 
+    WatcherRemovalManager getWatcherRemovalManager()
+    {
+        return null;
+    }
+
     private void suspendConnection()
     {
         if ( !connectionStateManager.setToSuspended() )
@@ -842,7 +868,7 @@
     {
         try
         {
-            if ( client.isConnected() )
+            if ( !operationAndData.isConnectionRequired() || client.isConnected() )
             {
                 operationAndData.callPerformBackgroundOperation();
             }
diff --git a/curator-framework/src/main/java/org/apache/curator/framework/imps/DeleteBuilderImpl.java b/curator-framework/src/main/java/org/apache/curator/framework/imps/DeleteBuilderImpl.java
index 51641b8..2a98f56 100644
--- a/curator-framework/src/main/java/org/apache/curator/framework/imps/DeleteBuilderImpl.java
+++ b/curator-framework/src/main/java/org/apache/curator/framework/imps/DeleteBuilderImpl.java
@@ -205,7 +205,7 @@
                     @Override
                     public void retriesExhausted(OperationAndData<String> operationAndData)
                     {
-                        client.getFailedDeleteManager().addFailedDelete(unfixedPath);
+                        client.getFailedDeleteManager().addFailedOperation(unfixedPath);
                     }
                 };
             }
@@ -261,7 +261,7 @@
             //Only retry a guaranteed delete if it's a retryable error
             if( (RetryLoop.isRetryException(e) || (e instanceof InterruptedException)) && guaranteed )
             {
-                client.getFailedDeleteManager().addFailedDelete(unfixedPath);
+                client.getFailedDeleteManager().addFailedOperation(unfixedPath);
             }
             throw e;
         }
diff --git a/curator-framework/src/main/java/org/apache/curator/framework/imps/ExistsBuilderImpl.java b/curator-framework/src/main/java/org/apache/curator/framework/imps/ExistsBuilderImpl.java
index 345bcf5..a6316ac 100644
--- a/curator-framework/src/main/java/org/apache/curator/framework/imps/ExistsBuilderImpl.java
+++ b/curator-framework/src/main/java/org/apache/curator/framework/imps/ExistsBuilderImpl.java
@@ -141,7 +141,7 @@
         }
         else
         {
-            client.getZooKeeper().exists(operationAndData.getData(), watching.getWatcher(), callback, backgrounding.getContext());
+            client.getZooKeeper().exists(operationAndData.getData(), watching.getWatcher(client, operationAndData.getData()), callback, backgrounding.getContext());
         }
     }
 
@@ -223,7 +223,7 @@
                     }
                     else
                     {
-                        returnStat = client.getZooKeeper().exists(path, watching.getWatcher());
+                        returnStat = client.getZooKeeper().exists(path, watching.getWatcher(client, path));
                     }
                     return returnStat;
                 }
diff --git a/curator-framework/src/main/java/org/apache/curator/framework/imps/FailedDeleteManager.java b/curator-framework/src/main/java/org/apache/curator/framework/imps/FailedDeleteManager.java
index deb7f40..934ae40 100644
--- a/curator-framework/src/main/java/org/apache/curator/framework/imps/FailedDeleteManager.java
+++ b/curator-framework/src/main/java/org/apache/curator/framework/imps/FailedDeleteManager.java
@@ -19,45 +19,18 @@
 package org.apache.curator.framework.imps;
 
 import org.apache.curator.framework.CuratorFramework;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
 
-class FailedDeleteManager
+class FailedDeleteManager extends FailedOperationManager<String>
 {
-    private final Logger log = LoggerFactory.getLogger(getClass());
-    private final CuratorFramework client;
-    
-    volatile FailedDeleteManagerListener debugListener = null;
-    
-    interface FailedDeleteManagerListener
-    {
-       public void pathAddedForDelete(String path);
-    }
-
     FailedDeleteManager(CuratorFramework client)
     {
-        this.client = client;
+        super(client);
     }
 
-    void addFailedDelete(String path)
+    @Override
+    protected void executeGuaranteedOperationInBackground(String path)
+            throws Exception
     {
-        if ( debugListener != null )
-        {
-            debugListener.pathAddedForDelete(path);
-        }
-        
-        
-        if ( client.getState() == CuratorFrameworkState.STARTED )
-        {
-            log.debug("Path being added to guaranteed delete set: " + path);
-            try
-            {
-                client.delete().guaranteed().inBackground().forPath(path);
-            }
-            catch ( Exception e )
-            {
-                addFailedDelete(path);
-            }
-        }
+        client.delete().guaranteed().inBackground().forPath(path);
     }
 }
diff --git a/curator-framework/src/main/java/org/apache/curator/framework/imps/FailedOperationManager.java b/curator-framework/src/main/java/org/apache/curator/framework/imps/FailedOperationManager.java
new file mode 100644
index 0000000..405561b
--- /dev/null
+++ b/curator-framework/src/main/java/org/apache/curator/framework/imps/FailedOperationManager.java
@@ -0,0 +1,68 @@
+/**
+ * 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.curator.framework.imps;
+
+import org.apache.curator.framework.CuratorFramework;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import com.google.common.annotations.VisibleForTesting;
+
+abstract class FailedOperationManager<T>
+{
+    private final Logger log = LoggerFactory.getLogger(getClass());
+    protected final CuratorFramework client;
+    
+    @VisibleForTesting
+    volatile FailedOperationManagerListener<T> debugListener = null;
+    
+    interface FailedOperationManagerListener<T>
+    {
+       public void pathAddedForGuaranteedOperation(T detail);
+    }
+
+    FailedOperationManager(CuratorFramework client)
+    {
+        this.client = client;
+    }
+
+    void addFailedOperation(T details)
+    {
+        if ( debugListener != null )
+        {
+            debugListener.pathAddedForGuaranteedOperation(details);
+        }
+        
+        
+        if ( client.getState() == CuratorFrameworkState.STARTED )
+        {
+            log.debug("Details being added to guaranteed operation set: " + details);
+            try
+            {
+                executeGuaranteedOperationInBackground(details);
+            }
+            catch ( Exception e )
+            {
+                addFailedOperation(details);
+            }
+        }
+    }
+    
+    protected abstract void executeGuaranteedOperationInBackground(T details) throws Exception;
+}
diff --git a/curator-framework/src/main/java/org/apache/curator/framework/imps/FailedRemoveWatchManager.java b/curator-framework/src/main/java/org/apache/curator/framework/imps/FailedRemoveWatchManager.java
new file mode 100644
index 0000000..f635660
--- /dev/null
+++ b/curator-framework/src/main/java/org/apache/curator/framework/imps/FailedRemoveWatchManager.java
@@ -0,0 +1,56 @@
+/**
+ * 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.curator.framework.imps;
+
+import org.apache.curator.framework.CuratorFramework;
+import org.apache.zookeeper.Watcher;
+
+class FailedRemoveWatchManager extends FailedOperationManager<FailedRemoveWatchManager.FailedRemoveWatchDetails>
+{
+    FailedRemoveWatchManager(CuratorFramework client)
+    {
+        super(client);
+    }
+
+    @Override
+    protected void executeGuaranteedOperationInBackground(FailedRemoveWatchDetails details)
+            throws Exception
+    {
+        if(details.watcher == null)
+        {
+            client.watches().removeAll().guaranteed().inBackground().forPath(details.path);
+        }
+        else
+        {
+            client.watches().remove(details.watcher).guaranteed().inBackground().forPath(details.path);
+        }
+    }
+    
+    static class FailedRemoveWatchDetails
+    {
+        public final String path;
+        public final Watcher watcher;
+        
+        public FailedRemoveWatchDetails(String path, Watcher watcher)
+        {
+            this.path = path;
+            this.watcher = watcher;
+        }
+    }
+}
diff --git a/curator-framework/src/main/java/org/apache/curator/framework/imps/GetChildrenBuilderImpl.java b/curator-framework/src/main/java/org/apache/curator/framework/imps/GetChildrenBuilderImpl.java
index 03010ce..7929ba3 100644
--- a/curator-framework/src/main/java/org/apache/curator/framework/imps/GetChildrenBuilderImpl.java
+++ b/curator-framework/src/main/java/org/apache/curator/framework/imps/GetChildrenBuilderImpl.java
@@ -172,7 +172,7 @@
         }
         else
         {
-            client.getZooKeeper().getChildren(operationAndData.getData(), watching.getWatcher(), callback, backgrounding.getContext());
+            client.getZooKeeper().getChildren(operationAndData.getData(), watching.getWatcher(client, operationAndData.getData()), callback, backgrounding.getContext());
         }
     }
 
@@ -211,7 +211,7 @@
                     }
                     else
                     {
-                        children = client.getZooKeeper().getChildren(path, watching.getWatcher(), responseStat);
+                        children = client.getZooKeeper().getChildren(path, watching.getWatcher(client, path), responseStat);
                     }
                     return children;
                 }
diff --git a/curator-framework/src/main/java/org/apache/curator/framework/imps/GetConfigBuilderImpl.java b/curator-framework/src/main/java/org/apache/curator/framework/imps/GetConfigBuilderImpl.java
index a837809..5468bd4 100644
--- a/curator-framework/src/main/java/org/apache/curator/framework/imps/GetConfigBuilderImpl.java
+++ b/curator-framework/src/main/java/org/apache/curator/framework/imps/GetConfigBuilderImpl.java
@@ -30,6 +30,7 @@
 import org.apache.curator.framework.api.GetConfigBuilder;
 import org.apache.zookeeper.AsyncCallback;
 import org.apache.zookeeper.Watcher;
+import org.apache.zookeeper.ZooDefs;
 import org.apache.zookeeper.data.Stat;
 import java.util.concurrent.Callable;
 import java.util.concurrent.Executor;
@@ -153,7 +154,7 @@
         }
         else
         {
-            client.getZooKeeper().getConfig(watching.getWatcher(), callback, backgrounding.getContext());
+            client.getZooKeeper().getConfig(watching.getWatcher(client, ZooDefs.CONFIG_NODE), callback, backgrounding.getContext());
         }
     }
 
@@ -174,7 +175,7 @@
                         {
                             return client.getZooKeeper().getConfig(true, stat);
                         }
-                        return client.getZooKeeper().getConfig(watching.getWatcher(), stat);
+                        return client.getZooKeeper().getConfig(watching.getWatcher(client, ZooDefs.CONFIG_NODE), stat);
                     }
                 }
             );
diff --git a/curator-framework/src/main/java/org/apache/curator/framework/imps/GetDataBuilderImpl.java b/curator-framework/src/main/java/org/apache/curator/framework/imps/GetDataBuilderImpl.java
index 23da075..31ad598 100644
--- a/curator-framework/src/main/java/org/apache/curator/framework/imps/GetDataBuilderImpl.java
+++ b/curator-framework/src/main/java/org/apache/curator/framework/imps/GetDataBuilderImpl.java
@@ -260,7 +260,7 @@
         }
         else
         {
-            client.getZooKeeper().getData(operationAndData.getData(), watching.getWatcher(), callback, backgrounding.getContext());
+            client.getZooKeeper().getData(operationAndData.getData(), watching.getWatcher(client, operationAndData.getData()), callback, backgrounding.getContext());
         }
     }
 
@@ -299,7 +299,7 @@
                     }
                     else
                     {
-                        responseData = client.getZooKeeper().getData(path, watching.getWatcher(), responseStat);
+                        responseData = client.getZooKeeper().getData(path, watching.getWatcher(client, path), responseStat);
                     }
                     return responseData;
                 }
diff --git a/curator-framework/src/main/java/org/apache/curator/framework/imps/OperationAndData.java b/curator-framework/src/main/java/org/apache/curator/framework/imps/OperationAndData.java
index 38f59a0..b46cddb 100644
--- a/curator-framework/src/main/java/org/apache/curator/framework/imps/OperationAndData.java
+++ b/curator-framework/src/main/java/org/apache/curator/framework/imps/OperationAndData.java
@@ -40,25 +40,37 @@
     private final AtomicLong sleepUntilTimeMs = new AtomicLong(0);
     private final long ordinal = nextOrdinal.getAndIncrement();
     private final Object context;
+    private final boolean connectionRequired;
 
     interface ErrorCallback<T>
     {
         void retriesExhausted(OperationAndData<T> operationAndData);
     }
-
-    OperationAndData(BackgroundOperation<T> operation, T data, BackgroundCallback callback, ErrorCallback<T> errorCallback, Object context)
+    
+    OperationAndData(BackgroundOperation<T> operation, T data, BackgroundCallback callback, ErrorCallback<T> errorCallback, Object context, boolean connectionRequired)
     {
         this.operation = operation;
         this.data = data;
         this.callback = callback;
         this.errorCallback = errorCallback;
         this.context = context;
+        this.connectionRequired = connectionRequired;
+    }      
+
+    OperationAndData(BackgroundOperation<T> operation, T data, BackgroundCallback callback, ErrorCallback<T> errorCallback, Object context)
+    {
+        this(operation, data, callback, errorCallback, context, true);
     }
 
     Object getContext()
     {
         return context;
     }
+    
+    boolean isConnectionRequired()
+    {
+        return connectionRequired;
+    }
 
     void callPerformBackgroundOperation() throws Exception
     {
diff --git a/curator-framework/src/main/java/org/apache/curator/framework/imps/RemoveWatchesBuilderImpl.java b/curator-framework/src/main/java/org/apache/curator/framework/imps/RemoveWatchesBuilderImpl.java
new file mode 100644
index 0000000..d872ced
--- /dev/null
+++ b/curator-framework/src/main/java/org/apache/curator/framework/imps/RemoveWatchesBuilderImpl.java
@@ -0,0 +1,315 @@
+/**
+ * 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.curator.framework.imps;
+
+import java.util.concurrent.Callable;
+import java.util.concurrent.Executor;
+
+import org.apache.curator.RetryLoop;
+import org.apache.curator.TimeTrace;
+import org.apache.curator.framework.api.BackgroundCallback;
+import org.apache.curator.framework.api.BackgroundPathable;
+import org.apache.curator.framework.api.BackgroundPathableQuietlyable;
+import org.apache.curator.framework.api.CuratorEvent;
+import org.apache.curator.framework.api.CuratorEventType;
+import org.apache.curator.framework.api.CuratorWatcher;
+import org.apache.curator.framework.api.Pathable;
+import org.apache.curator.framework.api.RemoveWatchesLocal;
+import org.apache.curator.framework.api.RemoveWatchesBuilder;
+import org.apache.curator.framework.api.RemoveWatchesType;
+import org.apache.curator.utils.DebugUtils;
+import org.apache.zookeeper.AsyncCallback;
+import org.apache.zookeeper.KeeperException;
+import org.apache.zookeeper.Watcher;
+import org.apache.zookeeper.Watcher.WatcherType;
+import org.apache.zookeeper.ZooKeeper;
+
+
+public class RemoveWatchesBuilderImpl implements RemoveWatchesBuilder, RemoveWatchesType, RemoveWatchesLocal, BackgroundOperation<String>
+{
+    private CuratorFrameworkImpl client;
+    private Watcher watcher;
+    private WatcherType watcherType;
+    private boolean guaranteed;
+    private boolean local;
+    private boolean quietly;    
+    private Backgrounding backgrounding;
+    
+    public RemoveWatchesBuilderImpl(CuratorFrameworkImpl client)
+    {
+        this.client = client;
+        this.watcher = null;
+        this.watcherType = WatcherType.Any;
+        this.guaranteed = false;
+        this.local = false;
+        this.quietly = false;
+        this.backgrounding = new Backgrounding();
+    }
+
+    void internalRemoval(Watcher watcher, String path) throws Exception
+    {
+        this.watcher = watcher;
+        watcherType = WatcherType.Any;
+        quietly = true;
+        guaranteed = true;
+        if ( Boolean.getBoolean(DebugUtils.PROPERTY_REMOVE_WATCHERS_IN_FOREGROUND) )
+        {
+            this.backgrounding = new Backgrounding();
+            pathInForeground(path);
+        }
+        else
+        {
+            this.backgrounding = new Backgrounding(true);
+            pathInBackground(path);
+        }
+    }
+
+    @Override
+    public RemoveWatchesType remove(Watcher watcher)
+    {
+        if(watcher == null) {
+            this.watcher = null;
+        } else {
+            //Try and get the namespaced version of the watcher.
+            this.watcher = client.getNamespaceWatcherMap().get(watcher);
+            
+            //If this is not present then default to the original watcher. This shouldn't happen in practice unless the user
+            //has added a watch directly to the ZK client rather than via the CuratorFramework.
+            if(this.watcher == null) {
+                this.watcher = watcher;
+            }
+        }
+
+        return this;
+    }
+    
+    @Override
+    public RemoveWatchesType remove(CuratorWatcher watcher)
+    {
+        this.watcher = watcher == null ? null : client.getNamespaceWatcherMap().get(watcher);
+        return this;
+    }    
+
+    @Override
+    public RemoveWatchesType removeAll()
+    {
+        this.watcher = null;
+        return this;
+    }
+
+    @Override
+    public RemoveWatchesLocal ofType(WatcherType watcherType)
+    {
+        this.watcherType = watcherType;
+        
+        return this;
+    }
+
+    @Override
+    public Pathable<Void> inBackground(BackgroundCallback callback, Object context)
+    {
+        backgrounding = new Backgrounding(callback, context);
+        return this;
+    }
+
+    @Override
+    public Pathable<Void> inBackground(BackgroundCallback callback, Object context, Executor executor)
+    {
+        backgrounding = new Backgrounding(client, callback, context, executor);
+        return this;
+    }
+
+    @Override
+    public Pathable<Void> inBackground(BackgroundCallback callback)
+    {
+        backgrounding = new Backgrounding(callback);
+        return this;
+    }
+
+    @Override
+    public Pathable<Void> inBackground(BackgroundCallback callback, Executor executor)
+    {
+        backgrounding = new Backgrounding(client, callback, executor);
+        return this;
+    }
+
+    @Override
+    public Pathable<Void> inBackground()
+    {
+        backgrounding = new Backgrounding(true);
+        return this;
+    }
+
+    @Override
+    public Pathable<Void> inBackground(Object context)
+    {
+        backgrounding = new Backgrounding(context);
+        return this;
+    }
+    
+    @Override
+    public RemoveWatchesLocal guaranteed()
+    {
+        guaranteed = true;
+        return this;
+    }    
+
+    @Override
+    public BackgroundPathableQuietlyable<Void> locally()
+    {
+        local = true;
+        return this;
+    }
+    
+    @Override
+    public BackgroundPathable<Void> quietly()
+    {
+        quietly = true;
+        return this;
+    }
+    
+    @Override
+    public Void forPath(String path) throws Exception
+    {
+        final String adjustedPath = client.fixForNamespace(path);
+        
+        if(backgrounding.inBackground())
+        {
+            pathInBackground(adjustedPath);
+        }
+        else
+        {
+            pathInForeground(adjustedPath);
+        }        
+        
+        return null;
+    }    
+    
+    private void pathInBackground(final String path)
+    {
+        OperationAndData.ErrorCallback<String>  errorCallback = null;
+        
+        //Only need an error callback if we're in guaranteed mode
+        if(guaranteed)
+        {
+            errorCallback = new OperationAndData.ErrorCallback<String>()
+            {
+                @Override
+                public void retriesExhausted(OperationAndData<String> operationAndData)
+                {
+                    client.getFailedRemoveWatcherManager().addFailedOperation(new FailedRemoveWatchManager.FailedRemoveWatchDetails(path, watcher));
+                }            
+            };
+        }
+        
+        client.processBackgroundOperation(new OperationAndData<String>(this, path, backgrounding.getCallback(),
+                                                                       errorCallback, backgrounding.getContext(), !local), null);
+    }
+    
+    private void pathInForeground(final String path) throws Exception
+    {
+        //For the local case we don't want to use the normal retry loop and we don't want to block until a connection is available.
+        //We just execute the removeWatch, and if it fails, ZK will just remove local watches.
+        if(local)
+        {
+            ZooKeeper zkClient = client.getZooKeeper();
+            if(watcher == null)
+            {
+                zkClient.removeAllWatches(path, watcherType, local);    
+            }
+            else
+            {
+                zkClient.removeWatches(path, watcher, watcherType, local);
+            }
+        }
+        else
+        {
+            RetryLoop.callWithRetry(client.getZookeeperClient(), 
+                    new Callable<Void>()
+                    {
+                        @Override
+                        public Void call() throws Exception
+                        {
+                            try
+                            {
+                                ZooKeeper zkClient = client.getZookeeperClient().getZooKeeper();    
+                                
+                                if(watcher == null)
+                                {
+                                    zkClient.removeAllWatches(path, watcherType, local);    
+                                }
+                                else
+                                {
+                                    zkClient.removeWatches(path, watcher, watcherType, local);
+                                }
+                            }
+                            catch(Exception e)
+                            {
+                                if( RetryLoop.isRetryException(e) && guaranteed )
+                                {
+                                    //Setup the guaranteed handler
+                                    client.getFailedRemoveWatcherManager().addFailedOperation(new FailedRemoveWatchManager.FailedRemoveWatchDetails(path, watcher));
+                                    throw e;
+                                }
+                                else if(e instanceof KeeperException.NoWatcherException && quietly)
+                                {
+                                    //Ignore
+                                }
+                                else
+                                {
+                                    //Rethrow
+                                    throw e;
+                                }
+                            }
+                     
+                            return null;
+                        }
+            });
+        }
+    }
+    
+    @Override
+    public void performBackgroundOperation(final OperationAndData<String> operationAndData)
+            throws Exception
+    {
+        final TimeTrace   trace = client.getZookeeperClient().startTracer("RemoteWatches-Background");
+        
+        AsyncCallback.VoidCallback callback = new AsyncCallback.VoidCallback()
+        {
+            @Override
+            public void processResult(int rc, String path, Object ctx)
+            {
+                trace.commit();
+                CuratorEvent event = new CuratorEventImpl(client, CuratorEventType.REMOVE_WATCHES, rc, path, null, ctx, null, null, null, null, null);
+                client.processBackgroundOperation(operationAndData, event);                
+            }
+        };
+        
+        ZooKeeper zkClient = client.getZooKeeper();
+        if(watcher == null)
+        {
+            zkClient.removeAllWatches(operationAndData.getData(), watcherType, local, callback, operationAndData.getContext());    
+        }
+        else
+        {
+            zkClient.removeWatches(operationAndData.getData(), watcher, watcherType, local, callback, operationAndData.getContext());
+        }
+        
+    }
+}
\ No newline at end of file
diff --git a/curator-framework/src/main/java/org/apache/curator/framework/imps/WatcherRemovalFacade.java b/curator-framework/src/main/java/org/apache/curator/framework/imps/WatcherRemovalFacade.java
new file mode 100644
index 0000000..156341e
--- /dev/null
+++ b/curator-framework/src/main/java/org/apache/curator/framework/imps/WatcherRemovalFacade.java
@@ -0,0 +1,176 @@
+/**
+ * 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.curator.framework.imps;
+
+import org.apache.curator.CuratorZookeeperClient;
+import org.apache.curator.RetryLoop;
+import org.apache.curator.framework.CuratorFramework;
+import org.apache.curator.framework.WatcherRemoveCuratorFramework;
+import org.apache.curator.framework.api.CuratorEvent;
+import org.apache.curator.framework.api.CuratorListener;
+import org.apache.curator.framework.api.UnhandledErrorListener;
+import org.apache.curator.framework.listen.Listenable;
+import org.apache.curator.framework.state.ConnectionStateListener;
+import org.apache.curator.utils.EnsurePath;
+import org.apache.zookeeper.ZooKeeper;
+
+class WatcherRemovalFacade extends CuratorFrameworkImpl implements WatcherRemoveCuratorFramework
+{
+    private final CuratorFrameworkImpl client;
+    private final WatcherRemovalManager removalManager;
+
+    WatcherRemovalFacade(CuratorFrameworkImpl client)
+    {
+        super(client);
+        this.client = client;
+        removalManager = new WatcherRemovalManager(client);
+    }
+
+    @Override
+    public WatcherRemoveCuratorFramework newWatcherRemoveCuratorFramework()
+    {
+        return client.newWatcherRemoveCuratorFramework();
+    }
+
+    WatcherRemovalManager getRemovalManager()
+    {
+        return removalManager;
+    }
+
+    @Override
+    public void removeWatchers()
+    {
+        removalManager.removeWatchers();
+    }
+
+    @Override
+    WatcherRemovalManager getWatcherRemovalManager()
+    {
+        return removalManager;
+    }
+
+    @Override
+    public CuratorFramework nonNamespaceView()
+    {
+        return client.nonNamespaceView();
+    }
+
+    @Override
+    public CuratorFramework usingNamespace(String newNamespace)
+    {
+        return client.usingNamespace(newNamespace);
+    }
+
+    @Override
+    public String getNamespace()
+    {
+        return client.getNamespace();
+    }
+
+    @Override
+    public void start()
+    {
+        throw new UnsupportedOperationException();
+    }
+
+    @Override
+    public void close()
+    {
+        throw new UnsupportedOperationException();
+    }
+
+    @Override
+    public Listenable<ConnectionStateListener> getConnectionStateListenable()
+    {
+        return client.getConnectionStateListenable();
+    }
+
+    @Override
+    public Listenable<CuratorListener> getCuratorListenable()
+    {
+        return client.getCuratorListenable();
+    }
+
+    @Override
+    public Listenable<UnhandledErrorListener> getUnhandledErrorListenable()
+    {
+        return client.getUnhandledErrorListenable();
+    }
+
+    @Override
+    public void sync(String path, Object context)
+    {
+        client.sync(path, context);
+    }
+
+    @Override
+    public CuratorZookeeperClient getZookeeperClient()
+    {
+        return client.getZookeeperClient();
+    }
+
+    @Override
+    RetryLoop newRetryLoop()
+    {
+        return client.newRetryLoop();
+    }
+
+    @Override
+    ZooKeeper getZooKeeper() throws Exception
+    {
+        return client.getZooKeeper();
+    }
+
+    @Override
+    <DATA_TYPE> void processBackgroundOperation(OperationAndData<DATA_TYPE> operationAndData, CuratorEvent event)
+    {
+        client.processBackgroundOperation(operationAndData, event);
+    }
+
+    @Override
+    void logError(String reason, Throwable e)
+    {
+        client.logError(reason, e);
+    }
+
+    @Override
+    String unfixForNamespace(String path)
+    {
+        return client.unfixForNamespace(path);
+    }
+
+    @Override
+    String fixForNamespace(String path)
+    {
+        return client.fixForNamespace(path);
+    }
+
+    @Override
+    public EnsurePath newNamespaceAwareEnsurePath(String path)
+    {
+        return client.newNamespaceAwareEnsurePath(path);
+    }
+
+    @Override
+    FailedDeleteManager getFailedDeleteManager()
+    {
+        return client.getFailedDeleteManager();
+    }
+}
diff --git a/curator-framework/src/main/java/org/apache/curator/framework/imps/WatcherRemovalManager.java b/curator-framework/src/main/java/org/apache/curator/framework/imps/WatcherRemovalManager.java
new file mode 100644
index 0000000..a691a94
--- /dev/null
+++ b/curator-framework/src/main/java/org/apache/curator/framework/imps/WatcherRemovalManager.java
@@ -0,0 +1,137 @@
+/**
+ * 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.curator.framework.imps;
+
+import com.google.common.annotations.VisibleForTesting;
+import com.google.common.base.Preconditions;
+import com.google.common.collect.Sets;
+import org.apache.zookeeper.WatchedEvent;
+import org.apache.zookeeper.Watcher;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import java.util.HashSet;
+import java.util.Set;
+
+public class WatcherRemovalManager
+{
+    private final Logger log = LoggerFactory.getLogger(getClass());
+    private final CuratorFrameworkImpl client;
+    private final Set<WrappedWatcher> entries = Sets.newHashSet();  // guarded by sync
+
+    WatcherRemovalManager(CuratorFrameworkImpl client)
+    {
+        this.client = client;
+    }
+
+    synchronized Watcher add(String path, Watcher watcher)
+    {
+        path = Preconditions.checkNotNull(path, "path cannot be null");
+        watcher = Preconditions.checkNotNull(watcher, "watcher cannot be null");
+
+        WrappedWatcher wrappedWatcher = new WrappedWatcher(watcher, path);
+        entries.add(wrappedWatcher);
+        return wrappedWatcher;
+    }
+
+    @VisibleForTesting
+    synchronized Set<? extends Watcher> getEntries()
+    {
+        return Sets.newHashSet(entries);
+    }
+
+    void removeWatchers()
+    {
+        HashSet<WrappedWatcher> localEntries;
+        synchronized(this)
+        {
+            localEntries = Sets.newHashSet(entries);
+        }
+        for ( WrappedWatcher entry : localEntries )
+        {
+            try
+            {
+                log.debug("Removing watcher for path: " + entry.path);
+                RemoveWatchesBuilderImpl builder = new RemoveWatchesBuilderImpl(client);
+                builder.internalRemoval(entry, entry.path);
+            }
+            catch ( Exception e )
+            {
+                log.error("Could not remove watcher for path: " + entry.path);
+            }
+        }
+    }
+
+    private synchronized void internalRemove(WrappedWatcher entry)
+    {
+        entries.remove(entry);
+    }
+
+    private class WrappedWatcher implements Watcher
+    {
+        private final Watcher watcher;
+        private final String path;
+
+        WrappedWatcher(Watcher watcher, String path)
+        {
+            this.watcher = watcher;
+            this.path = path;
+        }
+
+        @Override
+        public void process(WatchedEvent event)
+        {
+            if ( event.getType() != Event.EventType.None )
+            {
+                internalRemove(this);
+            }
+            watcher.process(event);
+        }
+
+        @Override
+        public boolean equals(Object o)
+        {
+            if ( this == o )
+            {
+                return true;
+            }
+            if ( o == null || getClass() != o.getClass() )
+            {
+                return false;
+            }
+
+            WrappedWatcher entry = (WrappedWatcher)o;
+
+            //noinspection SimplifiableIfStatement
+            if ( !watcher.equals(entry.watcher) )
+            {
+                return false;
+            }
+            return path.equals(entry.path);
+
+        }
+
+        @Override
+        public int hashCode()
+        {
+            int result = watcher.hashCode();
+            result = 31 * result + path.hashCode();
+            return result;
+        }
+    }
+}
diff --git a/curator-framework/src/main/java/org/apache/curator/framework/imps/Watching.java b/curator-framework/src/main/java/org/apache/curator/framework/imps/Watching.java
index a9d0ab1..4bebbd5 100644
--- a/curator-framework/src/main/java/org/apache/curator/framework/imps/Watching.java
+++ b/curator-framework/src/main/java/org/apache/curator/framework/imps/Watching.java
@@ -50,8 +50,12 @@
         watched = false;
     }
 
-    Watcher getWatcher()
+    Watcher getWatcher(CuratorFrameworkImpl client, String unfixedPath)
     {
+        if ( (watcher != null) && (client.getWatcherRemovalManager() != null) )
+        {
+            return client.getWatcherRemovalManager().add(unfixedPath, watcher);
+        }
         return watcher;
     }
 
diff --git a/curator-framework/src/test/java/org/apache/curator/framework/imps/TestFailedDeleteManager.java b/curator-framework/src/test/java/org/apache/curator/framework/imps/TestFailedDeleteManager.java
index 6599745..943529f 100644
--- a/curator-framework/src/test/java/org/apache/curator/framework/imps/TestFailedDeleteManager.java
+++ b/curator-framework/src/test/java/org/apache/curator/framework/imps/TestFailedDeleteManager.java
@@ -22,7 +22,6 @@
 import org.apache.curator.framework.CuratorFrameworkFactory;
 import org.apache.curator.framework.api.BackgroundCallback;
 import org.apache.curator.framework.api.CuratorEvent;
-import org.apache.curator.framework.imps.FailedDeleteManager.FailedDeleteManagerListener;
 import org.apache.curator.framework.state.ConnectionState;
 import org.apache.curator.framework.state.ConnectionStateListener;
 import org.apache.curator.retry.ExponentialBackoffRetry;
@@ -291,11 +290,11 @@
         
         final AtomicBoolean pathAdded = new AtomicBoolean(false);
         
-        ((CuratorFrameworkImpl)client).getFailedDeleteManager().debugListener = new FailedDeleteManagerListener()
+        ((CuratorFrameworkImpl)client).getFailedDeleteManager().debugListener = new FailedOperationManager.FailedOperationManagerListener<String>()
         {
             
             @Override
-            public void pathAddedForDelete(String path)
+            public void pathAddedForGuaranteedOperation(String path)
             {
                 pathAdded.set(true);
             }
@@ -325,11 +324,11 @@
         
         final AtomicBoolean pathAdded = new AtomicBoolean(false);
         
-        ((CuratorFrameworkImpl)client).getFailedDeleteManager().debugListener = new FailedDeleteManagerListener()
+        ((CuratorFrameworkImpl)client).getFailedDeleteManager().debugListener = new FailedOperationManager.FailedOperationManagerListener<String>()
         {
             
             @Override
-            public void pathAddedForDelete(String path)
+            public void pathAddedForGuaranteedOperation(String path)
             {
                 pathAdded.set(true);
             }
diff --git a/curator-framework/src/test/java/org/apache/curator/framework/imps/TestRemoveWatches.java b/curator-framework/src/test/java/org/apache/curator/framework/imps/TestRemoveWatches.java
new file mode 100644
index 0000000..4e02e95
--- /dev/null
+++ b/curator-framework/src/test/java/org/apache/curator/framework/imps/TestRemoveWatches.java
@@ -0,0 +1,655 @@
+/**
+ * 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.curator.framework.imps;
+
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.Semaphore;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicReference;
+
+import org.apache.curator.framework.CuratorFramework;
+import org.apache.curator.framework.CuratorFrameworkFactory;
+import org.apache.curator.framework.api.BackgroundCallback;
+import org.apache.curator.framework.api.CuratorEvent;
+import org.apache.curator.framework.api.CuratorEventType;
+import org.apache.curator.framework.api.CuratorListener;
+import org.apache.curator.framework.api.CuratorWatcher;
+import org.apache.curator.framework.imps.FailedRemoveWatchManager.FailedRemoveWatchDetails;
+import org.apache.curator.framework.state.ConnectionState;
+import org.apache.curator.framework.state.ConnectionStateListener;
+import org.apache.curator.retry.ExponentialBackoffRetry;
+import org.apache.curator.retry.RetryOneTime;
+import org.apache.curator.test.BaseClassForTests;
+import org.apache.curator.test.Timing;
+import org.apache.curator.utils.CloseableUtils;
+import org.apache.zookeeper.KeeperException;
+import org.apache.zookeeper.WatchedEvent;
+import org.apache.zookeeper.Watcher;
+import org.apache.zookeeper.Watcher.Event.EventType;
+import org.apache.zookeeper.Watcher.WatcherType;
+import org.testng.Assert;
+import org.testng.annotations.Test;
+
+public class TestRemoveWatches extends BaseClassForTests
+{
+    private AtomicReference<ConnectionState> registerConnectionStateListener(CuratorFramework client)
+    {
+        final AtomicReference<ConnectionState> state = new AtomicReference<ConnectionState>();
+        client.getConnectionStateListenable().addListener(new ConnectionStateListener()
+        {
+            
+            @Override
+            public void stateChanged(CuratorFramework client, ConnectionState newState)
+            {
+                state.set(newState);
+                synchronized(state)
+                {
+                    state.notify();
+                }
+            }
+        });
+        
+        return state;
+    }
+    
+    private boolean blockUntilDesiredConnectionState(AtomicReference<ConnectionState> stateRef, Timing timing, final ConnectionState desiredState)
+    {
+        if(stateRef.get() == desiredState)
+        {
+            return true;
+        }
+        
+        synchronized(stateRef)
+        {
+            if(stateRef.get() == desiredState)
+            {
+                return true;
+            }
+            
+            try
+            {
+                stateRef.wait(timing.milliseconds());
+                return stateRef.get() == desiredState;
+            }
+            catch(InterruptedException e)
+            {
+                Thread.currentThread().interrupt();
+                return false;
+            }
+        }
+    }
+    
+    @Test
+    public void testRemoveCuratorDefaultWatcher() throws Exception
+    {
+        Timing timing = new Timing();
+        CuratorFramework client = CuratorFrameworkFactory.builder().
+                connectString(server.getConnectString()).
+                retryPolicy(new RetryOneTime(1)).
+                build();
+        try
+        {
+            client.start();
+            
+            final CountDownLatch removedLatch = new CountDownLatch(1);
+            
+            final String path = "/";            
+            client.getCuratorListenable().addListener(new CuratorListener()
+            {                
+                @Override
+                public void eventReceived(CuratorFramework client, CuratorEvent event)
+                        throws Exception
+                {
+                    if(event.getType() == CuratorEventType.WATCHED && event.getWatchedEvent().getType() == EventType.DataWatchRemoved) {                        
+                        removedLatch.countDown();
+                    }        
+                }
+            });
+                        
+            client.checkExists().watched().forPath(path);
+            
+            client.watches().removeAll().forPath(path);
+            
+            Assert.assertTrue(timing.awaitLatch(removedLatch), "Timed out waiting for watch removal");
+        }
+        finally
+        {
+            CloseableUtils.closeQuietly(client);
+        }
+    }
+    
+    @Test
+    public void testRemoveCuratorWatch() throws Exception
+    {       
+        Timing timing = new Timing();
+        CuratorFramework client = CuratorFrameworkFactory.builder().
+                connectString(server.getConnectString()).
+                retryPolicy(new RetryOneTime(1)).
+                build();
+        try
+        {
+            client.start();
+            
+            final CountDownLatch removedLatch = new CountDownLatch(1);
+            
+            final String path = "/";            
+            CuratorWatcher watcher = new CuratorWatcher()
+            {
+                
+                @Override
+                public void process(WatchedEvent event) throws Exception
+                {
+                    if(event.getPath().equals(path) && event.getType() == EventType.DataWatchRemoved) {
+                        removedLatch.countDown();
+                    }
+                }
+            };
+                        
+            client.checkExists().usingWatcher(watcher).forPath(path);
+            
+            client.watches().remove(watcher).forPath(path);
+            
+            Assert.assertTrue(timing.awaitLatch(removedLatch), "Timed out waiting for watch removal");
+        }
+        finally
+        {
+            CloseableUtils.closeQuietly(client);
+        }
+    }    
+    
+    @Test
+    public void testRemoveWatch() throws Exception
+    {       
+        Timing timing = new Timing();
+        CuratorFramework client = CuratorFrameworkFactory.builder().
+                connectString(server.getConnectString()).
+                retryPolicy(new RetryOneTime(1)).
+                build();
+        try
+        {
+            client.start();
+            
+            final CountDownLatch removedLatch = new CountDownLatch(1);
+            
+            final String path = "/";    
+            Watcher watcher = new CountDownWatcher(path, removedLatch, EventType.DataWatchRemoved);
+            
+            client.checkExists().usingWatcher(watcher).forPath(path);
+            
+            client.watches().remove(watcher).forPath(path);
+            
+            Assert.assertTrue(timing.awaitLatch(removedLatch), "Timed out waiting for watch removal");
+        }
+        finally
+        {
+            CloseableUtils.closeQuietly(client);
+        }
+    }
+    
+    @Test
+    public void testRemoveWatchInBackgroundWithCallback() throws Exception
+    {       
+        Timing timing = new Timing();
+        CuratorFramework client = CuratorFrameworkFactory.builder().
+                connectString(server.getConnectString()).
+                retryPolicy(new RetryOneTime(1)).
+                build();
+        try
+        {            
+            client.start();
+         
+            //Make sure that the event fires on both the watcher and the callback.
+            final CountDownLatch removedLatch = new CountDownLatch(2);
+            final String path = "/";
+            Watcher watcher = new CountDownWatcher(path, removedLatch, EventType.DataWatchRemoved);
+            
+            BackgroundCallback callback = new BackgroundCallback()
+            {
+                
+                @Override
+                public void processResult(CuratorFramework client, CuratorEvent event)
+                        throws Exception
+                {
+                    if(event.getType() == CuratorEventType.REMOVE_WATCHES && event.getPath().equals(path)) {
+                        removedLatch.countDown();
+                    }
+                }
+            };
+            
+            
+            client.checkExists().usingWatcher(watcher).forPath(path);
+            
+            client.watches().remove(watcher).ofType(WatcherType.Any).inBackground(callback).forPath(path);
+            
+            Assert.assertTrue(timing.awaitLatch(removedLatch), "Timed out waiting for watch removal");
+            
+        }
+        finally
+        {
+            CloseableUtils.closeQuietly(client);
+        }
+    }
+    
+    @Test
+    public void testRemoveWatchInBackgroundWithNoCallback() throws Exception
+    {       
+        Timing timing = new Timing();
+        CuratorFramework client = CuratorFrameworkFactory.builder().
+                connectString(server.getConnectString()).
+                retryPolicy(new RetryOneTime(1)).
+                build();
+        try
+        {
+            client.start();
+            
+            final String path = "/";
+            final CountDownLatch removedLatch = new CountDownLatch(1);
+            Watcher watcher = new CountDownWatcher(path, removedLatch, EventType.DataWatchRemoved);
+            
+            client.checkExists().usingWatcher(watcher).forPath(path);
+            
+            client.watches().remove(watcher).inBackground().forPath(path);
+            
+            Assert.assertTrue(timing.awaitLatch(removedLatch), "Timed out waiting for watch removal");
+            
+        }
+        finally
+        {
+            CloseableUtils.closeQuietly(client);
+        }
+    }        
+    
+    @Test
+    public void testRemoveAllWatches() throws Exception
+    {       
+        Timing timing = new Timing();
+        CuratorFramework client = CuratorFrameworkFactory.builder().
+                connectString(server.getConnectString()).
+                retryPolicy(new RetryOneTime(1)).
+                build();
+        try
+        {
+            client.start();
+            
+            final String path = "/";
+            final CountDownLatch removedLatch = new CountDownLatch(2);
+            
+            Watcher watcher1 = new CountDownWatcher(path, removedLatch, EventType.ChildWatchRemoved);            
+            Watcher watcher2 = new CountDownWatcher(path, removedLatch, EventType.DataWatchRemoved);                        
+            
+            client.getChildren().usingWatcher(watcher1).forPath(path);
+            client.checkExists().usingWatcher(watcher2).forPath(path);
+            
+            client.watches().removeAll().forPath(path);
+            
+            Assert.assertTrue(timing.awaitLatch(removedLatch), "Timed out waiting for watch removal");
+        }
+        finally
+        {
+            CloseableUtils.closeQuietly(client);
+        }
+    }  
+    
+    @Test
+    public void testRemoveAllDataWatches() throws Exception
+    {       
+        Timing timing = new Timing();
+        CuratorFramework client = CuratorFrameworkFactory.builder().
+                connectString(server.getConnectString()).
+                retryPolicy(new RetryOneTime(1)).
+                build();
+        try
+        {
+            client.start();
+            
+            final String path = "/";
+            final AtomicBoolean removedFlag = new AtomicBoolean(false);
+            final CountDownLatch removedLatch = new CountDownLatch(1);
+            
+            Watcher watcher1 = new BooleanWatcher(path, removedFlag, EventType.ChildWatchRemoved);            
+            Watcher watcher2 = new CountDownWatcher(path, removedLatch, EventType.DataWatchRemoved);                        
+            
+            client.getChildren().usingWatcher(watcher1).forPath(path);
+            client.checkExists().usingWatcher(watcher2).forPath(path);
+            
+            client.watches().removeAll().ofType(WatcherType.Data).forPath(path);
+            
+            Assert.assertTrue(timing.awaitLatch(removedLatch), "Timed out waiting for watch removal");
+            Assert.assertEquals(removedFlag.get(), false);
+        }
+        finally
+        {
+            CloseableUtils.closeQuietly(client);
+        }
+    }
+    
+    @Test
+    public void testRemoveAllChildWatches() throws Exception
+    {       
+        Timing timing = new Timing();
+        CuratorFramework client = CuratorFrameworkFactory.builder().
+                connectString(server.getConnectString()).
+                retryPolicy(new RetryOneTime(1)).
+                build();
+        try
+        {
+            client.start();
+            
+            final String path = "/";
+            final AtomicBoolean removedFlag = new AtomicBoolean(false);
+            final CountDownLatch removedLatch = new CountDownLatch(1);
+            
+            Watcher watcher1 = new BooleanWatcher(path, removedFlag, EventType.DataWatchRemoved);            
+            Watcher watcher2 = new CountDownWatcher(path, removedLatch, EventType.ChildWatchRemoved);                        
+                        
+            client.checkExists().usingWatcher(watcher1).forPath(path);
+            client.getChildren().usingWatcher(watcher2).forPath(path);
+            
+            client.watches().removeAll().ofType(WatcherType.Children).forPath(path);
+            
+            Assert.assertTrue(timing.awaitLatch(removedLatch), "Timed out waiting for watch removal");
+            Assert.assertEquals(removedFlag.get(), false);
+        }
+        finally
+        {
+            CloseableUtils.closeQuietly(client);
+        }
+    }     
+    
+    @Test
+    public void testRemoveLocalWatch() throws Exception {
+        Timing timing = new Timing();
+        CuratorFramework client = CuratorFrameworkFactory.builder().
+                connectString(server.getConnectString()).
+                retryPolicy(new RetryOneTime(1)).
+                build();
+        try
+        {
+            client.start();
+            
+            AtomicReference<ConnectionState> stateRef = registerConnectionStateListener(client);
+            
+            final String path = "/";
+            
+            final CountDownLatch removedLatch = new CountDownLatch(1);
+            
+            Watcher watcher = new CountDownWatcher(path, removedLatch, EventType.DataWatchRemoved);        
+            
+            client.checkExists().usingWatcher(watcher).forPath(path);
+            
+            //Stop the server so we can check if we can remove watches locally when offline
+            server.stop();
+            
+            Assert.assertTrue(blockUntilDesiredConnectionState(stateRef, timing, ConnectionState.SUSPENDED));
+                       
+            client.watches().removeAll().locally().forPath(path);
+            
+            Assert.assertTrue(timing.awaitLatch(removedLatch), "Timed out waiting for watch removal");
+        }
+        finally
+        {
+            CloseableUtils.closeQuietly(client);
+        }
+    }
+    
+    @Test
+    public void testRemoveLocalWatchInBackground() throws Exception {
+        Timing timing = new Timing();
+        CuratorFramework client = CuratorFrameworkFactory.builder().
+                connectString(server.getConnectString()).
+                retryPolicy(new RetryOneTime(1)).
+                build();
+        try
+        {
+            client.start();
+            
+            AtomicReference<ConnectionState> stateRef = registerConnectionStateListener(client);
+            
+            final String path = "/";
+            
+            final CountDownLatch removedLatch = new CountDownLatch(1);
+            
+            Watcher watcher = new CountDownWatcher(path, removedLatch, EventType.DataWatchRemoved);        
+            
+            client.checkExists().usingWatcher(watcher).forPath(path);
+            
+            //Stop the server so we can check if we can remove watches locally when offline
+            server.stop();
+            
+            Assert.assertTrue(blockUntilDesiredConnectionState(stateRef, timing, ConnectionState.SUSPENDED));
+                       
+            client.watches().removeAll().locally().inBackground().forPath(path);
+            
+            Assert.assertTrue(timing.awaitLatch(removedLatch), "Timed out waiting for watch removal");
+        }
+        finally
+        {
+            CloseableUtils.closeQuietly(client);
+        }
+    }    
+    
+    /**
+     * Test the case where we try and remove an unregistered watcher. In this case we expect a NoWatcherException to
+     * be thrown. 
+     * @throws Exception
+     */
+    @Test(expectedExceptions=KeeperException.NoWatcherException.class)
+    public void testRemoveUnregisteredWatcher() throws Exception
+    {
+        CuratorFramework client = CuratorFrameworkFactory.builder().
+                connectString(server.getConnectString()).
+                retryPolicy(new RetryOneTime(1)).
+                build();
+        try
+        {
+            client.start();
+            
+            final String path = "/";            
+            Watcher watcher = new Watcher() {
+                @Override
+                public void process(WatchedEvent event)
+                {
+                }                
+            };
+            
+            client.watches().remove(watcher).forPath(path);
+        }
+        finally
+        {
+            CloseableUtils.closeQuietly(client);
+        }
+    }
+    
+    /**
+     * Test the case where we try and remove an unregistered watcher but have the quietly flag set. In this case we expect success. 
+     * @throws Exception
+     */
+    @Test
+    public void testRemoveUnregisteredWatcherQuietly() throws Exception
+    {
+        Timing timing = new Timing();
+        CuratorFramework client = CuratorFrameworkFactory.builder().
+                connectString(server.getConnectString()).
+                retryPolicy(new RetryOneTime(1)).
+                build();
+        try
+        {
+            client.start();
+            
+            final AtomicBoolean watcherRemoved = new AtomicBoolean(false);
+            
+            final String path = "/";            
+            Watcher watcher = new BooleanWatcher(path, watcherRemoved, EventType.DataWatchRemoved);
+            
+            client.watches().remove(watcher).quietly().forPath(path);
+            
+            timing.sleepABit();
+            
+            //There should be no watcher removed as none were registered.
+            Assert.assertEquals(watcherRemoved.get(), false);
+        }
+        finally
+        {
+            CloseableUtils.closeQuietly(client);
+        }
+    }
+    
+    @Test
+    public void testGuaranteedRemoveWatch() throws Exception {
+        Timing timing = new Timing();
+        CuratorFramework client = CuratorFrameworkFactory.builder().
+                connectString(server.getConnectString()).
+                retryPolicy(new RetryOneTime(1)).
+                build();
+        try
+        {
+            client.start();
+            
+            AtomicReference<ConnectionState> stateRef = registerConnectionStateListener(client);
+                       
+            String path = "/";
+            
+            CountDownLatch removeLatch = new CountDownLatch(1);
+            
+            Watcher watcher = new CountDownWatcher(path, removeLatch, EventType.DataWatchRemoved);            
+            client.checkExists().usingWatcher(watcher).forPath(path);
+            
+            server.stop();           
+            
+            Assert.assertTrue(blockUntilDesiredConnectionState(stateRef, timing, ConnectionState.SUSPENDED));
+            
+            //Remove the watch while we're not connected
+            try 
+            {
+                client.watches().remove(watcher).guaranteed().forPath(path);
+                Assert.fail();
+            }
+            catch(KeeperException.ConnectionLossException e)
+            {
+                //Expected
+            }
+            
+            server.restart();
+            
+            timing.awaitLatch(removeLatch);            
+        }
+        finally
+        {
+            CloseableUtils.closeQuietly(client);
+        }
+    }
+    
+    @Test
+    public void testGuaranteedRemoveWatchInBackground() throws Exception {
+        Timing timing = new Timing();
+        CuratorFramework client = CuratorFrameworkFactory.newClient(server.getConnectString(), timing.session(), timing.connection(),
+                                                                    new ExponentialBackoffRetry(100, 3));
+        try
+        {
+            client.start();
+            
+            AtomicReference<ConnectionState> stateRef = registerConnectionStateListener(client);
+                        
+            final CountDownLatch guaranteeAddedLatch = new CountDownLatch(1);
+            
+            ((CuratorFrameworkImpl)client).getFailedRemoveWatcherManager().debugListener = new FailedOperationManager.FailedOperationManagerListener<FailedRemoveWatchManager.FailedRemoveWatchDetails>()
+            {
+
+                @Override
+                public void pathAddedForGuaranteedOperation(
+                        FailedRemoveWatchDetails detail)
+                {
+                    guaranteeAddedLatch.countDown();
+                }
+            };
+            
+            String path = "/";
+            
+            CountDownLatch removeLatch = new CountDownLatch(1);
+            
+            Watcher watcher = new CountDownWatcher(path, removeLatch, EventType.DataWatchRemoved);            
+            client.checkExists().usingWatcher(watcher).forPath(path);
+            
+            server.stop();           
+            Assert.assertTrue(blockUntilDesiredConnectionState(stateRef, timing, ConnectionState.SUSPENDED));
+            
+            //Remove the watch while we're not connected
+            client.watches().remove(watcher).guaranteed().inBackground().forPath(path);
+            
+            timing.awaitLatch(guaranteeAddedLatch);
+            
+            server.restart();
+            
+            timing.awaitLatch(removeLatch);            
+        }
+        finally
+        {
+            CloseableUtils.closeQuietly(client);
+        }
+    }    
+    
+    private static class CountDownWatcher implements Watcher {
+        private String path;
+        private EventType eventType;
+        private CountDownLatch removeLatch;
+        
+        public CountDownWatcher(String path, CountDownLatch removeLatch, EventType eventType) {
+            this.path = path;
+            this.eventType = eventType;
+            this.removeLatch = removeLatch;            
+        }
+        
+        @Override
+        public void process(WatchedEvent event)
+        {
+            if(event.getPath() == null || event.getType() == null) {
+                return;
+            }
+            
+            if(event.getPath().equals(path) && event.getType() == eventType) {
+                removeLatch.countDown();
+            }
+        }  
+    }
+    
+    private static class BooleanWatcher implements Watcher {
+        private String path;
+        private EventType eventType;
+        private AtomicBoolean removedFlag;
+        
+        public BooleanWatcher(String path, AtomicBoolean removedFlag, EventType eventType) {
+            this.path = path;
+            this.eventType = eventType;
+            this.removedFlag = removedFlag;            
+        }
+        
+        @Override
+        public void process(WatchedEvent event)
+        {
+            if(event.getPath() == null || event.getType() == null) {
+                return;
+            }
+            
+            if(event.getPath().equals(path) && event.getType() == eventType) {
+                removedFlag.set(true);
+            }
+        }  
+    }    
+}
diff --git a/curator-framework/src/test/java/org/apache/curator/framework/imps/TestWatcherRemovalManager.java b/curator-framework/src/test/java/org/apache/curator/framework/imps/TestWatcherRemovalManager.java
new file mode 100644
index 0000000..e20c450
--- /dev/null
+++ b/curator-framework/src/test/java/org/apache/curator/framework/imps/TestWatcherRemovalManager.java
@@ -0,0 +1,257 @@
+/**
+ * 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.curator.framework.imps;
+
+import org.apache.curator.framework.CuratorFramework;
+import org.apache.curator.framework.CuratorFrameworkFactory;
+import org.apache.curator.framework.WatcherRemoveCuratorFramework;
+import org.apache.curator.retry.RetryOneTime;
+import org.apache.curator.test.BaseClassForTests;
+import org.apache.curator.test.Timing;
+import org.apache.curator.test.WatchersDebug;
+import org.apache.curator.utils.CloseableUtils;
+import org.apache.zookeeper.WatchedEvent;
+import org.apache.zookeeper.Watcher;
+import org.testng.Assert;
+import org.testng.annotations.Test;
+import java.util.List;
+import java.util.concurrent.CountDownLatch;
+
+public class TestWatcherRemovalManager extends BaseClassForTests
+{
+    @Test
+    public void testBasic() throws Exception
+    {
+        CuratorFramework client = CuratorFrameworkFactory.newClient(server.getConnectString(), new RetryOneTime(1));
+        try
+        {
+            client.start();
+            internalTryBasic(client);
+        }
+        finally
+        {
+            CloseableUtils.closeQuietly(client);
+        }
+    }
+
+    @Test
+    public void testBasicNamespace1() throws Exception
+    {
+        CuratorFramework client = CuratorFrameworkFactory.newClient(server.getConnectString(), new RetryOneTime(1));
+        try
+        {
+            client.start();
+            internalTryBasic(client.usingNamespace("foo"));
+        }
+        finally
+        {
+            CloseableUtils.closeQuietly(client);
+        }
+    }
+
+    @Test
+    public void testBasicNamespace2() throws Exception
+    {
+        CuratorFramework client = CuratorFrameworkFactory.builder()
+            .connectString(server.getConnectString())
+            .retryPolicy(new RetryOneTime(1))
+            .namespace("hey")
+            .build();
+        try
+        {
+            client.start();
+            internalTryBasic(client);
+        }
+        finally
+        {
+            CloseableUtils.closeQuietly(client);
+        }
+    }
+
+    @Test
+    public void testBasicNamespace3() throws Exception
+    {
+        CuratorFramework client = CuratorFrameworkFactory.builder()
+            .connectString(server.getConnectString())
+            .retryPolicy(new RetryOneTime(1))
+            .namespace("hey")
+            .build();
+        try
+        {
+            client.start();
+            internalTryBasic(client.usingNamespace("lakjsf"));
+        }
+        finally
+        {
+            CloseableUtils.closeQuietly(client);
+        }
+    }
+
+    @Test
+    public void testSameWatcher() throws Exception
+    {
+        CuratorFramework client = CuratorFrameworkFactory.newClient(server.getConnectString(), new RetryOneTime(1));
+        try
+        {
+            client.start();
+
+            WatcherRemovalFacade removerClient = (WatcherRemovalFacade)client.newWatcherRemoveCuratorFramework();
+
+            Watcher watcher = new Watcher()
+            {
+                @Override
+                public void process(WatchedEvent event)
+                {
+                    // NOP
+                }
+            };
+
+            removerClient.getData().usingWatcher(watcher).forPath("/");
+            Assert.assertEquals(removerClient.getRemovalManager().getEntries().size(), 1);
+            removerClient.getData().usingWatcher(watcher).forPath("/");
+            Assert.assertEquals(removerClient.getRemovalManager().getEntries().size(), 1);
+        }
+        finally
+        {
+            CloseableUtils.closeQuietly(client);
+        }
+    }
+
+    @Test
+    public void testTriggered() throws Exception
+    {
+        CuratorFramework client = CuratorFrameworkFactory.newClient(server.getConnectString(), new RetryOneTime(1));
+        try
+        {
+            client.start();
+
+            WatcherRemovalFacade removerClient = (WatcherRemovalFacade)client.newWatcherRemoveCuratorFramework();
+
+            final CountDownLatch latch = new CountDownLatch(1);
+            Watcher watcher = new Watcher()
+            {
+                @Override
+                public void process(WatchedEvent event)
+                {
+                    if ( event.getType() == Event.EventType.NodeCreated )
+                    {
+                        latch.countDown();
+                    }
+                }
+            };
+
+            removerClient.checkExists().usingWatcher(watcher).forPath("/yo");
+            Assert.assertEquals(removerClient.getRemovalManager().getEntries().size(), 1);
+            removerClient.create().forPath("/yo");
+
+            Assert.assertTrue(new Timing().awaitLatch(latch));
+
+            Assert.assertEquals(removerClient.getRemovalManager().getEntries().size(), 0);
+        }
+        finally
+        {
+            CloseableUtils.closeQuietly(client);
+        }
+    }
+
+    @Test
+    public void testResetFromWatcher() throws Exception
+    {
+        Timing timing = new Timing();
+        CuratorFramework client = CuratorFrameworkFactory.newClient(server.getConnectString(), new RetryOneTime(1));
+        try
+        {
+            client.start();
+
+            final WatcherRemovalFacade removerClient = (WatcherRemovalFacade)client.newWatcherRemoveCuratorFramework();
+
+            final CountDownLatch createdLatch = new CountDownLatch(1);
+            final CountDownLatch deletedLatch = new CountDownLatch(1);
+            Watcher watcher = new Watcher()
+            {
+                @Override
+                public void process(WatchedEvent event)
+                {
+                    if ( event.getType() == Event.EventType.NodeCreated )
+                    {
+                        try
+                        {
+                            removerClient.checkExists().usingWatcher(this).forPath("/yo");
+                        }
+                        catch ( Exception e )
+                        {
+                            e.printStackTrace();
+                        }
+                        createdLatch.countDown();
+                    }
+                    else if ( event.getType() == Event.EventType.NodeDeleted )
+                    {
+                        deletedLatch.countDown();
+                    }
+                }
+            };
+
+            removerClient.checkExists().usingWatcher(watcher).forPath("/yo");
+            Assert.assertEquals(removerClient.getRemovalManager().getEntries().size(), 1);
+            removerClient.create().forPath("/yo");
+
+            Assert.assertTrue(timing.awaitLatch(createdLatch));
+            Assert.assertEquals(removerClient.getRemovalManager().getEntries().size(), 1);
+
+            removerClient.delete().forPath("/yo");
+
+            Assert.assertTrue(timing.awaitLatch(deletedLatch));
+
+            Assert.assertEquals(removerClient.getRemovalManager().getEntries().size(), 0);
+        }
+        finally
+        {
+            CloseableUtils.closeQuietly(client);
+        }
+    }
+
+    private void internalTryBasic(CuratorFramework client) throws Exception
+    {
+        WatcherRemoveCuratorFramework removerClient = client.newWatcherRemoveCuratorFramework();
+
+        final CountDownLatch latch = new CountDownLatch(1);
+        Watcher watcher = new Watcher()
+        {
+            @Override
+            public void process(WatchedEvent event)
+            {
+                if ( event.getType() == Event.EventType.DataWatchRemoved )
+                {
+                    latch.countDown();
+                }
+            }
+        };
+        removerClient.checkExists().usingWatcher(watcher).forPath("/hey");
+
+        List<String> existWatches = WatchersDebug.getExistWatches(client.getZookeeperClient().getZooKeeper());
+        Assert.assertEquals(existWatches.size(), 1);
+
+        removerClient.removeWatchers();
+
+        Assert.assertTrue(new Timing().awaitLatch(latch));
+
+        existWatches = WatchersDebug.getExistWatches(client.getZookeeperClient().getZooKeeper());
+        Assert.assertEquals(existWatches.size(), 0);
+    }
+}
diff --git a/curator-recipes/src/main/java/org/apache/curator/framework/recipes/cache/NodeCache.java b/curator-recipes/src/main/java/org/apache/curator/framework/recipes/cache/NodeCache.java
index 72ee5ff..49b9a3f 100644
--- a/curator-recipes/src/main/java/org/apache/curator/framework/recipes/cache/NodeCache.java
+++ b/curator-recipes/src/main/java/org/apache/curator/framework/recipes/cache/NodeCache.java
@@ -23,6 +23,7 @@
 import com.google.common.base.Objects;
 import com.google.common.base.Preconditions;
 import org.apache.curator.framework.CuratorFramework;
+import org.apache.curator.framework.WatcherRemoveCuratorFramework;
 import org.apache.curator.framework.api.BackgroundCallback;
 import org.apache.curator.framework.api.CuratorEvent;
 import org.apache.curator.framework.api.CuratorWatcher;
@@ -53,7 +54,7 @@
 public class NodeCache implements Closeable
 {
     private final Logger log = LoggerFactory.getLogger(getClass());
-    private final CuratorFramework client;
+    private final WatcherRemoveCuratorFramework client;
     private final String path;
     private final boolean dataIsCompressed;
     private final AtomicReference<ChildData> data = new AtomicReference<ChildData>(null);
@@ -127,7 +128,7 @@
      */
     public NodeCache(CuratorFramework client, String path, boolean dataIsCompressed)
     {
-        this.client = client;
+        this.client = client.newWatcherRemoveCuratorFramework();
         this.path = PathUtils.validatePath(path);
         this.dataIsCompressed = dataIsCompressed;
     }
@@ -169,6 +170,7 @@
     {
         if ( state.compareAndSet(State.STARTED, State.CLOSED) )
         {
+            client.removeWatchers();
             listeners.clear();
         }
         client.getConnectionStateListenable().removeListener(connectionStateListener);
diff --git a/curator-recipes/src/main/java/org/apache/curator/framework/recipes/cache/PathChildrenCache.java b/curator-recipes/src/main/java/org/apache/curator/framework/recipes/cache/PathChildrenCache.java
index b5d912c..99a652d 100644
--- a/curator-recipes/src/main/java/org/apache/curator/framework/recipes/cache/PathChildrenCache.java
+++ b/curator-recipes/src/main/java/org/apache/curator/framework/recipes/cache/PathChildrenCache.java
@@ -27,6 +27,7 @@
 import com.google.common.collect.Maps;
 import com.google.common.collect.Sets;
 import org.apache.curator.framework.CuratorFramework;
+import org.apache.curator.framework.WatcherRemoveCuratorFramework;
 import org.apache.curator.framework.api.BackgroundCallback;
 import org.apache.curator.framework.api.CuratorEvent;
 import org.apache.curator.framework.listen.ListenerContainer;
@@ -42,16 +43,13 @@
 import org.apache.zookeeper.data.Stat;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
+
 import java.io.Closeable;
 import java.io.IOException;
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
-import java.util.concurrent.ConcurrentMap;
-import java.util.concurrent.Exchanger;
-import java.util.concurrent.ExecutorService;
-import java.util.concurrent.Executors;
-import java.util.concurrent.ThreadFactory;
+import java.util.concurrent.*;
 import java.util.concurrent.atomic.AtomicReference;
 
 /**
@@ -67,7 +65,7 @@
 public class PathChildrenCache implements Closeable
 {
     private final Logger log = LoggerFactory.getLogger(getClass());
-    private final CuratorFramework client;
+    private final WatcherRemoveCuratorFramework client;
     private final String path;
     private final CloseableExecutorService executorService;
     private final boolean cacheData;
@@ -215,7 +213,7 @@
      */
     public PathChildrenCache(CuratorFramework client, String path, boolean cacheData, boolean dataIsCompressed, final CloseableExecutorService executorService)
     {
-        this.client = client;
+        this.client = client.newWatcherRemoveCuratorFramework();
         this.path = PathUtils.validatePath(path);
         this.cacheData = cacheData;
         this.dataIsCompressed = dataIsCompressed;
@@ -374,6 +372,7 @@
             executorService.close();
             client.clearWatcherReferences(childrenWatcher);
             client.clearWatcherReferences(dataWatcher);
+            client.removeWatchers();
 
             // TODO
             // This seems to enable even more GC - I'm not sure why yet - it
diff --git a/curator-recipes/src/main/java/org/apache/curator/framework/recipes/cache/TreeCache.java b/curator-recipes/src/main/java/org/apache/curator/framework/recipes/cache/TreeCache.java
index 4f3ffb6..bda00bf 100644
--- a/curator-recipes/src/main/java/org/apache/curator/framework/recipes/cache/TreeCache.java
+++ b/curator-recipes/src/main/java/org/apache/curator/framework/recipes/cache/TreeCache.java
@@ -25,6 +25,7 @@
 import com.google.common.collect.ImmutableMap;
 import com.google.common.collect.Maps;
 import org.apache.curator.framework.CuratorFramework;
+import org.apache.curator.framework.WatcherRemoveCuratorFramework;
 import org.apache.curator.framework.api.BackgroundCallback;
 import org.apache.curator.framework.api.CuratorEvent;
 import org.apache.curator.framework.api.UnhandledErrorListener;
@@ -480,7 +481,7 @@
     private final AtomicBoolean isInitialized = new AtomicBoolean(false);
 
     private final TreeNode root;
-    private final CuratorFramework client;
+    private final WatcherRemoveCuratorFramework client;
     private final CloseableExecutorService executorService;
     private final boolean cacheData;
     private final boolean dataIsCompressed;
@@ -530,7 +531,7 @@
     {
         this.createParentNodes = createParentNodes;
         this.root = new TreeNode(validatePath(path), null);
-        this.client = client;
+        this.client = client.newWatcherRemoveCuratorFramework();
         this.cacheData = cacheData;
         this.dataIsCompressed = dataIsCompressed;
         this.maxDepth = maxDepth;
@@ -566,6 +567,7 @@
     {
         if ( treeState.compareAndSet(TreeState.STARTED, TreeState.CLOSED) )
         {
+            client.removeWatchers();
             client.getConnectionStateListenable().removeListener(connectionStateListener);
             listeners.clear();
             executorService.close();
diff --git a/curator-recipes/src/main/java/org/apache/curator/framework/recipes/leader/LeaderLatch.java b/curator-recipes/src/main/java/org/apache/curator/framework/recipes/leader/LeaderLatch.java
index a6d8145..da9b8b2 100644
--- a/curator-recipes/src/main/java/org/apache/curator/framework/recipes/leader/LeaderLatch.java
+++ b/curator-recipes/src/main/java/org/apache/curator/framework/recipes/leader/LeaderLatch.java
@@ -23,6 +23,7 @@
 import com.google.common.base.Function;
 import com.google.common.base.Preconditions;
 import org.apache.curator.framework.CuratorFramework;
+import org.apache.curator.framework.WatcherRemoveCuratorFramework;
 import org.apache.curator.framework.api.BackgroundCallback;
 import org.apache.curator.framework.api.CuratorEvent;
 import org.apache.curator.framework.listen.ListenerContainer;
@@ -63,7 +64,7 @@
 public class LeaderLatch implements Closeable
 {
     private final Logger log = LoggerFactory.getLogger(getClass());
-    private final CuratorFramework client;
+    private final WatcherRemoveCuratorFramework client;
     private final String latchPath;
     private final String id;
     private final AtomicReference<State> state = new AtomicReference<State>(State.LATENT);
@@ -143,7 +144,7 @@
      */
     public LeaderLatch(CuratorFramework client, String latchPath, String id, CloseMode closeMode)
     {
-        this.client = Preconditions.checkNotNull(client, "client cannot be null");
+        this.client = Preconditions.checkNotNull(client, "client cannot be null").newWatcherRemoveCuratorFramework();
         this.latchPath = PathUtils.validatePath(latchPath);
         this.id = Preconditions.checkNotNull(id, "id cannot be null");
         this.closeMode = Preconditions.checkNotNull(closeMode, "closeMode cannot be null");
@@ -206,6 +207,7 @@
         try
         {
             setNode(null);
+            client.removeWatchers();
         }
         catch ( Exception e )
         {
diff --git a/curator-recipes/src/main/java/org/apache/curator/framework/recipes/locks/InterProcessSemaphoreMutex.java b/curator-recipes/src/main/java/org/apache/curator/framework/recipes/locks/InterProcessSemaphoreMutex.java
index 88b5f5d..444b10d 100644
--- a/curator-recipes/src/main/java/org/apache/curator/framework/recipes/locks/InterProcessSemaphoreMutex.java
+++ b/curator-recipes/src/main/java/org/apache/curator/framework/recipes/locks/InterProcessSemaphoreMutex.java
@@ -20,6 +20,7 @@
 
 import com.google.common.base.Preconditions;
 import org.apache.curator.framework.CuratorFramework;
+import org.apache.curator.framework.WatcherRemoveCuratorFramework;
 import java.util.concurrent.TimeUnit;
 
 /**
@@ -29,6 +30,7 @@
 public class InterProcessSemaphoreMutex implements InterProcessLock
 {
     private final InterProcessSemaphoreV2 semaphore;
+    private final WatcherRemoveCuratorFramework watcherRemoveClient;
     private volatile Lease lease;
 
     /**
@@ -37,7 +39,8 @@
      */
     public InterProcessSemaphoreMutex(CuratorFramework client, String path)
     {
-        this.semaphore = new InterProcessSemaphoreV2(client, path, 1);
+        watcherRemoveClient = client.newWatcherRemoveCuratorFramework();
+        this.semaphore = new InterProcessSemaphoreV2(watcherRemoveClient, path, 1);
     }
 
     @Override
@@ -66,6 +69,7 @@
         try
         {
             lease.close();
+            watcherRemoveClient.removeWatchers();
         }
         finally
         {
diff --git a/curator-recipes/src/main/java/org/apache/curator/framework/recipes/locks/InterProcessSemaphoreV2.java b/curator-recipes/src/main/java/org/apache/curator/framework/recipes/locks/InterProcessSemaphoreV2.java
index f4af39b..3bf2ec3 100644
--- a/curator-recipes/src/main/java/org/apache/curator/framework/recipes/locks/InterProcessSemaphoreV2.java
+++ b/curator-recipes/src/main/java/org/apache/curator/framework/recipes/locks/InterProcessSemaphoreV2.java
@@ -22,15 +22,16 @@
 import com.google.common.base.Preconditions;
 import com.google.common.collect.ImmutableList;
 import com.google.common.collect.Sets;
-
-import org.apache.curator.utils.CloseableUtils;
 import org.apache.curator.RetryLoop;
 import org.apache.curator.framework.CuratorFramework;
+import org.apache.curator.framework.WatcherRemoveCuratorFramework;
 import org.apache.curator.framework.api.PathAndBytesable;
 import org.apache.curator.framework.imps.CuratorFrameworkState;
 import org.apache.curator.framework.recipes.shared.SharedCountListener;
 import org.apache.curator.framework.recipes.shared.SharedCountReader;
 import org.apache.curator.framework.state.ConnectionState;
+import org.apache.curator.utils.CloseableUtils;
+import org.apache.curator.utils.PathUtils;
 import org.apache.curator.utils.ZKPaths;
 import org.apache.zookeeper.CreateMode;
 import org.apache.zookeeper.KeeperException;
@@ -38,13 +39,13 @@
 import org.apache.zookeeper.Watcher;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
+
 import java.io.IOException;
 import java.util.Arrays;
 import java.util.Collection;
 import java.util.List;
 import java.util.Set;
 import java.util.concurrent.TimeUnit;
-import org.apache.curator.utils.PathUtils;
 
 /**
  * <p>
@@ -78,7 +79,7 @@
 {
     private final Logger log = LoggerFactory.getLogger(getClass());
     private final InterProcessMutex lock;
-    private final CuratorFramework client;
+    private final WatcherRemoveCuratorFramework client;
     private final String leasesPath;
     private final Watcher watcher = new Watcher()
     {
@@ -122,7 +123,7 @@
 
     private InterProcessSemaphoreV2(CuratorFramework client, String path, int maxLeases, SharedCountReader count)
     {
-        this.client = client;
+        this.client = client.newWatcherRemoveCuratorFramework();
         path = PathUtils.validatePath(path);
         lock = new InterProcessMutex(client, ZKPaths.makePath(path, LOCK_PARENT));
         this.maxLeases = (count != null) ? count.getCount() : maxLeases;
@@ -352,36 +353,43 @@
             String nodeName = ZKPaths.getNodeFromPath(path);
             builder.add(makeLease(path));
 
-            synchronized(this)
+            try
             {
-                for(;;)
+                synchronized(this)
                 {
-                    List<String> children = client.getChildren().usingWatcher(watcher).forPath(leasesPath);
-                    if ( !children.contains(nodeName) )
+                    for(;;)
                     {
-                        log.error("Sequential path not found: " + path);
-                        return InternalAcquireResult.RETRY_DUE_TO_MISSING_NODE;
-                    }
-
-                    if ( children.size() <= maxLeases )
-                    {
-                        break;
-                    }
-                    if ( hasWait )
-                    {
-                        long thisWaitMs = getThisWaitMs(startMs, waitMs);
-                        if ( thisWaitMs <= 0 )
+                        List<String> children = client.getChildren().usingWatcher(watcher).forPath(leasesPath);
+                        if ( !children.contains(nodeName) )
                         {
-                            return InternalAcquireResult.RETURN_NULL;
+                            log.error("Sequential path not found: " + path);
+                            return InternalAcquireResult.RETRY_DUE_TO_MISSING_NODE;
                         }
-                        wait(thisWaitMs);
-                    }
-                    else
-                    {
-                        wait();
+
+                        if ( children.size() <= maxLeases )
+                        {
+                            break;
+                        }
+                        if ( hasWait )
+                        {
+                            long thisWaitMs = getThisWaitMs(startMs, waitMs);
+                            if ( thisWaitMs <= 0 )
+                            {
+                                return InternalAcquireResult.RETURN_NULL;
+                            }
+                            wait(thisWaitMs);
+                        }
+                        else
+                        {
+                            wait();
+                        }
                     }
                 }
             }
+            finally
+            {
+                client.removeWatchers();
+            }
         }
         finally
         {
diff --git a/curator-recipes/src/main/java/org/apache/curator/framework/recipes/locks/LockInternals.java b/curator-recipes/src/main/java/org/apache/curator/framework/recipes/locks/LockInternals.java
index 2b4d3d9..4b0da11 100644
--- a/curator-recipes/src/main/java/org/apache/curator/framework/recipes/locks/LockInternals.java
+++ b/curator-recipes/src/main/java/org/apache/curator/framework/recipes/locks/LockInternals.java
@@ -24,11 +24,11 @@
 import com.google.common.collect.Lists;
 import org.apache.curator.RetryLoop;
 import org.apache.curator.framework.CuratorFramework;
+import org.apache.curator.framework.WatcherRemoveCuratorFramework;
 import org.apache.curator.framework.api.CuratorWatcher;
 import org.apache.curator.framework.imps.CuratorFrameworkState;
 import org.apache.curator.utils.PathUtils;
 import org.apache.curator.utils.ZKPaths;
-import org.apache.zookeeper.CreateMode;
 import org.apache.zookeeper.KeeperException;
 import org.apache.zookeeper.WatchedEvent;
 import org.apache.zookeeper.Watcher;
@@ -42,7 +42,7 @@
 
 public class LockInternals
 {
-    private final CuratorFramework                  client;
+    private final WatcherRemoveCuratorFramework     client;
     private final String                            path;
     private final String                            basePath;
     private final LockInternalsDriver               driver;
@@ -100,7 +100,7 @@
         this.lockName = lockName;
         this.maxLeases = maxLeases;
 
-        this.client = client;
+        this.client = client.newWatcherRemoveCuratorFramework();
         this.basePath = PathUtils.validatePath(path);
         this.path = ZKPaths.makePath(path, lockName);
     }
@@ -116,8 +116,9 @@
         revocable.set(entry);
     }
 
-    void releaseLock(String lockPath) throws Exception
+    final void releaseLock(String lockPath) throws Exception
     {
+        client.removeWatchers();
         revocable.set(null);
         deleteOurPath(lockPath);
     }
diff --git a/curator-recipes/src/main/java/org/apache/curator/framework/recipes/nodes/PersistentEphemeralNode.java b/curator-recipes/src/main/java/org/apache/curator/framework/recipes/nodes/PersistentEphemeralNode.java
index 0d963e0..0b482ef 100644
--- a/curator-recipes/src/main/java/org/apache/curator/framework/recipes/nodes/PersistentEphemeralNode.java
+++ b/curator-recipes/src/main/java/org/apache/curator/framework/recipes/nodes/PersistentEphemeralNode.java
@@ -22,6 +22,7 @@
 import com.google.common.base.Preconditions;
 
 import org.apache.curator.framework.CuratorFramework;
+import org.apache.curator.framework.WatcherRemoveCuratorFramework;
 import org.apache.curator.framework.api.ACLBackgroundPathAndBytesable;
 import org.apache.curator.framework.api.BackgroundCallback;
 import org.apache.curator.framework.api.CreateModable;
@@ -58,7 +59,7 @@
 {
     private final AtomicReference<CountDownLatch> initialCreateLatch = new AtomicReference<CountDownLatch>(new CountDownLatch(1));
     private final Logger log = LoggerFactory.getLogger(getClass());
-    private final CuratorFramework client;
+    private final WatcherRemoveCuratorFramework client;
     private final CreateModable<ACLBackgroundPathAndBytesable<String>> createMethod;
     private final AtomicReference<String> nodePath = new AtomicReference<String>(null);
     private final String basePath;
@@ -212,7 +213,7 @@
      */
     public PersistentEphemeralNode(CuratorFramework client, Mode mode, String basePath, byte[] initData)
     {
-        this.client = Preconditions.checkNotNull(client, "client cannot be null");
+        this.client = Preconditions.checkNotNull(client, "client cannot be null").newWatcherRemoveCuratorFramework();
         this.basePath = PathUtils.validatePath(basePath);
         this.mode = Preconditions.checkNotNull(mode, "mode cannot be null");
         final byte[] data = Preconditions.checkNotNull(initData, "data cannot be null");
@@ -314,6 +315,8 @@
         {
             throw new IOException(e);
         }
+
+        client.removeWatchers();
     }
 
     /**
diff --git a/curator-recipes/src/main/java/org/apache/curator/framework/recipes/queue/ChildrenCache.java b/curator-recipes/src/main/java/org/apache/curator/framework/recipes/queue/ChildrenCache.java
index 032dc7a..e5c7e8c 100644
--- a/curator-recipes/src/main/java/org/apache/curator/framework/recipes/queue/ChildrenCache.java
+++ b/curator-recipes/src/main/java/org/apache/curator/framework/recipes/queue/ChildrenCache.java
@@ -21,6 +21,7 @@
 import com.google.common.collect.ImmutableList;
 import com.google.common.collect.Lists;
 import org.apache.curator.framework.CuratorFramework;
+import org.apache.curator.framework.WatcherRemoveCuratorFramework;
 import org.apache.curator.framework.api.BackgroundCallback;
 import org.apache.curator.framework.api.CuratorEvent;
 import org.apache.curator.framework.api.CuratorWatcher;
@@ -36,7 +37,7 @@
 
 class ChildrenCache implements Closeable
 {
-    private final CuratorFramework client;
+    private final WatcherRemoveCuratorFramework client;
     private final String path;
     private final AtomicReference<Data> children = new AtomicReference<Data>(new Data(Lists.<String>newArrayList(), 0));
     private final AtomicBoolean isClosed = new AtomicBoolean(false);
@@ -79,7 +80,7 @@
 
     ChildrenCache(CuratorFramework client, String path)
     {
-        this.client = client;
+        this.client = client.newWatcherRemoveCuratorFramework();
         this.path = PathUtils.validatePath(path);
     }
 
@@ -91,6 +92,7 @@
     @Override
     public void close() throws IOException
     {
+        client.removeWatchers();
         isClosed.set(true);
         notifyFromCallback();
     }
diff --git a/curator-recipes/src/main/java/org/apache/curator/framework/recipes/shared/SharedValue.java b/curator-recipes/src/main/java/org/apache/curator/framework/recipes/shared/SharedValue.java
index 6ce6bf4..dcd30ba 100644
--- a/curator-recipes/src/main/java/org/apache/curator/framework/recipes/shared/SharedValue.java
+++ b/curator-recipes/src/main/java/org/apache/curator/framework/recipes/shared/SharedValue.java
@@ -22,6 +22,7 @@
 import com.google.common.base.Function;
 import com.google.common.base.Preconditions;
 import org.apache.curator.framework.CuratorFramework;
+import org.apache.curator.framework.WatcherRemoveCuratorFramework;
 import org.apache.curator.framework.api.CuratorWatcher;
 import org.apache.curator.framework.listen.ListenerContainer;
 import org.apache.curator.framework.state.ConnectionState;
@@ -45,7 +46,7 @@
 {
     private final Logger log = LoggerFactory.getLogger(getClass());
     private final ListenerContainer<SharedValueListener> listeners = new ListenerContainer<SharedValueListener>();
-    private final CuratorFramework client;
+    private final WatcherRemoveCuratorFramework client;
     private final String path;
     private final byte[] seedValue;
     private final AtomicReference<State> state = new AtomicReference<State>(State.LATENT);
@@ -87,7 +88,7 @@
      */
     public SharedValue(CuratorFramework client, String path, byte[] seedValue)
     {
-        this.client = client;
+        this.client = client.newWatcherRemoveCuratorFramework();
         this.path = PathUtils.validatePath(path);
         this.seedValue = Arrays.copyOf(seedValue, seedValue.length);
         currentValue = new AtomicReference<VersionedValue<byte[]>>(new VersionedValue<byte[]>(0, Arrays.copyOf(seedValue, seedValue.length)));
@@ -233,8 +234,9 @@
     @Override
     public void close() throws IOException
     {
-        client.getConnectionStateListenable().removeListener(connectionStateListener);
         state.set(State.CLOSED);
+        client.removeWatchers();
+        client.getConnectionStateListenable().removeListener(connectionStateListener);
         listeners.clear();
     }
 
diff --git a/curator-recipes/src/test/java/org/apache/curator/framework/imps/TestCleanState.java b/curator-recipes/src/test/java/org/apache/curator/framework/imps/TestCleanState.java
new file mode 100644
index 0000000..82de1fc
--- /dev/null
+++ b/curator-recipes/src/test/java/org/apache/curator/framework/imps/TestCleanState.java
@@ -0,0 +1,68 @@
+/**
+ * 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.curator.framework.imps;
+
+import org.apache.curator.framework.CuratorFramework;
+import org.apache.curator.test.WatchersDebug;
+import org.apache.curator.utils.CloseableUtils;
+import org.apache.zookeeper.ZooKeeper;
+
+public class TestCleanState
+{
+    public static void closeAndTestClean(CuratorFramework client)
+    {
+        if ( client == null )
+        {
+            return;
+        }
+
+        try
+        {
+            CuratorFrameworkImpl internalClient = (CuratorFrameworkImpl)client;
+            ZooKeeper zooKeeper = internalClient.getZooKeeper();
+            if ( zooKeeper != null )
+            {
+                if ( WatchersDebug.getChildWatches(zooKeeper).size() != 0 )
+                {
+                    throw new AssertionError("One or more child watchers are still registered: " + WatchersDebug.getChildWatches(zooKeeper));
+                }
+                if ( WatchersDebug.getExistWatches(zooKeeper).size() != 0 )
+                {
+                    throw new AssertionError("One or more exists watchers are still registered: " + WatchersDebug.getExistWatches(zooKeeper));
+                }
+                if ( WatchersDebug.getDataWatches(zooKeeper).size() != 0 )
+                {
+                    throw new AssertionError("One or more data watchers are still registered: " + WatchersDebug.getDataWatches(zooKeeper));
+                }
+            }
+        }
+        catch ( Exception e )
+        {
+            e.printStackTrace();    // not sure what to do here
+        }
+        finally
+        {
+            CloseableUtils.closeQuietly(client);
+        }
+    }
+
+    private TestCleanState()
+    {
+    }
+}
diff --git a/curator-recipes/src/test/java/org/apache/curator/framework/recipes/cache/BaseTestTreeCache.java b/curator-recipes/src/test/java/org/apache/curator/framework/recipes/cache/BaseTestTreeCache.java
index ab37785..f32c9b2 100644
--- a/curator-recipes/src/test/java/org/apache/curator/framework/recipes/cache/BaseTestTreeCache.java
+++ b/curator-recipes/src/test/java/org/apache/curator/framework/recipes/cache/BaseTestTreeCache.java
@@ -22,6 +22,7 @@
 import org.apache.curator.framework.CuratorFramework;
 import org.apache.curator.framework.CuratorFrameworkFactory;
 import org.apache.curator.framework.api.UnhandledErrorListener;
+import org.apache.curator.framework.imps.TestCleanState;
 import org.apache.curator.retry.RetryOneTime;
 import org.apache.curator.test.BaseClassForTests;
 import org.apache.curator.test.Timing;
@@ -123,7 +124,7 @@
             finally
             {
                 CloseableUtils.closeQuietly(cache);
-                CloseableUtils.closeQuietly(client);
+                TestCleanState.closeAndTestClean(client);
             }
         }
         finally
diff --git a/curator-recipes/src/test/java/org/apache/curator/framework/recipes/cache/TestNodeCache.java b/curator-recipes/src/test/java/org/apache/curator/framework/recipes/cache/TestNodeCache.java
index 27af6ac..d6d495a 100644
--- a/curator-recipes/src/test/java/org/apache/curator/framework/recipes/cache/TestNodeCache.java
+++ b/curator-recipes/src/test/java/org/apache/curator/framework/recipes/cache/TestNodeCache.java
@@ -18,6 +18,7 @@
  */
 package org.apache.curator.framework.recipes.cache;
 
+import org.apache.curator.framework.imps.TestCleanState;
 import org.apache.curator.test.BaseClassForTests;
 import org.apache.curator.utils.CloseableUtils;
 import org.apache.curator.framework.CuratorFramework;
@@ -98,7 +99,7 @@
         finally
         {
             CloseableUtils.closeQuietly(cache);
-            CloseableUtils.closeQuietly(client);
+            TestCleanState.closeAndTestClean(client);
         }
     }
 
@@ -161,7 +162,7 @@
         finally
         {
             CloseableUtils.closeQuietly(cache);
-            CloseableUtils.closeQuietly(client);
+            TestCleanState.closeAndTestClean(client);
         }
     }
 
@@ -204,7 +205,7 @@
         finally
         {
             CloseableUtils.closeQuietly(cache);
-            CloseableUtils.closeQuietly(client);
+            TestCleanState.closeAndTestClean(client);
         }
     }
 
@@ -252,7 +253,7 @@
         finally
         {
             CloseableUtils.closeQuietly(cache);
-            CloseableUtils.closeQuietly(client);
+            TestCleanState.closeAndTestClean(client);
         }
     }
 }
diff --git a/curator-recipes/src/test/java/org/apache/curator/framework/recipes/cache/TestPathChildrenCache.java b/curator-recipes/src/test/java/org/apache/curator/framework/recipes/cache/TestPathChildrenCache.java
index 14d061f..3571ca7 100644
--- a/curator-recipes/src/test/java/org/apache/curator/framework/recipes/cache/TestPathChildrenCache.java
+++ b/curator-recipes/src/test/java/org/apache/curator/framework/recipes/cache/TestPathChildrenCache.java
@@ -28,7 +28,7 @@
 import org.apache.curator.framework.api.CuratorEvent;
 import org.apache.curator.framework.api.Pathable;
 import org.apache.curator.framework.api.UnhandledErrorListener;
-import org.apache.curator.framework.imps.CuratorFrameworkImpl;
+import org.apache.curator.framework.imps.TestCleanState;
 import org.apache.curator.retry.RetryOneTime;
 import org.apache.curator.test.BaseClassForTests;
 import org.apache.curator.test.ExecuteCalledWatchingExecutorService;
@@ -97,7 +97,7 @@
         finally
         {
             CloseableUtils.closeQuietly(cache);
-            CloseableUtils.closeQuietly(client);
+            TestCleanState.closeAndTestClean(client);
         }
     }
 
@@ -139,7 +139,7 @@
         finally
         {
             CloseableUtils.closeQuietly(cache);
-            CloseableUtils.closeQuietly(client);
+            TestCleanState.closeAndTestClean(client);
         }
     }
 
@@ -193,7 +193,7 @@
         finally
         {
             CloseableUtils.closeQuietly(cache);
-            CloseableUtils.closeQuietly(client);
+            TestCleanState.closeAndTestClean(client);
         }
     }
 
@@ -242,7 +242,7 @@
         finally
         {
             CloseableUtils.closeQuietly(cache);
-            CloseableUtils.closeQuietly(client);
+            TestCleanState.closeAndTestClean(client);
         }
     }
 
@@ -251,6 +251,7 @@
     {
         Timing timing = new Timing();
 
+        PathChildrenCache cache = null;
         CuratorFramework client = CuratorFrameworkFactory.newClient(server.getConnectString(), timing.session(), timing.connection(), new RetryOneTime(1));
         client.start();
         try
@@ -258,7 +259,7 @@
             final CountDownLatch updatedLatch = new CountDownLatch(1);
             final CountDownLatch addedLatch = new CountDownLatch(1);
             client.create().creatingParentsIfNeeded().forPath("/test");
-            PathChildrenCache cache = new PathChildrenCache(client, "/test", false);
+            cache = new PathChildrenCache(client, "/test", false);
             cache.getListenable().addListener
                 (
                     new PathChildrenCacheListener()
@@ -287,7 +288,8 @@
         }
         finally
         {
-            CloseableUtils.closeQuietly(client);
+            CloseableUtils.closeQuietly(cache);
+            TestCleanState.closeAndTestClean(client);
         }
     }
 
@@ -315,7 +317,7 @@
         }
         finally
         {
-            CloseableUtils.closeQuietly(client);
+            TestCleanState.closeAndTestClean(client);
         }
     }
 
@@ -391,7 +393,7 @@
         }
         finally
         {
-            client.close();
+            TestCleanState.closeAndTestClean(client);
         }
     }
 
@@ -485,7 +487,7 @@
         }
         finally
         {
-            client.close();
+            TestCleanState.closeAndTestClean(client);
         }
     }
 
@@ -494,6 +496,7 @@
     public void testIssue27() throws Exception
     {
         Timing timing = new Timing();
+        PathChildrenCache cache = null;
         CuratorFramework client = CuratorFrameworkFactory.newClient(server.getConnectString(), timing.session(), timing.connection(), new RetryOneTime(1));
         client.start();
         try
@@ -507,7 +510,7 @@
 
             final List<PathChildrenCacheEvent.Type> events = Lists.newArrayList();
             final Semaphore semaphore = new Semaphore(0);
-            PathChildrenCache cache = new PathChildrenCache(client, "/base", true);
+            cache = new PathChildrenCache(client, "/base", true);
             cache.getListenable().addListener
                 (
                     new PathChildrenCacheListener()
@@ -542,7 +545,8 @@
         }
         finally
         {
-            client.close();
+            CloseableUtils.closeQuietly(cache);
+            TestCleanState.closeAndTestClean(client);
         }
     }
 
@@ -551,6 +555,7 @@
     public void testIssue27Alt() throws Exception
     {
         Timing timing = new Timing();
+        PathChildrenCache cache = null;
         CuratorFramework client = CuratorFrameworkFactory.newClient(server.getConnectString(), timing.session(), timing.connection(), new RetryOneTime(1));
         client.start();
         try
@@ -564,7 +569,7 @@
 
             final List<PathChildrenCacheEvent.Type> events = Lists.newArrayList();
             final Semaphore semaphore = new Semaphore(0);
-            PathChildrenCache cache = new PathChildrenCache(client, "/base", true);
+            cache = new PathChildrenCache(client, "/base", true);
             cache.getListenable().addListener
                 (
                     new PathChildrenCacheListener()
@@ -594,7 +599,8 @@
         }
         finally
         {
-            client.close();
+            CloseableUtils.closeQuietly(cache);
+            TestCleanState.closeAndTestClean(client);
         }
     }
 
@@ -602,6 +608,7 @@
     public void testKilledSession() throws Exception
     {
         Timing timing = new Timing();
+        PathChildrenCache cache = null;
         CuratorFramework client = null;
         try
         {
@@ -609,7 +616,7 @@
             client.start();
             client.create().forPath("/test");
 
-            PathChildrenCache cache = new PathChildrenCache(client, "/test", true);
+            cache = new PathChildrenCache(client, "/test", true);
             cache.start();
 
             final CountDownLatch childAddedLatch = new CountDownLatch(1);
@@ -653,7 +660,8 @@
         }
         finally
         {
-            CloseableUtils.closeQuietly(client);
+            CloseableUtils.closeQuietly(cache);
+            TestCleanState.closeAndTestClean(client);
         }
     }
 
@@ -677,7 +685,7 @@
         }
         finally
         {
-            client.close();
+            TestCleanState.closeAndTestClean(client);
         }
     }
 
@@ -721,7 +729,7 @@
         finally
         {
             CloseableUtils.closeQuietly(cache);
-            CloseableUtils.closeQuietly(client);
+            TestCleanState.closeAndTestClean(client);
         }
     }
 
@@ -809,7 +817,7 @@
         }
         finally
         {
-            client.close();
+            TestCleanState.closeAndTestClean(client);
         }
     }
 
@@ -878,7 +886,7 @@
         }
         finally
         {
-            client.close();
+            TestCleanState.closeAndTestClean(client);
         }
     }
 
@@ -911,8 +919,9 @@
             timing.sleepABit();
             Assert.assertFalse(exec.isExecuteCalled());
         }
-        finally {
-            client.close();
+        finally
+        {
+            TestCleanState.closeAndTestClean(client);
         }
 
     }
@@ -957,9 +966,10 @@
             latch.await(5, TimeUnit.SECONDS);
 
             Assert.assertTrue(latch.getCount() == 1, "Unexpected exception occurred");
-        } finally
+        }
+        finally
         {
-            CloseableUtils.closeQuietly(client);
+            TestCleanState.closeAndTestClean(client);
         }
     }
 }
diff --git a/curator-recipes/src/test/java/org/apache/curator/framework/recipes/leader/TestLeaderLatch.java b/curator-recipes/src/test/java/org/apache/curator/framework/recipes/leader/TestLeaderLatch.java
index 96e6d45..3742fb7 100644
--- a/curator-recipes/src/test/java/org/apache/curator/framework/recipes/leader/TestLeaderLatch.java
+++ b/curator-recipes/src/test/java/org/apache/curator/framework/recipes/leader/TestLeaderLatch.java
@@ -24,6 +24,7 @@
 import com.google.common.util.concurrent.ThreadFactoryBuilder;
 import org.apache.curator.framework.CuratorFramework;
 import org.apache.curator.framework.CuratorFrameworkFactory;
+import org.apache.curator.framework.imps.TestCleanState;
 import org.apache.curator.framework.state.ConnectionState;
 import org.apache.curator.framework.state.ConnectionStateListener;
 import org.apache.curator.retry.RetryNTimes;
@@ -96,7 +97,7 @@
         finally
         {
             CloseableUtils.closeQuietly(latch);
-            CloseableUtils.closeQuietly(client);
+            TestCleanState.closeAndTestClean(client);
         }
     }
 
@@ -126,7 +127,7 @@
         finally
         {
             CloseableUtils.closeQuietly(latch);
-            CloseableUtils.closeQuietly(client);
+            TestCleanState.closeAndTestClean(client);
         }
     }
 
@@ -158,7 +159,7 @@
         }
         finally
         {
-            CloseableUtils.closeQuietly(client);
+            TestCleanState.closeAndTestClean(client);
         }
     }
 
@@ -213,7 +214,7 @@
             {
                 CloseableUtils.closeQuietly(latch);
             }
-            CloseableUtils.closeQuietly(client);
+            TestCleanState.closeAndTestClean(client);
         }
     }
 
@@ -256,9 +257,8 @@
             {
                 CloseableUtils.closeQuietly(latch);
             }
-            CloseableUtils.closeQuietly(client);
+            TestCleanState.closeAndTestClean(client);
         }
-
     }
 
     @Test
@@ -320,7 +320,7 @@
         finally
         {
             executorService.shutdownNow();
-            CloseableUtils.closeQuietly(client);
+            TestCleanState.closeAndTestClean(client);
         }
     }
 
@@ -416,7 +416,7 @@
                     CloseableUtils.closeQuietly(latch);
                 }
             }
-            CloseableUtils.closeQuietly(client);
+            TestCleanState.closeAndTestClean(client);
         }
     }
 
@@ -504,7 +504,7 @@
                     CloseableUtils.closeQuietly(latch);
                 }
             }
-            CloseableUtils.closeQuietly(client);
+            TestCleanState.closeAndTestClean(client);
         }
     }
 
@@ -583,7 +583,7 @@
             {
                 CloseableUtils.closeQuietly(notifiedLeader);
             }
-            CloseableUtils.closeQuietly(client);
+            TestCleanState.closeAndTestClean(client);
         }
     }
 
@@ -639,7 +639,7 @@
         finally
         {
             CloseableUtils.closeQuietly(leader);
-            CloseableUtils.closeQuietly(client);
+            TestCleanState.closeAndTestClean(client);
             CloseableUtils.closeQuietly(server);
         }
     }
@@ -709,7 +709,7 @@
             {
                 CloseableUtils.closeQuietly(latch);
             }
-            CloseableUtils.closeQuietly(client);
+            TestCleanState.closeAndTestClean(client);
         }
     }
 
@@ -745,6 +745,6 @@
     {
         Timing timing = new Timing();
         CuratorFramework client = CuratorFrameworkFactory.newClient(server.getConnectString(), timing.session(), timing.connection(), new RetryOneTime(1));
-        LeaderLatch latch = new LeaderLatch(client, "parent");
+        new LeaderLatch(client, "parent");
     }
 }
diff --git a/curator-recipes/src/test/java/org/apache/curator/framework/recipes/locks/TestInterProcessMultiMutex.java b/curator-recipes/src/test/java/org/apache/curator/framework/recipes/locks/TestInterProcessMultiMutex.java
index 09b5fe6..dceff88 100644
--- a/curator-recipes/src/test/java/org/apache/curator/framework/recipes/locks/TestInterProcessMultiMutex.java
+++ b/curator-recipes/src/test/java/org/apache/curator/framework/recipes/locks/TestInterProcessMultiMutex.java
@@ -20,6 +20,7 @@
 
 import org.apache.curator.framework.CuratorFramework;
 import org.apache.curator.framework.CuratorFrameworkFactory;
+import org.apache.curator.framework.imps.TestCleanState;
 import org.apache.curator.retry.RetryOneTime;
 import org.testng.Assert;
 import org.testng.annotations.Test;
@@ -84,13 +85,14 @@
             }
             catch ( Exception e )
             {
+                // ignore
             }
             Assert.assertFalse(goodLock.isAcquiredInThisProcess());
             Assert.assertTrue(otherGoodLock.isAcquiredInThisProcess());
         }
         finally
         {
-            client.close();
+            TestCleanState.closeAndTestClean(client);
         }
     }
 
@@ -142,13 +144,14 @@
             }
             catch ( Exception e )
             {
+                // ignore
             }
             Assert.assertFalse(goodLock.isAcquiredInThisProcess());
             Assert.assertTrue(goodLockWasLocked.get());
         }
         finally
         {
-            client.close();
+            TestCleanState.closeAndTestClean(client);
         }
     }
 }
diff --git a/curator-recipes/src/test/java/org/apache/curator/framework/recipes/locks/TestInterProcessMutex.java b/curator-recipes/src/test/java/org/apache/curator/framework/recipes/locks/TestInterProcessMutex.java
index a2c079e..c37d88d 100644
--- a/curator-recipes/src/test/java/org/apache/curator/framework/recipes/locks/TestInterProcessMutex.java
+++ b/curator-recipes/src/test/java/org/apache/curator/framework/recipes/locks/TestInterProcessMutex.java
@@ -20,6 +20,7 @@
 
 import org.apache.curator.framework.CuratorFramework;
 import org.apache.curator.framework.CuratorFrameworkFactory;
+import org.apache.curator.framework.imps.TestCleanState;
 import org.apache.curator.retry.RetryOneTime;
 import org.apache.curator.test.KillSession;
 import org.apache.zookeeper.CreateMode;
@@ -106,7 +107,7 @@
         }
         finally
         {
-            client.close();
+            TestCleanState.closeAndTestClean(client);
         }
     }
 
@@ -151,7 +152,7 @@
         }
         finally
         {
-            client.close();
+            TestCleanState.closeAndTestClean(client);
         }
     }
 }
diff --git a/curator-recipes/src/test/java/org/apache/curator/framework/recipes/locks/TestInterProcessMutexBase.java b/curator-recipes/src/test/java/org/apache/curator/framework/recipes/locks/TestInterProcessMutexBase.java
index 99ea11f..f44d238 100644
--- a/curator-recipes/src/test/java/org/apache/curator/framework/recipes/locks/TestInterProcessMutexBase.java
+++ b/curator-recipes/src/test/java/org/apache/curator/framework/recipes/locks/TestInterProcessMutexBase.java
@@ -22,12 +22,12 @@
 import com.google.common.collect.Lists;
 import org.apache.curator.framework.CuratorFramework;
 import org.apache.curator.framework.CuratorFrameworkFactory;
+import org.apache.curator.framework.imps.TestCleanState;
 import org.apache.curator.framework.state.ConnectionState;
 import org.apache.curator.framework.state.ConnectionStateListener;
 import org.apache.curator.retry.ExponentialBackoffRetry;
 import org.apache.curator.test.BaseClassForTests;
 import org.apache.curator.test.KillSession;
-import org.apache.curator.test.TestingServer;
 import org.apache.curator.test.Timing;
 import org.apache.curator.utils.CloseableUtils;
 import org.apache.curator.utils.ZKPaths;
@@ -126,7 +126,7 @@
         }
         finally
         {
-            CloseableUtils.closeQuietly(client);
+            TestCleanState.closeAndTestClean(client);
         }
     }
 
@@ -278,7 +278,7 @@
         }
         finally
         {
-            client.close();
+            TestCleanState.closeAndTestClean(client);
         }
     }
 
@@ -344,7 +344,7 @@
         }
         finally
         {
-            client.close();
+            TestCleanState.closeAndTestClean(client);
         }
     }
 
@@ -390,7 +390,7 @@
         }
         finally
         {
-            client.close();
+            TestCleanState.closeAndTestClean(client);
         }
     }
 
@@ -407,7 +407,7 @@
         }
         finally
         {
-            client.close();
+            TestCleanState.closeAndTestClean(client);
         }
     }
 
@@ -539,11 +539,14 @@
                 Assert.assertTrue(acquiredLatchForClient1.await(10, TimeUnit.SECONDS));
                 Assert.assertTrue(mutexForClient1.isAcquiredInThisProcess());
             }
+
+            future1.get();
+            future2.get();
         }
         finally
         {
-            CloseableUtils.closeQuietly(client1);
-            CloseableUtils.closeQuietly(client2);
+            TestCleanState.closeAndTestClean(client1);
+            TestCleanState.closeAndTestClean(client2);
         }
     }
 }
diff --git a/curator-recipes/src/test/java/org/apache/curator/framework/recipes/locks/TestInterProcessReadWriteLock.java b/curator-recipes/src/test/java/org/apache/curator/framework/recipes/locks/TestInterProcessReadWriteLock.java
index f7636ed..48e4805 100644
--- a/curator-recipes/src/test/java/org/apache/curator/framework/recipes/locks/TestInterProcessReadWriteLock.java
+++ b/curator-recipes/src/test/java/org/apache/curator/framework/recipes/locks/TestInterProcessReadWriteLock.java
@@ -16,14 +16,15 @@
  * specific language governing permissions and limitations
  * under the License.
  */
+
 package org.apache.curator.framework.recipes.locks;
 
 import com.google.common.collect.Lists;
-import org.apache.curator.test.BaseClassForTests;
-import org.apache.curator.utils.CloseableUtils;
 import org.apache.curator.framework.CuratorFramework;
 import org.apache.curator.framework.CuratorFrameworkFactory;
+import org.apache.curator.framework.imps.TestCleanState;
 import org.apache.curator.retry.RetryOneTime;
+import org.apache.curator.test.BaseClassForTests;
 import org.testng.Assert;
 import org.testng.annotations.Test;
 import java.util.Collection;
@@ -31,6 +32,7 @@
 import java.util.Random;
 import java.util.concurrent.Callable;
 import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorCompletionService;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
 import java.util.concurrent.Future;
@@ -40,21 +42,22 @@
 public class TestInterProcessReadWriteLock extends BaseClassForTests
 {
     @Test
-    public void     testGetParticipantNodes() throws Exception
+    public void testGetParticipantNodes() throws Exception
     {
-        final int               READERS = 20;
-        final int               WRITERS = 8;
+        final int READERS = 20;
+        final int WRITERS = 8;
 
-        CuratorFramework        client = CuratorFrameworkFactory.newClient(server.getConnectString(), new RetryOneTime(1));
+        CuratorFramework client = CuratorFrameworkFactory.newClient(server.getConnectString(), new RetryOneTime(1));
         try
         {
             client.start();
 
-            final CountDownLatch              latch = new CountDownLatch(READERS + WRITERS);
-            final CountDownLatch              readLatch = new CountDownLatch(READERS);
-            final InterProcessReadWriteLock   lock = new InterProcessReadWriteLock(client, "/lock");
+            final CountDownLatch latch = new CountDownLatch(READERS + WRITERS);
+            final CountDownLatch readLatch = new CountDownLatch(READERS);
+            final InterProcessReadWriteLock lock = new InterProcessReadWriteLock(client, "/lock");
 
-            ExecutorService                   service = Executors.newCachedThreadPool();
+            final CountDownLatch exitLatch = new CountDownLatch(1);
+            ExecutorCompletionService<Void> service = new ExecutorCompletionService<Void>(Executors.newCachedThreadPool());
             for ( int i = 0; i < READERS; ++i )
             {
                 service.submit
@@ -65,8 +68,16 @@
                         public Void call() throws Exception
                         {
                             lock.readLock().acquire();
-                            latch.countDown();
-                            readLatch.countDown();
+                            try
+                            {
+                                latch.countDown();
+                                readLatch.countDown();
+                                exitLatch.await();
+                            }
+                            finally
+                            {
+                                lock.readLock().release();
+                            }
                             return null;
                         }
                     }
@@ -84,6 +95,14 @@
                             Assert.assertTrue(readLatch.await(10, TimeUnit.SECONDS));
                             latch.countDown();  // must be before as there can only be one writer
                             lock.writeLock().acquire();
+                            try
+                            {
+                                exitLatch.await();
+                            }
+                            finally
+                            {
+                                lock.writeLock().release();
+                            }
                             return null;
                         }
                     }
@@ -97,22 +116,28 @@
 
             Assert.assertEquals(readers.size(), READERS);
             Assert.assertEquals(writers.size(), WRITERS);
+
+            exitLatch.countDown();
+            for ( int i = 0; i < (READERS + WRITERS); ++i )
+            {
+                service.take().get();
+            }
         }
         finally
         {
-            CloseableUtils.closeQuietly(client);
+            TestCleanState.closeAndTestClean(client);
         }
     }
 
     @Test
-    public void     testThatUpgradingIsDisallowed() throws Exception
+    public void testThatUpgradingIsDisallowed() throws Exception
     {
-        CuratorFramework        client = CuratorFrameworkFactory.newClient(server.getConnectString(), new RetryOneTime(1));
+        CuratorFramework client = CuratorFrameworkFactory.newClient(server.getConnectString(), new RetryOneTime(1));
         try
         {
             client.start();
 
-            InterProcessReadWriteLock   lock = new InterProcessReadWriteLock(client, "/lock");
+            InterProcessReadWriteLock lock = new InterProcessReadWriteLock(client, "/lock");
             lock.readLock().acquire();
             Assert.assertFalse(lock.writeLock().acquire(5, TimeUnit.SECONDS));
 
@@ -120,70 +145,80 @@
         }
         finally
         {
-            CloseableUtils.closeQuietly(client);
+            TestCleanState.closeAndTestClean(client);
         }
     }
 
     @Test
-    public void     testThatDowngradingRespectsThreads() throws Exception
+    public void testThatDowngradingRespectsThreads() throws Exception
     {
-        CuratorFramework        client = CuratorFrameworkFactory.newClient(server.getConnectString(), new RetryOneTime(1));
+        CuratorFramework client = CuratorFrameworkFactory.newClient(server.getConnectString(), new RetryOneTime(1));
         try
         {
             client.start();
 
-            final InterProcessReadWriteLock   lock = new InterProcessReadWriteLock(client, "/lock");
-            ExecutorService                   t1 = Executors.newSingleThreadExecutor();
-            ExecutorService                   t2 = Executors.newSingleThreadExecutor();
+            final InterProcessReadWriteLock lock = new InterProcessReadWriteLock(client, "/lock");
+            ExecutorService t1 = Executors.newSingleThreadExecutor();
+            ExecutorService t2 = Executors.newSingleThreadExecutor();
 
-            final CountDownLatch              latch = new CountDownLatch(1);
+            final CountDownLatch latch = new CountDownLatch(1);
 
-            Future<Object>                    f1 = t1.submit
-            (
-                new Callable<Object>()
-                {
-                    @Override
-                    public Object call() throws Exception
+            final CountDownLatch releaseLatch = new CountDownLatch(1);
+            Future<Object> f1 = t1.submit
+                (
+                    new Callable<Object>()
                     {
-                        lock.writeLock().acquire();
-                        latch.countDown();
-                        return null;
+                        @Override
+                        public Object call() throws Exception
+                        {
+                            lock.writeLock().acquire();
+                            latch.countDown();
+                            try
+                            {
+                                releaseLatch.await();
+                            }
+                            finally
+                            {
+                                lock.writeLock().release();
+                            }
+                            return null;
+                        }
                     }
-                }
-            );
+                );
 
-            Future<Object>                    f2 = t2.submit
-            (
-                new Callable<Object>()
-                {
-                    @Override
-                    public Object call() throws Exception
+            Future<Object> f2 = t2.submit
+                (
+                    new Callable<Object>()
                     {
-                        Assert.assertTrue(latch.await(10, TimeUnit.SECONDS));
-                        Assert.assertFalse(lock.readLock().acquire(5, TimeUnit.SECONDS));
-                        return null;
+                        @Override
+                        public Object call() throws Exception
+                        {
+                            Assert.assertTrue(latch.await(10, TimeUnit.SECONDS));
+                            Assert.assertFalse(lock.readLock().acquire(5, TimeUnit.SECONDS));
+                            return null;
+                        }
                     }
-                }
-            );
+                );
 
-            f1.get();
             f2.get();
+            releaseLatch.countDown();
+            f1.get();
         }
         finally
         {
-            CloseableUtils.closeQuietly(client);
+            TestCleanState.closeAndTestClean(client);
         }
     }
 
     @Test
-    public void     testDowngrading() throws Exception
+    public void testDowngrading() throws Exception
     {
-        CuratorFramework        client = CuratorFrameworkFactory.newClient(server.getConnectString(), new RetryOneTime(1));
+        CuratorFramework client = CuratorFrameworkFactory.newClient(server.getConnectString(), new RetryOneTime(1));
         try
         {
             client.start();
 
-            InterProcessReadWriteLock   lock = new InterProcessReadWriteLock(client, "/lock");
+            InterProcessReadWriteLock lock = new InterProcessReadWriteLock(client, "/lock");
             lock.writeLock().acquire();
             Assert.assertTrue(lock.readLock().acquire(5, TimeUnit.SECONDS));
             lock.writeLock().release();
@@ -192,60 +227,60 @@
         }
         finally
         {
-            CloseableUtils.closeQuietly(client);
+            TestCleanState.closeAndTestClean(client);
         }
     }
 
     @Test
-    public void     testBasic() throws Exception
+    public void testBasic() throws Exception
     {
-        final int               CONCURRENCY = 8;
-        final int               ITERATIONS = 100;
+        final int CONCURRENCY = 8;
+        final int ITERATIONS = 100;
 
-        final Random            random = new Random();
-        final AtomicInteger     concurrentCount = new AtomicInteger(0);
-        final AtomicInteger     maxConcurrentCount = new AtomicInteger(0);
-        final AtomicInteger     writeCount = new AtomicInteger(0);
-        final AtomicInteger     readCount = new AtomicInteger(0);
+        final Random random = new Random();
+        final AtomicInteger concurrentCount = new AtomicInteger(0);
+        final AtomicInteger maxConcurrentCount = new AtomicInteger(0);
+        final AtomicInteger writeCount = new AtomicInteger(0);
+        final AtomicInteger readCount = new AtomicInteger(0);
 
-        List<Future<Void>>  futures = Lists.newArrayList();
-        ExecutorService     service = Executors.newCachedThreadPool();
+        List<Future<Void>> futures = Lists.newArrayList();
+        ExecutorService service = Executors.newCachedThreadPool();
         for ( int i = 0; i < CONCURRENCY; ++i )
         {
-            Future<Void>    future = service.submit
-            (
-                new Callable<Void>()
-                {
-                    @Override
-                    public Void call() throws Exception
+            Future<Void> future = service.submit
+                (
+                    new Callable<Void>()
                     {
-                        CuratorFramework        client = CuratorFrameworkFactory.newClient(server.getConnectString(), new RetryOneTime(1));
-                        client.start();
-                        try
+                        @Override
+                        public Void call() throws Exception
                         {
-                            InterProcessReadWriteLock   lock = new InterProcessReadWriteLock(client, "/lock");
-                            for ( int i = 0; i < ITERATIONS; ++i )
+                            CuratorFramework client = CuratorFrameworkFactory.newClient(server.getConnectString(), new RetryOneTime(1));
+                            client.start();
+                            try
                             {
-                                if ( random.nextInt(100) < 10 )
+                                InterProcessReadWriteLock lock = new InterProcessReadWriteLock(client, "/lock");
+                                for ( int i = 0; i < ITERATIONS; ++i )
                                 {
-                                    doLocking(lock.writeLock(), concurrentCount, maxConcurrentCount, random, 1);
-                                    writeCount.incrementAndGet();
-                                }
-                                else
-                                {
-                                    doLocking(lock.readLock(), concurrentCount, maxConcurrentCount, random, Integer.MAX_VALUE);
-                                    readCount.incrementAndGet();
+                                    if ( random.nextInt(100) < 10 )
+                                    {
+                                        doLocking(lock.writeLock(), concurrentCount, maxConcurrentCount, random, 1);
+                                        writeCount.incrementAndGet();
+                                    }
+                                    else
+                                    {
+                                        doLocking(lock.readLock(), concurrentCount, maxConcurrentCount, random, Integer.MAX_VALUE);
+                                        readCount.incrementAndGet();
+                                    }
                                 }
                             }
+                            finally
+                            {
+                                TestCleanState.closeAndTestClean(client);
+                            }
+                            return null;
                         }
-                        finally
-                        {
-                            CloseableUtils.closeQuietly(client);
-                        }
-                        return null;
                     }
-                }
-            );
+                );
             futures.add(future);
         }
 
@@ -262,17 +297,17 @@
     }
 
     @Test
-    public void     testSetNodeData() throws Exception
+    public void testSetNodeData() throws Exception
     {
-        CuratorFramework        client = CuratorFrameworkFactory.newClient(server.getConnectString(), new RetryOneTime(1));
+        CuratorFramework client = CuratorFrameworkFactory.newClient(server.getConnectString(), new RetryOneTime(1));
 
         try
         {
             client.start();
 
-            final byte[] nodeData = new byte[] { 1, 2, 3, 4 };
+            final byte[] nodeData = new byte[]{1, 2, 3, 4};
 
-            InterProcessReadWriteLock   lock = new InterProcessReadWriteLock(client, "/lock", nodeData);
+            InterProcessReadWriteLock lock = new InterProcessReadWriteLock(client, "/lock", nodeData);
 
             // mutate passed-in node data, lock has made copy
             nodeData[0] = 5;
@@ -284,13 +319,13 @@
 
             byte dataInZk[] = client.getData().forPath("/lock/" + children.get(0));
             Assert.assertNotNull(dataInZk);
-            Assert.assertEquals(new byte[] { 1, 2, 3, 4 }, dataInZk);
+            Assert.assertEquals(new byte[]{1, 2, 3, 4}, dataInZk);
 
             lock.writeLock().release();
         }
         finally
         {
-            CloseableUtils.closeQuietly(client);
+            TestCleanState.closeAndTestClean(client);
         }
     }
 
@@ -299,7 +334,7 @@
         try
         {
             Assert.assertTrue(lock.acquire(10, TimeUnit.SECONDS));
-            int     localConcurrentCount;
+            int localConcurrentCount;
             synchronized(this)
             {
                 localConcurrentCount = concurrentCount.incrementAndGet();
diff --git a/curator-recipes/src/test/java/org/apache/curator/framework/recipes/locks/TestInterProcessSemaphore.java b/curator-recipes/src/test/java/org/apache/curator/framework/recipes/locks/TestInterProcessSemaphore.java
index 631b7c7..3ba75d8 100644
--- a/curator-recipes/src/test/java/org/apache/curator/framework/recipes/locks/TestInterProcessSemaphore.java
+++ b/curator-recipes/src/test/java/org/apache/curator/framework/recipes/locks/TestInterProcessSemaphore.java
@@ -20,13 +20,14 @@
 package org.apache.curator.framework.recipes.locks;
 
 import com.google.common.collect.Lists;
-import org.apache.curator.test.BaseClassForTests;
-import org.apache.curator.utils.CloseableUtils;
 import org.apache.curator.framework.CuratorFramework;
 import org.apache.curator.framework.CuratorFrameworkFactory;
+import org.apache.curator.framework.imps.TestCleanState;
 import org.apache.curator.framework.recipes.shared.SharedCount;
 import org.apache.curator.retry.RetryOneTime;
+import org.apache.curator.test.BaseClassForTests;
 import org.apache.curator.test.Timing;
+import org.apache.curator.utils.CloseableUtils;
 import org.testng.Assert;
 import org.testng.annotations.Test;
 import java.util.Collection;
@@ -100,10 +101,12 @@
 
             future1.get();
             future2.get();
+
+            count.close();
         }
         finally
         {
-            CloseableUtils.closeQuietly(client);
+            TestCleanState.closeAndTestClean(client);
         }
     }
 
@@ -140,8 +143,8 @@
         }
         finally
         {
-            CloseableUtils.closeQuietly(client1);
-            CloseableUtils.closeQuietly(client2);
+            TestCleanState.closeAndTestClean(client1);
+            TestCleanState.closeAndTestClean(client2);
         }
     }
 
@@ -226,7 +229,7 @@
                                     }
                                     finally
                                     {
-                                        client.close();
+                                        TestCleanState.closeAndTestClean(client);
                                     }
                                     return null;
                                 }
@@ -299,7 +302,7 @@
                             }
                             finally
                             {
-                                client.close();
+                                TestCleanState.closeAndTestClean(client);
                             }
                             return null;
                         }
@@ -401,7 +404,7 @@
         }
         finally
         {
-            client.close();
+            TestCleanState.closeAndTestClean(client);
         }
     }
 
@@ -445,7 +448,7 @@
         }
         finally
         {
-            client.close();
+            TestCleanState.closeAndTestClean(client);
         }
     }
 
@@ -463,7 +466,7 @@
         }
         finally
         {
-            client.close();
+            TestCleanState.closeAndTestClean(client);
         }
     }
 
@@ -499,7 +502,7 @@
             {
                 CloseableUtils.closeQuietly(l);
             }
-            CloseableUtils.closeQuietly(client);
+            TestCleanState.closeAndTestClean(client);
         }
     }
 
@@ -528,7 +531,7 @@
             {
                 CloseableUtils.closeQuietly(l);
             }
-            CloseableUtils.closeQuietly(client);
+            TestCleanState.closeAndTestClean(client);
         }
     }
 
diff --git a/curator-recipes/src/test/java/org/apache/curator/framework/recipes/locks/TestInterProcessSemaphoreCluster.java b/curator-recipes/src/test/java/org/apache/curator/framework/recipes/locks/TestInterProcessSemaphoreCluster.java
index 2aa8a72..f4cb7bb 100644
--- a/curator-recipes/src/test/java/org/apache/curator/framework/recipes/locks/TestInterProcessSemaphoreCluster.java
+++ b/curator-recipes/src/test/java/org/apache/curator/framework/recipes/locks/TestInterProcessSemaphoreCluster.java
@@ -22,6 +22,7 @@
 import org.apache.curator.ensemble.EnsembleProvider;
 import org.apache.curator.framework.CuratorFramework;
 import org.apache.curator.framework.CuratorFrameworkFactory;
+import org.apache.curator.framework.imps.TestCleanState;
 import org.apache.curator.framework.state.ConnectionState;
 import org.apache.curator.framework.state.ConnectionStateListener;
 import org.apache.curator.retry.ExponentialBackoffRetry;
@@ -147,7 +148,7 @@
                             }
                             finally
                             {
-                                CloseableUtils.closeQuietly(client);
+                                TestCleanState.closeAndTestClean(client);
                             }
                             return null;
                         }
diff --git a/curator-recipes/src/test/java/org/apache/curator/framework/recipes/locks/TestLockACLs.java b/curator-recipes/src/test/java/org/apache/curator/framework/recipes/locks/TestLockACLs.java
index 2d9a9aa..d1e6db5 100644
--- a/curator-recipes/src/test/java/org/apache/curator/framework/recipes/locks/TestLockACLs.java
+++ b/curator-recipes/src/test/java/org/apache/curator/framework/recipes/locks/TestLockACLs.java
@@ -19,6 +19,7 @@
 
 package org.apache.curator.framework.recipes.locks;
 
+import org.apache.curator.framework.imps.TestCleanState;
 import org.apache.curator.test.BaseClassForTests;
 import org.apache.curator.utils.CloseableUtils;
 import org.apache.curator.RetryPolicy;
@@ -74,7 +75,7 @@
         }
         finally
         {
-            CloseableUtils.closeQuietly(client);
+            TestCleanState.closeAndTestClean(client);
         }
     }
 
diff --git a/curator-recipes/src/test/java/org/apache/curator/framework/recipes/locks/TestLockCleanlinessWithFaults.java b/curator-recipes/src/test/java/org/apache/curator/framework/recipes/locks/TestLockCleanlinessWithFaults.java
index 457be75..dc14c11 100644
--- a/curator-recipes/src/test/java/org/apache/curator/framework/recipes/locks/TestLockCleanlinessWithFaults.java
+++ b/curator-recipes/src/test/java/org/apache/curator/framework/recipes/locks/TestLockCleanlinessWithFaults.java
@@ -20,6 +20,7 @@
 
 import org.apache.curator.framework.CuratorFramework;
 import org.apache.curator.framework.CuratorFrameworkFactory;
+import org.apache.curator.framework.imps.TestCleanState;
 import org.apache.curator.retry.RetryNTimes;
 import org.apache.curator.test.BaseClassForTests;
 import org.apache.curator.utils.CloseableUtils;
@@ -67,7 +68,7 @@
         }
         finally
         {
-            CloseableUtils.closeQuietly(client);
+            TestCleanState.closeAndTestClean(client);
         }
     }
 }
diff --git a/curator-recipes/src/test/java/org/apache/curator/framework/recipes/nodes/TestPersistentEphemeralNode.java b/curator-recipes/src/test/java/org/apache/curator/framework/recipes/nodes/TestPersistentEphemeralNode.java
index 9f5907a..c81cc65 100644
--- a/curator-recipes/src/test/java/org/apache/curator/framework/recipes/nodes/TestPersistentEphemeralNode.java
+++ b/curator-recipes/src/test/java/org/apache/curator/framework/recipes/nodes/TestPersistentEphemeralNode.java
@@ -20,17 +20,16 @@
 
 import com.google.common.base.Throwables;
 import com.google.common.collect.Lists;
-
 import org.apache.curator.framework.CuratorFramework;
 import org.apache.curator.framework.CuratorFrameworkFactory;
 import org.apache.curator.framework.api.BackgroundCallback;
 import org.apache.curator.framework.api.CuratorEvent;
+import org.apache.curator.framework.imps.TestCleanState;
 import org.apache.curator.framework.state.ConnectionState;
 import org.apache.curator.framework.state.ConnectionStateListener;
 import org.apache.curator.retry.RetryOneTime;
 import org.apache.curator.test.BaseClassForTests;
 import org.apache.curator.test.KillSession;
-import org.apache.curator.test.TestingServer;
 import org.apache.curator.test.Timing;
 import org.apache.curator.utils.CloseableUtils;
 import org.apache.curator.utils.ZKPaths;
@@ -64,6 +63,7 @@
     private final Timing timing = new Timing();
 
     @AfterMethod
+    @Override
     public void teardown() throws Exception
     {
         for ( PersistentEphemeralNode node : createdNodes )
@@ -73,10 +73,8 @@
 
         for ( CuratorFramework curator : curatorInstances )
         {
-            curator.close();
+            TestCleanState.closeAndTestClean(curator);
         }
-
-        super.teardown();
     }
 
     @Test
@@ -122,7 +120,7 @@
         }
         finally
         {
-            CloseableUtils.closeQuietly(client);
+            TestCleanState.closeAndTestClean(client);
         }
     }
 
@@ -132,10 +130,11 @@
         server.stop();
 
         CuratorFramework client = CuratorFrameworkFactory.newClient(server.getConnectString(), timing.session(), timing.connection(), new RetryOneTime(1));
+        PersistentEphemeralNode node = null;
         try
         {
             client.start();
-            PersistentEphemeralNode node = new PersistentEphemeralNode(client, PersistentEphemeralNode.Mode.EPHEMERAL, "/abc/node", "hello".getBytes());
+            node = new PersistentEphemeralNode(client, PersistentEphemeralNode.Mode.EPHEMERAL, "/abc/node", "hello".getBytes());
             node.start();
 
             final CountDownLatch connectedLatch = new CountDownLatch(1);
@@ -164,7 +163,8 @@
         }
         finally
         {
-            CloseableUtils.closeQuietly(client);
+            CloseableUtils.closeQuietly(node);
+            TestCleanState.closeAndTestClean(client);
         }
     }
 
@@ -247,7 +247,7 @@
             {
                 node.close();
             }
-            client.close();
+            TestCleanState.closeAndTestClean(client);
         }
     }
 
diff --git a/curator-recipes/src/test/java/org/apache/curator/framework/recipes/shared/TestSharedCount.java b/curator-recipes/src/test/java/org/apache/curator/framework/recipes/shared/TestSharedCount.java
index 659154a..2bdd278 100644
--- a/curator-recipes/src/test/java/org/apache/curator/framework/recipes/shared/TestSharedCount.java
+++ b/curator-recipes/src/test/java/org/apache/curator/framework/recipes/shared/TestSharedCount.java
@@ -23,6 +23,7 @@
 import com.google.common.util.concurrent.ThreadFactoryBuilder;
 import org.apache.curator.framework.CuratorFramework;
 import org.apache.curator.framework.CuratorFrameworkFactory;
+import org.apache.curator.framework.imps.TestCleanState;
 import org.apache.curator.framework.state.ConnectionState;
 import org.apache.curator.retry.RetryOneTime;
 import org.apache.curator.test.BaseClassForTests;
@@ -147,7 +148,7 @@
             }
             for ( CuratorFramework client : clients )
             {
-                CloseableUtils.closeQuietly(client);
+                TestCleanState.closeAndTestClean(client);
             }
         }
     }
@@ -170,7 +171,7 @@
         finally
         {
             CloseableUtils.closeQuietly(count);
-            CloseableUtils.closeQuietly(client);
+            TestCleanState.closeAndTestClean(client);
         }
     }
 
@@ -215,7 +216,7 @@
         finally
         {
             CloseableUtils.closeQuietly(count);
-            CloseableUtils.closeQuietly(client);
+            TestCleanState.closeAndTestClean(client);
         }
     }
 
@@ -252,8 +253,8 @@
         {
             CloseableUtils.closeQuietly(count2);
             CloseableUtils.closeQuietly(count1);
-            CloseableUtils.closeQuietly(client2);
-            CloseableUtils.closeQuietly(client1);
+            TestCleanState.closeAndTestClean(client2);
+            TestCleanState.closeAndTestClean(client1);
         }
     }
 }
diff --git a/curator-test/src/main/java/org/apache/curator/test/BaseClassForTests.java b/curator-test/src/main/java/org/apache/curator/test/BaseClassForTests.java
index 13c3138..6ef3bb0 100644
--- a/curator-test/src/main/java/org/apache/curator/test/BaseClassForTests.java
+++ b/curator-test/src/main/java/org/apache/curator/test/BaseClassForTests.java
@@ -35,6 +35,7 @@
 
     private static final int    RETRY_WAIT_MS = 5000;
     private static final String INTERNAL_PROPERTY_DONT_LOG_CONNECTION_ISSUES;
+    private static final String INTERNAL_PROPERTY_REMOVE_WATCHERS_IN_FOREGROUND;
     private static final String INTERNAL_RETRY_FAILED_TESTS;
     static
     {
@@ -53,6 +54,17 @@
         }
         INTERNAL_PROPERTY_DONT_LOG_CONNECTION_ISSUES = logConnectionIssues;
         INTERNAL_RETRY_FAILED_TESTS = retryFailedTests;
+        String s = null;
+        try
+        {
+            // use reflection to avoid adding a circular dependency in the pom
+            s = (String)Class.forName("org.apache.curator.utils.DebugUtils").getField("PROPERTY_REMOVE_WATCHERS_IN_FOREGROUND").get(null);
+        }
+        catch ( Exception e )
+        {
+            e.printStackTrace();
+        }
+        INTERNAL_PROPERTY_REMOVE_WATCHERS_IN_FOREGROUND = s;
     }
 
     @BeforeSuite(alwaysRun = true)
@@ -71,6 +83,7 @@
         {
             System.setProperty(INTERNAL_PROPERTY_DONT_LOG_CONNECTION_ISSUES, "true");
         }
+        System.setProperty(INTERNAL_PROPERTY_REMOVE_WATCHERS_IN_FOREGROUND, "true");
 
         while ( server == null )
         {
@@ -89,6 +102,7 @@
     @AfterMethod
     public void teardown() throws Exception
     {
+        System.clearProperty(INTERNAL_PROPERTY_REMOVE_WATCHERS_IN_FOREGROUND);
         if ( server != null )
         {
             try
diff --git a/curator-test/src/main/java/org/apache/curator/test/WatchersDebug.java b/curator-test/src/main/java/org/apache/curator/test/WatchersDebug.java
new file mode 100644
index 0000000..e4c3b7e
--- /dev/null
+++ b/curator-test/src/main/java/org/apache/curator/test/WatchersDebug.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.curator.test;
+
+import org.apache.zookeeper.ZooKeeper;
+import java.lang.reflect.Method;
+import java.util.List;
+
+public class WatchersDebug
+{
+    private static final Method getDataWatches;
+    private static final Method getExistWatches;
+    private static final Method getChildWatches;
+    static
+    {
+        Method localGetDataWatches = null;
+        Method localGetExistWatches = null;
+        Method localGetChildWatches = null;
+        try
+        {
+            localGetDataWatches = getMethod("getDataWatches");
+            localGetExistWatches = getMethod("getExistWatches");
+            localGetChildWatches = getMethod("getChildWatches");
+        }
+        catch ( NoSuchMethodException e )
+        {
+            e.printStackTrace();
+        }
+        getDataWatches = localGetDataWatches;
+        getExistWatches = localGetExistWatches;
+        getChildWatches = localGetChildWatches;
+    }
+
+    public static List<String> getDataWatches(ZooKeeper zooKeeper)
+    {
+        return callMethod(zooKeeper, WatchersDebug.getDataWatches);
+    }
+
+    public static List<String> getExistWatches(ZooKeeper zooKeeper)
+    {
+        return callMethod(zooKeeper, getExistWatches);
+    }
+
+    public static List<String> getChildWatches(ZooKeeper zooKeeper)
+    {
+        return callMethod(zooKeeper, getChildWatches);
+    }
+
+    private WatchersDebug()
+    {
+    }
+
+    private static Method getMethod(String name) throws NoSuchMethodException
+    {
+        Method m = ZooKeeper.class.getDeclaredMethod(name);
+        m.setAccessible(true);
+        return m;
+    }
+
+    private static List<String> callMethod(ZooKeeper zooKeeper, Method method)
+    {
+        if ( zooKeeper == null )
+        {
+            return null;
+        }
+        try
+        {
+            //noinspection unchecked
+            return (List<String>)method.invoke(zooKeeper);
+        }
+        catch ( Exception e )
+        {
+            throw new RuntimeException(e);
+        }
+    }
+}