Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

CURATOR-710: Fix leaking watch in EnsembleTracker #508

Merged
merged 4 commits into from
Jan 19, 2025
Merged

CURATOR-710: Fix leaking watch in EnsembleTracker #508

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
cd03c72
CURATOR-710: Fix leaking watch in EnsembleTracker
kezhuw Oct 16, 2024
6b78a3b
Comments a bit after review
kezhuw Nov 6, 2024
5cf5e1c
avoid mutable state
tisonkun Jan 16, 2025
9cc6766
Merge pull request #1 from tisonkun/patch
kezhuw Jan 16, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
13 changes: 8 additions & 5 deletions curator-framework/src/main/java/org/apache/curator/framework/imps/GetConfigBuilderImpl.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -33,21 +33,24 @@
public class GetConfigBuilderImpl
implements GetConfigBuilder, BackgroundOperation<Void>, ErrorListenerEnsembleable<byte[]> {
private final CuratorFrameworkImpl client;
private final WatcherRemovalManager watcherRemovalManager;

private Backgrounding backgrounding;
private Watching watching;
private Stat stat;

public GetConfigBuilderImpl(CuratorFrameworkImpl client) {
this.client = (CuratorFrameworkImpl) client.usingNamespace(null);
this.watcherRemovalManager = client.getWatcherRemovalManager();
backgrounding = new Backgrounding();
watching = new Watching(this.client);
watching = new Watching(this.client).setWatcherRemovalManager(watcherRemovalManager);
}

public GetConfigBuilderImpl(CuratorFrameworkImpl client, Backgrounding backgrounding, Watcher watcher, Stat stat) {
this.client = (CuratorFrameworkImpl) client.usingNamespace(null);
this.watcherRemovalManager = client.getWatcherRemovalManager();
this.backgrounding = backgrounding;
this.watching = new Watching(this.client, watcher);
this.watching = new Watching(this.client, watcher).setWatcherRemovalManager(watcherRemovalManager);
this.stat = stat;
}

Expand Down Expand Up @@ -110,19 +113,19 @@ public BackgroundEnsembleable<byte[]> usingWatcher(CuratorWatcher watcher) {

@Override
public BackgroundEnsembleable<byte[]> watched() {
watching = new Watching(client, true);
watching = new Watching(client, true).setWatcherRemovalManager(watcherRemovalManager);
return new InternalBackgroundEnsembleable();
}

@Override
public BackgroundEnsembleable<byte[]> usingWatcher(Watcher watcher) {
watching = new Watching(client, watcher);
watching = new Watching(client, watcher).setWatcherRemovalManager(watcherRemovalManager);
return new InternalBackgroundEnsembleable();
}

@Override
public BackgroundEnsembleable<byte[]> usingWatcher(CuratorWatcher watcher) {
watching = new Watching(client, watcher);
watching = new Watching(client, watcher).setWatcherRemovalManager(watcherRemovalManager);
return new InternalBackgroundEnsembleable();
}

Expand Down
16 changes: 12 additions & 4 deletions curator-framework/src/main/java/org/apache/curator/framework/imps/Watching.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -28,36 +28,46 @@ public class Watching {
private final CuratorWatcher curatorWatcher;
private final boolean watched;
private final CuratorFrameworkImpl client;
private WatcherRemovalManager watcherRemovalManager;
private NamespaceWatcher namespaceWatcher;

public Watching(CuratorFrameworkImpl client, boolean watched) {
this.client = client;
this.watcherRemovalManager = client.getWatcherRemovalManager();
this.watcher = null;
this.curatorWatcher = null;
this.watched = watched;
}

public Watching(CuratorFrameworkImpl client, Watcher watcher) {
this.client = client;
this.watcherRemovalManager = client.getWatcherRemovalManager();
this.watcher = watcher;
this.curatorWatcher = null;
this.watched = false;
}

public Watching(CuratorFrameworkImpl client, CuratorWatcher watcher) {
this.client = client;
this.watcherRemovalManager = client.getWatcherRemovalManager();
this.watcher = null;
this.curatorWatcher = watcher;
this.watched = false;
}

public Watching(CuratorFrameworkImpl client) {
this.client = client;
this.watcherRemovalManager = client.getWatcherRemovalManager();
watcher = null;
watched = false;
curatorWatcher = null;
}

Watching setWatcherRemovalManager(WatcherRemovalManager watcherRemovalManager) {
this.watcherRemovalManager = watcherRemovalManager;
return this;
}
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

It seems the watcherRemovalManager already set in the constructor by this.watcherRemovalManager = client.getWatcherRemovalManager();?

Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Or otherwise we can merge this setter into the constructor as a parameter.


Watcher getWatcher(String unfixedPath) {
namespaceWatcher = null;
if (watcher != null) {
Expand Down Expand Up @@ -85,10 +95,8 @@ void commitWatcher(int rc, boolean isExists) {
doCommit = (rc == KeeperException.Code.OK.intValue());
}

if (doCommit && (namespaceWatcher != null)) {
if (client.getWatcherRemovalManager() != null) {
client.getWatcherRemovalManager().add(namespaceWatcher);
}
if (doCommit && namespaceWatcher != null && watcherRemovalManager != null) {
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Can you elaborate a bit when watcherRemovalManager can be different from client.getWatcherRemovalManager().

Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I agree with the question: we can save some memory by not having the field. The performance impact is negligible

Copy link
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

  1. Given a fresh new CuratorFrameworkImpl, say framework0.
  2. Let's assume framework1 = framework0.newWatcherRemoveCuratorFramework(). framework1.getWatcherRemovalManager will be the WatcherRemovalManager.
  3. Let's assume framework2 = frame1.usingNamespace(null). framework2 will loss WatcherRemovalManager from above.
  4. framework1.removeWatchers will not drop watchers from new Watching(framework2, ...).

Step.2 is what exactly EnsembleTracker does.

curator/curator-framework/src/main/java/org/apache/curator/framework/imps/EnsembleTracker.java

Lines 87 to 90 in 6a27ff0

EnsembleTracker(CuratorFramework client, EnsembleProvider ensembleProvider) {
this.client = client.newWatcherRemoveCuratorFramework();
this.ensembleProvider = ensembleProvider;
}

Step.3 is what exactly GetConfigBuilderImpl(CURATOR-667(#474)) does currently.

curator/curator-framework/src/main/java/org/apache/curator/framework/imps/GetConfigBuilderImpl.java

Lines 41 to 45 in 6a27ff0

public GetConfigBuilderImpl(CuratorFrameworkImpl client) {
this.client = (CuratorFrameworkImpl) client.usingNamespace(null);
backgrounding = new Backgrounding();
watching = new Watching(this.client);
}

Step.4 is where this bug emerges.

curator/curator-framework/src/main/java/org/apache/curator/framework/imps/EnsembleTracker.java

Lines 99 to 104 in 6a27ff0

public void close() {
if (state.compareAndSet(State.STARTED, State.CLOSED)) {
client.removeWatchers();
client.getConnectionStateListenable().removeListener(connectionStateListener);
}
}

I have pushed a fixup commit 6b78a3b with comments to doc this.

Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I get the point now. Let me check the code to see if we can have other less stateful/mutable solution 😆

Copy link
Member

@tisonkun tisonkun Jan 16, 2025 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Here is a patch to avoid the later set kezhuw#1.

And I agree that we need some effort for #517 to keep the inheritance hierarchy easier to discover and extend.

Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Briefly, the root cause is WatcherRemovalFacade::usingNamespace drop the removalManager. Then by convey the manager in the method, things should work well.

Copy link
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I saw the patch. I think it is what I am trying to express in #517 (comment).

whether CuratorFramework::usingNamespace should inherit functionalities of this ?

It changes the behavior of CuratorFramework::usingNamespace, but I am ok.

Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Yeah .. I suppose it's more like a bug to fix, since CURATOR-710 indicates that it's a bug anyway.

To provide an exit gate, perhaps we can add a new method to return a "basic CuratorFramework" to drop the removalManager explicitly, just like usingNamespace(null) to drop the namespace in facade. But we can leave it to #517.

watcherRemovalManager.add(namespaceWatcher);
}
}
}
64 changes: 64 additions & 0 deletions ...-framework/src/test/java/org/apache/curator/framework/imps/TestWatcherRemovalManager.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -22,6 +22,7 @@
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertTrue;
import static org.junit.jupiter.api.Assertions.fail;
import java.util.Collections;
import java.util.List;
import java.util.concurrent.Callable;
import java.util.concurrent.CountDownLatch;
Expand All @@ -34,12 +35,27 @@
import org.apache.curator.test.Timing;
import org.apache.curator.test.WatchersDebug;
import org.apache.curator.test.compatibility.CuratorTestBase;
import org.apache.curator.utils.DebugUtils;
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.ZooDefs;
import org.apache.zookeeper.server.quorum.QuorumPeer;
import org.apache.zookeeper.server.quorum.flexible.QuorumMaj;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;

public class TestWatcherRemovalManager extends CuratorTestBase {
private static final String superUserPasswordDigest = "curator-test:zghsj3JfJqK7DbWf0RQ1BgbJH9w="; // ran from
private static final String superUserPassword = "curator-test";

@BeforeEach
@Override
public void setup() throws Exception {
System.setProperty("zookeeper.DigestAuthenticationProvider.superDigest", superUserPasswordDigest);
super.setup();
}

@Test
public void testSameWatcherDifferentPaths1Triggered() throws Exception {
CuratorFramework client = CuratorFrameworkFactory.newClient(server.getConnectString(), new RetryOneTime(1));
Expand Down Expand Up @@ -302,6 +318,54 @@ public void testBasicNamespace3() throws Exception {
}
}

@Test
public void testEnsembleTracker() throws Exception {
// given: client with ensemble tracker
CuratorFramework client = CuratorFrameworkFactory.builder()
.connectString(server.getConnectString())
.retryPolicy(new RetryOneTime(1))
.namespace("hey")
.ensembleTracker(true)
.authorization("digest", superUserPassword.getBytes())
.build();
try {
client.start();

// We are using standalone, so "/zookeeper/config" will be empty.
// So let's set it directly.
QuorumMaj quorumMaj = new QuorumMaj(Collections.singletonMap(
1L,
new QuorumPeer.QuorumServer(1, "127.0.0.1:2182:2183:participant;" + server.getConnectString())));
quorumMaj.setVersion(1);
client.usingNamespace(null)
.setData()
.forPath(ZooDefs.CONFIG_NODE, quorumMaj.toString().getBytes());

// when: zookeeper config node data fetched
while (client.getCurrentConfig().getVersion() == 0) {
Thread.sleep(100);
}

// then: the watcher must be attached
assertEquals(
1,
WatchersDebug.getDataWatches(client.getZookeeperClient().getZooKeeper())
.size());

// when: ensemble tracker closed
System.setProperty(DebugUtils.PROPERTY_REMOVE_WATCHERS_IN_FOREGROUND, "true");
((CuratorFrameworkImpl) client).getEnsembleTracker().close();

// then: the watcher must be removed
assertEquals(
0,
WatchersDebug.getDataWatches(client.getZookeeperClient().getZooKeeper())
.size());
} finally {
TestCleanState.closeAndTestClean(client);
}
}

@Test
public void testSameWatcher() throws Exception {
CuratorFramework client = CuratorFrameworkFactory.newClient(server.getConnectString(), new RetryOneTime(1));
Expand Down
Loading