diff --git a/curator-recipes/src/main/java/org/apache/curator/framework/recipes/barriers/DistributedBarrier.java b/curator-recipes/src/main/java/org/apache/curator/framework/recipes/barriers/DistributedBarrier.java index 7894d106f..587ffb4d3 100644 --- a/curator-recipes/src/main/java/org/apache/curator/framework/recipes/barriers/DistributedBarrier.java +++ b/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; /** *

@@ -37,6 +40,8 @@ * */ 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); + } + } + } } diff --git a/curator-recipes/src/test/java/org/apache/curator/framework/recipes/barriers/TestDistributedBarrier.java b/curator-recipes/src/test/java/org/apache/curator/framework/recipes/barriers/TestDistributedBarrier.java index 8646f99c3..35e38355e 100644 --- a/curator-recipes/src/test/java/org/apache/curator/framework/recipes/barriers/TestDistributedBarrier.java +++ b/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));