GH-1170: Fix watcher leak in DistributedBarrier#waitOnBarrier

curator-recipes/src/main/java/org/apache/curator/framework/recipes/barriers/DistributedBarrier.java

@@ -22,9 +22,12 @@
 import java.util.concurrent.TimeUnit;
 import org.apache.curator.framework.CuratorFramework;
 import org.apache.curator.utils.PathUtils;
+import org.apache.curator.utils.ThreadUtils;
 import org.apache.zookeeper.KeeperException;
 import org.apache.zookeeper.WatchedEvent;
 import org.apache.zookeeper.Watcher;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 /**
  * <p>
@@ -37,6 +40,8 @@
  * </blockquote>
  */
 public class DistributedBarrier {
+    private static final Logger log = LoggerFactory.getLogger(DistributedBarrier.class);
+
     private final CuratorFramework client;
     private final String barrierPath;
     private final Watcher watcher = new Watcher() {
@@ -104,23 +109,45 @@ public synchronized boolean waitOnBarrier(long maxWait, TimeUnit unit) throws Ex
         long maxWaitMs = hasMaxWait ? TimeUnit.MILLISECONDS.convert(maxWait, unit) : Long.MAX_VALUE;
 
         boolean result;
-        for (; ; ) {
-            result = (client.checkExists().usingWatcher(watcher).forPath(barrierPath) == null);
-            if (result) {
-                break;
-            }
-
-            if (hasMaxWait) {
-                long elapsed = System.currentTimeMillis() - startMs;
-                long thisWaitMs = maxWaitMs - elapsed;
-                if (thisWaitMs <= 0) {
+        try {
+            for (; ; ) {
+                result = (client.checkExists().usingWatcher(watcher).forPath(barrierPath) == null);
+                if (result) {
                     break;
                 }
-                wait(thisWaitMs);
-            } else {
-                wait();
+
+                if (hasMaxWait) {
+                    long elapsed = System.currentTimeMillis() - startMs;
+                    long thisWaitMs = maxWaitMs - elapsed;
+                    if (thisWaitMs <= 0) {
+                        break;
+                    }
+                    wait(thisWaitMs);
+                } else {
+                    wait();
+                }
             }
+        } finally {
+            // Each loop iteration above registers a new exists-watch via usingWatcher(). If we
+            // return before that watch has actually fired (e.g. the barrier node didn't exist yet,
+            // or we gave up after maxWait elapsed), the watch is left dangling on both the client
+            // and the server until the node is eventually created - or forever, in a system where
+            // barriers are created and waited-on repeatedly but the node happens to never appear
+            // again. Proactively remove it here so repeated use of a DistributedBarrier doesn't
+            // leak watches. If the watch already fired (the common/successful case), removal is a
+            // no-op handled quietly.
+            removeDanglingWatcher();
         }
         return result;
     }
+
+    private void removeDanglingWatcher() {
+        try {
+            client.watches().remove(watcher).quietly().forPath(barrierPath);
+        } catch (Exception e) {
+            if (!ThreadUtils.checkInterrupted(e)) {
+                log.debug("Could not remove barrier watcher for path: {}", barrierPath, e);
+            }
+        }
+    }
 }

curator-recipes/src/test/java/org/apache/curator/framework/recipes/barriers/TestDistributedBarrier.java

@@ -19,9 +19,11 @@
 
 package org.apache.curator.framework.recipes.barriers;
 
+import static org.junit.jupiter.api.Assertions.assertThrows;
 import static org.junit.jupiter.api.Assertions.assertTrue;
 import static org.junit.jupiter.api.Assertions.fail;
 import com.google.common.collect.Lists;
+import java.lang.reflect.Field;
 import java.util.List;
 import java.util.concurrent.Callable;
 import java.util.concurrent.ExecutorService;
@@ -34,6 +36,7 @@
 import org.apache.curator.test.BaseClassForTests;
 import org.apache.curator.utils.CloseableUtils;
 import org.apache.zookeeper.KeeperException;
+import org.apache.zookeeper.Watcher;
 import org.junit.jupiter.api.Test;
 
 public class TestDistributedBarrier extends BaseClassForTests {
@@ -153,6 +156,41 @@ public Object call() throws Exception {
         }
     }
 
+    /*
+     * GH-1170: waitOnBarrier() registers a new exists-watch (via usingWatcher()) on every loop
+     * iteration. When the barrier node doesn't exist, waitOnBarrier() returns immediately without
+     * that watch ever firing, leaving it dangling. In a system that creates and waits on many
+     * barriers, those dangling watches accumulate without bound. waitOnBarrier() must clean up
+     * the watch it just registered before returning.
+     */
+    @Test
+    public void testNoBarrierDoesNotLeakWatch() throws Exception {
+        CuratorFramework client = CuratorFrameworkFactory.newClient(server.getConnectString(), new RetryOneTime(1));
+        try {
+            client.start();
+
+            DistributedBarrier barrier = new DistributedBarrier(client, "/barrier-no-leak");
+            assertTrue(barrier.waitOnBarrier(10, TimeUnit.SECONDS));
+
+            Watcher watcher = getBarrierWatcher(barrier);
+
+            // If waitOnBarrier() already removed the watch it registered, trying to remove it
+            // again must fail with NoWatcherException because there is nothing left to remove.
+            // Before the fix, the watch was still registered, so this removal would succeed.
+            assertThrows(
+                    KeeperException.NoWatcherException.class,
+                    () -> client.watches().remove(watcher).forPath("/barrier-no-leak"));
+        } finally {
+            client.close();
+        }
+    }
+
+    private static Watcher getBarrierWatcher(DistributedBarrier barrier) throws Exception {
+        Field watcherField = DistributedBarrier.class.getDeclaredField("watcher");
+        watcherField.setAccessible(true);
+        return (Watcher) watcherField.get(barrier);
+    }
+
     @Test
     public void testBasic() throws Exception {
         CuratorFramework client = CuratorFrameworkFactory.newClient(server.getConnectString(), new RetryOneTime(1));