[Sanskar121543]

What

Fixes #1170 (CURATOR-654).

DistributedBarrier#waitOnBarrier registers a ZK exists-watch on the barrier path on every loop iteration via client.checkExists().usingWatcher(watcher).forPath(barrierPath), but never removes it. In particular:

• If the barrier node doesn't exist when waitOnBarrier() is called, checkExists() still registers the watch (ZooKeeper sets exists-watches regardless of whether the node is currently present, so it can notify on creation), but the method returns immediately (result == true) without the watch ever firing.
• If maxWait elapses before the node is removed, the method returns (false) with the last-registered watch still outstanding.

In a system that creates and waits on DistributedBarriers repeatedly (the scenario described in the original report), each such call leaves behind a dangling watch that is never cleaned up, client- and server-side, until the watched node happens to be created/changed - or never, which is how this leads to unbounded memory growth over time.

This PR was previously attempted in #435, which was closed as inactive; I picked it up as suggested in the issue thread, using Curator's current watch-removal API instead of that PR's approach.

Fix

Wrap the wait loop in a try/finally and remove the watch waitOnBarrier() just registered before returning, using client.watches().remove(watcher).quietly().forPath(barrierPath). .quietly() suppresses NoWatcherException, so this is a safe no-op on the common/successful path where the watch already fired (e.g. the barrier was removed and checkExists() returned null because of that event, not because the node was never there). Any other failure during cleanup (e.g. connection issues) is caught and logged at debug level rather than propagated, so it can't mask the real result of waitOnBarrier() or swallow an exception already in flight.

Testing

Added testNoBarrierDoesNotLeakWatch to TestDistributedBarrier, which:

1. Calls waitOnBarrier() on a barrier path that doesn't exist (same setup as the existing testNoBarrier test).
2. Grabs the DistributedBarrier's private watcher via reflection.
3. Asserts that removing that exact watcher via client.watches().remove(watcher).forPath(path) (without .quietly()) throws KeeperException.NoWatcherException - i.e. there's nothing left to remove, because waitOnBarrier() already cleaned it up.

I verified this test fails against the unpatched code (the removal succeeds instead of throwing, because the watch is still dangling) and passes with the fix. Ran the full TestDistributedBarrier suite (5/5 passing) against a real embedded ZooKeeper via BaseClassForTests/TestingServer - no broker mocking involved.