This is an automated email from the ASF dual-hosted git repository.
kezhuw pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/curator.git
The following commit(s) were added to refs/heads/master by this push:
new 0d09f0250 GH-1314: Fix PersistentWatcher with CuratorFramework
implementations that are not CuratorFrameworkBase (#1315)
0d09f0250 is described below
commit 0d09f025051f3a4def75bdeb472b0e7a20a0b12b
Author: Andrey D. Shindarev <[email protected]>
AuthorDate: Mon Oct 26 15:01:37 2026 +0300
GH-1314: Fix PersistentWatcher with CuratorFramework implementations that
are not CuratorFrameworkBase (#1315)
Use the underlying client only for CuratorFrameworkBase instances
(namespaced
facades, #1259) and fall back to the public getCuratorListenable()
otherwise,
so that PersistentWatcher and CuratorCache work with any CuratorFramework
implementation again, e.g. a decorator of the public interface.
Fixes #1314.
---
.../framework/recipes/watch/PersistentWatcher.java | 13 +++-
.../recipes/watch/TestPersistentWatcher.java | 79 ++++++++++++++++++++++
2 files changed, 89 insertions(+), 3 deletions(-)
diff --git
a/curator-recipes/src/main/java/org/apache/curator/framework/recipes/watch/PersistentWatcher.java
b/curator-recipes/src/main/java/org/apache/curator/framework/recipes/watch/PersistentWatcher.java
index 8cc1e9d63..b994fe139 100644
---
a/curator-recipes/src/main/java/org/apache/curator/framework/recipes/watch/PersistentWatcher.java
+++
b/curator-recipes/src/main/java/org/apache/curator/framework/recipes/watch/PersistentWatcher.java
@@ -87,11 +87,18 @@ public class PersistentWatcher implements Closeable {
public void start() {
Preconditions.checkState(state.compareAndSet(State.LATENT,
State.STARTED), "Already started");
client.getConnectionStateListenable().addListener(connectionStateListener);
- // This could be a namespaced facade which does not support
getCuratorListenable.
- ((CuratorFrameworkBase)
client).client().getCuratorListenable().addListener(clientCloseListener);
+ curatorListenable().addListener(clientCloseListener);
reset();
}
+ private Listenable<CuratorListener> curatorListenable() {
+ // A namespaced facade does not support getCuratorListenable(), so use
the underlying client (#1259).
+ // Other CuratorFramework implementations are handled through the
public API.
+ return client instanceof CuratorFrameworkBase
+ ? ((CuratorFrameworkBase)
client).client().getCuratorListenable()
+ : client.getCuratorListenable();
+ }
+
/**
* Remove the watcher
*/
@@ -99,7 +106,7 @@ public class PersistentWatcher implements Closeable {
public void close() {
if (state.compareAndSet(State.STARTED, State.CLOSED)) {
listeners.clear();
- ((CuratorFrameworkBase)
client).client().getCuratorListenable().removeListener(clientCloseListener);
+ curatorListenable().removeListener(clientCloseListener);
client.getConnectionStateListenable().removeListener(connectionStateListener);
try {
client.watchers().remove(watcher).guaranteed().inBackground().forPath(basePath);
diff --git
a/curator-recipes/src/test/java/org/apache/curator/framework/recipes/watch/TestPersistentWatcher.java
b/curator-recipes/src/test/java/org/apache/curator/framework/recipes/watch/TestPersistentWatcher.java
index 48a2c6ee9..485b1aec3 100644
---
a/curator-recipes/src/test/java/org/apache/curator/framework/recipes/watch/TestPersistentWatcher.java
+++
b/curator-recipes/src/test/java/org/apache/curator/framework/recipes/watch/TestPersistentWatcher.java
@@ -22,6 +22,8 @@ package org.apache.curator.framework.recipes.watch;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertNotNull;
import static org.junit.jupiter.api.Assertions.assertTrue;
+import java.lang.reflect.InvocationTargetException;
+import java.lang.reflect.Proxy;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.LinkedBlockingQueue;
@@ -155,6 +157,83 @@ public class TestPersistentWatcher extends CuratorTestBase
{
}
}
+ // GH-1314 PersistentWatcher with a CuratorFramework that is not a
CuratorFrameworkBase
+ @Test
+ public void testDelegatingClientWatching() throws Exception {
+ BlockingQueue<WatchedEvent> events = new LinkedBlockingQueue<>();
+
+ try (CuratorFramework client = CuratorFrameworkFactory.newClient(
+ server.getConnectString(), timing.session(),
timing.connection(), new RetryOneTime(1))) {
+ client.start();
+ // given: connected curator client
+ client.blockUntilConnected();
+
+ // given: started persistent watcher with a delegating client
+ PersistentWatcher persistentWatcher = new
PersistentWatcher(delegatingClient(client), "/top/main", true);
+ persistentWatcher.getListenable().addListener(events::add);
+ persistentWatcher.start();
+
+ // when: create paths
+ client.create().creatingParentsIfNeeded().forPath("/top/main/a");
+
+ // then: receive node watch events
+ WatchedEvent event1 = events.poll(5, TimeUnit.SECONDS);
+ assertNotNull(event1);
+ assertEquals(Watcher.Event.EventType.NodeCreated,
event1.getType());
+ assertEquals("/top/main", event1.getPath());
+
+ WatchedEvent event2 = events.poll(5, TimeUnit.SECONDS);
+ assertNotNull(event2);
+ assertEquals(Watcher.Event.EventType.NodeCreated,
event2.getType());
+ assertEquals("/top/main/a", event2.getPath());
+ }
+
+ // when: curator client closed
+ // then: listener get Closed notification
+ WatchedEvent event = events.poll(5, TimeUnit.SECONDS);
+ assertNotNull(event);
+ assertEquals(Watcher.Event.EventType.None, event.getType());
+ assertEquals(Watcher.Event.KeeperState.Closed, event.getState());
+ }
+
+ // GH-1314 Curator listeners set up through a delegating client are
de-registered on close
+ @Test
+ public void testDelegatingClientCuratorListenersNotLeaking() {
+ try (CuratorFramework client = CuratorFrameworkFactory.newClient(
+ server.getConnectString(), timing.session(),
timing.connection(), new RetryOneTime(1))) {
+ client.start();
+ CuratorFramework delegatingClient = delegatingClient(client);
+
+ for (int i = 0; i < 100; i++) {
+ try (PersistentWatcher persistentWatcher = new
PersistentWatcher(delegatingClient, "/top/main", true)) {
+ persistentWatcher.start();
+ }
+ }
+
+ assertEquals(
+ 0,
+ ((StandardListenerManager<?>)
client.getCuratorListenable()).size(),
+ "Curator listeners set up by the now closed
PersistentWatchers should have been de-registered");
+ }
+ }
+
+ /**
+ * A {@link CuratorFramework} implemented outside of Curator: it forwards
every call to the given client
+ * and is not a {@link
org.apache.curator.framework.imps.CuratorFrameworkBase}.
+ */
+ private static CuratorFramework delegatingClient(CuratorFramework client) {
+ return (CuratorFramework) Proxy.newProxyInstance(
+ CuratorFramework.class.getClassLoader(),
+ new Class<?>[] {CuratorFramework.class},
+ (proxy, method, args) -> {
+ try {
+ return method.invoke(client, args);
+ } catch (InvocationTargetException e) {
+ throw e.getCause();
+ }
+ });
+ }
+
private void internalTest(boolean recursive) throws Exception {
try (CuratorFramework client = CuratorFrameworkFactory.newClient(
server.getConnectString(), timing.session(),
timing.connection(), new RetryOneTime(1))) {
