From d9c1c139d01b95e2229dbed989c590fc2ad43e38 Mon Sep 17 00:00:00 2001 From: Sanskar121543 Date: Thu, 15 Oct 2026 18:23:00 -0400 Subject: [PATCH] GH-1170: Fix watcher leak in DistributedBarrier#waitOnBarrier Each iteration of the wait loop registers a new exists-watch on the barrier path via usingWatcher(). When the barrier node doesn't exist, or when maxWait elapses before the watch fires, waitOnBarrier() returned without ever removing that watch. In a system that creates and waits on many DistributedBarriers, those dangling watches accumulate on both the client and the server without bound, which can eventually exhaust memory. Remove the watch waitOnBarrier() just registered before returning, in a finally block so it runs on every exit path (barrier already gone, maxWait elapsed, or an exception). Removal is quiet: if the watch already fired (the common, successful case), there's nothing left to remove and this is a no-op. --- .../recipes/barriers/DistributedBarrier.java | 53 ++++++++++++++----- .../barriers/TestDistributedBarrier.java | 38 +++++++++++++ 2 files changed, 78 insertions(+), 13 deletions(-) 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));