This is an automated email from the ASF dual-hosted git repository.
asf-gitbox-commits pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/helix.git
The following commit(s) were added to refs/heads/master by this push:
new 5e7d6d7f9 Add opt-in persistent watcher mode for Helix ZkClients
(#2309)
5e7d6d7f9 is described below
commit 5e7d6d7f972080820742b3308abca3f6d30e2c6c
Author: Junkai Xue <[email protected]>
AuthorDate: Fri Nov 6 13:12:04 2026 -0800
Add opt-in persistent watcher mode for Helix ZkClients (#2309)
ZkClient has supported registering persistent ZooKeeper watches for its
listeners since 2023 (usePersistWatcher), which removes the window in which
a
one-time watch is being re-armed and avoids the overwrite hazard that made
one-time and persistent watches on the same path unsafe before ZooKeeper 3.9
(ZOOKEEPER-4466). Until now the flag could only be set on the raw ZkClient
constructor and only meta-client used it; HelixManager, HelixAdmin,
helix-rest
and every client built through RealmAwareZkClientConfig were stuck with
one-time watches, which is the mode suspected of missing notifications in
#2309.
Turning the mode on for a whole cluster immediately exposed a bug in
ZkClient that had been masked because only meta-client used the mode:
addPersistListener/removePersistListener took an Object and decided by
instanceof, testing IZkChildListener first. CallbackHandler implements both
IZkChildListener and IZkDataListener, so in persist mode every data
subscription a HelixManager made (ideal states, current states, configs, the
leader node) was registered as a child subscription and no NodeDataChanged
was ever delivered to it; a controller stopped reacting after the first
current state write and the cluster never converged. The two methods are now
typed overloads, so the subscribe call decides the listener kind.
This change makes the mode configurable, default off:
- RealmAwareZkClientConfig (and HelixZkClient.ZkClientConfig) gain
setUsePersistWatcher/isUsePersistWatcher. The default comes from the new
system property zk.zkclient.persistwatcher.enabled, so a whole process can
be switched (e.g. a CI run of the test suite) without code changes.
- DedicatedZkClientFactory, DedicatedZkClient, FederatedZkClient and
SharedZkClientFactory.InnerSharedZkClient pass the flag to the ZkClient
they build, using the same connectOnInit=true the shorter constructor
used.
- ZKHelixManager honors a RealmAwareZkClientConfig supplied through
HelixManagerProperty as an explicit override of the system property and
logs the mode it creates its client with.
- ParticipantManager read the existing live instance with a one-time watch
nobody listened for; a client in persist watcher mode refuses one-time
watches, so it now uses readDataAndStat without a watch. That read also
returns null instead of throwing when the live instance disappeared in the
meantime, which is what the following retry logic always expected.
Tests: TestZkClientPersistWatcher covers a listener implementing both
interfaces. TestDedicatedZkClient, TestSharedZkClient and
TestFederatedZkClient
verify that a config with the flag reaches the built client (one-time
watches
refused, listeners notified through persistent watches).
TestZKHelixManagerPersistWatcher runs a controller and two participants to
convergence with the system property set, verifies their clients hold only
persistent watches, and verifies a HelixManagerProperty config overrides the
property. The ZkTestHelper.getZkWatch helpers now also report persistent
watches.
Enabling the mode for the full helix-core suite in CI and flipping the
default are follow-ups; the test prints the cluster state when it does not
converge so such runs are diagnosable.
---
.../helix/manager/zk/ParticipantManager.java | 4 +-
.../apache/helix/manager/zk/ZKHelixManager.java | 8 +
.../test/java/org/apache/helix/ZkTestHelper.java | 12 ++
.../manager/TestZKHelixManagerPersistWatcher.java | 188 +++++++++++++++++++++
.../helix/zookeeper/api/client/HelixZkClient.java | 6 +
.../zookeeper/api/client/RealmAwareZkClient.java | 21 ++-
.../zookeeper/constant/ZkSystemPropertyKeys.java | 15 ++
.../zookeeper/impl/client/DedicatedZkClient.java | 3 +-
.../zookeeper/impl/client/FederatedZkClient.java | 3 +-
.../zookeeper/impl/client/SharedZkClient.java | 1 +
.../impl/factory/DedicatedZkClientFactory.java | 3 +-
.../impl/factory/SharedZkClientFactory.java | 2 +-
.../apache/helix/zookeeper/zkclient/ZkClient.java | 88 +++++-----
.../apache/helix/zookeeper/impl/ZkTestHelper.java | 12 ++
.../client/RealmAwareZkClientFactoryTestBase.java | 57 +++++++
.../impl/client/TestFederatedZkClient.java | 45 +++++
.../zkclient/TestZkClientPersistWatcher.java | 66 ++++++++
17 files changed, 490 insertions(+), 44 deletions(-)
diff --git
a/helix-core/src/main/java/org/apache/helix/manager/zk/ParticipantManager.java
b/helix-core/src/main/java/org/apache/helix/manager/zk/ParticipantManager.java
index 20910c891..8769f9854 100644
---
a/helix-core/src/main/java/org/apache/helix/manager/zk/ParticipantManager.java
+++
b/helix-core/src/main/java/org/apache/helix/manager/zk/ParticipantManager.java
@@ -312,7 +312,9 @@ public class ParticipantManager {
_clusterName);
Stat stat = new Stat();
- ZNRecord record = _zkclient.readData(liveInstancePath, stat, true);
+ // No watch is needed here (nobody listens for it, and a client in
persist-watcher mode
+ // rejects one-time watches); return null instead of throwing if the
node is already gone.
+ ZNRecord record = _zkclient.readDataAndStat(liveInstancePath, stat,
true);
if (record == null) {
/**
* live-instance is gone as we check it, retry create live-instance
diff --git
a/helix-core/src/main/java/org/apache/helix/manager/zk/ZKHelixManager.java
b/helix-core/src/main/java/org/apache/helix/manager/zk/ZKHelixManager.java
index cf549de2d..f5df9c2c6 100644
--- a/helix-core/src/main/java/org/apache/helix/manager/zk/ZKHelixManager.java
+++ b/helix-core/src/main/java/org/apache/helix/manager/zk/ZKHelixManager.java
@@ -1535,6 +1535,14 @@ public class ZKHelixManager implements HelixManager,
IZkStateListener {
.setMonitorKey(_clusterName)
.setMonitorInstanceName(_instanceName)
.setMonitorRootPathOnly(isMonitorRootPathOnly());
+ // Persistent watches default to the zk.zkclient.persistwatcher.enabled
system property; a
+ // RealmAwareZkClientConfig given through HelixManagerProperty overrides
it.
+ if (_helixManagerProperty != null &&
_helixManagerProperty.getZkClientConfig() != null) {
+ clientConfig.setUsePersistWatcher(
+ _helixManagerProperty.getZkClientConfig().isUsePersistWatcher());
+ }
+ LOG.info("Creating ZkClient for {} {} in cluster {} with
usePersistWatcher={}", _instanceType,
+ _instanceName, _clusterName, clientConfig.isUsePersistWatcher());
if (_instanceType == InstanceType.ADMINISTRATOR) {
return resolveZkClient(SharedZkClientFactory.getInstance(),
_realmAwareZkConnectionConfig,
diff --git a/helix-core/src/test/java/org/apache/helix/ZkTestHelper.java
b/helix-core/src/test/java/org/apache/helix/ZkTestHelper.java
index 3632ebfa1..fb73b53ad 100644
--- a/helix-core/src/test/java/org/apache/helix/ZkTestHelper.java
+++ b/helix-core/src/test/java/org/apache/helix/ZkTestHelper.java
@@ -417,9 +417,21 @@ public class ZkTestHelper {
field2.setAccessible(true);
Map<String, Set<Watcher>> childWatches = (Map<String, Set<Watcher>>)
field2.get(watchManager);
+ field2 = getField(watchManager.getClass(), "persistentWatches");
+ field2.setAccessible(true);
+ Map<String, Set<Watcher>> persistentWatches =
+ (Map<String, Set<Watcher>>) field2.get(watchManager);
+
+ field2 = getField(watchManager.getClass(), "persistentRecursiveWatches");
+ field2.setAccessible(true);
+ Map<String, Set<Watcher>> persistentRecursiveWatches =
+ (Map<String, Set<Watcher>>) field2.get(watchManager);
+
lists.put("dataWatches", new ArrayList<>(dataWatches.keySet()));
lists.put("existWatches", new ArrayList<>(existWatches.keySet()));
lists.put("childWatches", new ArrayList<>(childWatches.keySet()));
+ lists.put("persistentWatches", new
ArrayList<>(persistentWatches.keySet()));
+ lists.put("persistentRecursiveWatches", new
ArrayList<>(persistentRecursiveWatches.keySet()));
return lists;
}
diff --git
a/helix-core/src/test/java/org/apache/helix/integration/manager/TestZKHelixManagerPersistWatcher.java
b/helix-core/src/test/java/org/apache/helix/integration/manager/TestZKHelixManagerPersistWatcher.java
new file mode 100644
index 000000000..ca7b7b99c
--- /dev/null
+++
b/helix-core/src/test/java/org/apache/helix/integration/manager/TestZKHelixManagerPersistWatcher.java
@@ -0,0 +1,188 @@
+package org.apache.helix.integration.manager;
+
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+
+import org.apache.helix.HelixManagerProperty;
+import org.apache.helix.InstanceType;
+import org.apache.helix.PropertyPathBuilder;
+import org.apache.helix.TestHelper;
+import org.apache.helix.ZkTestHelper;
+import org.apache.helix.ZkUnitTestBase;
+import org.apache.helix.manager.zk.ZKHelixManager;
+import
org.apache.helix.tools.ClusterVerifiers.BestPossibleExternalViewVerifier;
+import org.apache.helix.tools.ClusterVerifiers.ZkHelixClusterVerifier;
+import org.apache.helix.zookeeper.api.client.RealmAwareZkClient;
+import org.apache.helix.zookeeper.constant.ZkSystemPropertyKeys;
+import org.apache.zookeeper.data.Stat;
+import org.testng.Assert;
+import org.testng.annotations.AfterClass;
+import org.testng.annotations.BeforeClass;
+import org.testng.annotations.Test;
+
+/**
+ * Runs a small cluster with the zk.zkclient.persistwatcher.enabled system
property set, so that
+ * every HelixManager registers persistent ZooKeeper watches instead of
one-time watches, and checks
+ * that a HelixManagerProperty client config can override the property per
manager.
+ */
+public class TestZKHelixManagerPersistWatcher extends ZkUnitTestBase {
+ private static final int NUM_NODES = 2;
+ private static final int START_PORT = 12918;
+
+ private final String _clusterName = TestHelper.getTestClassName();
+ private final List<MockParticipantManager> _participants = new ArrayList<>();
+ private ClusterControllerManager _controller;
+ private String _originalPersistWatcherProperty;
+
+ @BeforeClass
+ public void beforeClass() throws Exception {
+ _originalPersistWatcherProperty =
+ System.getProperty(ZkSystemPropertyKeys.ZK_PERSIST_WATCHER_ENABLED);
+ System.setProperty(ZkSystemPropertyKeys.ZK_PERSIST_WATCHER_ENABLED,
"true");
+
+ TestHelper.setupCluster(_clusterName, ZK_ADDR, START_PORT, "localhost",
"TestDB", 1, 4,
+ NUM_NODES, 2, "MasterSlave", true);
+ _controller = new ClusterControllerManager(ZK_ADDR, _clusterName,
"controller_0");
+ _controller.syncStart();
+ for (int i = 0; i < NUM_NODES; i++) {
+ MockParticipantManager participant =
+ new MockParticipantManager(ZK_ADDR, _clusterName, "localhost_" +
(START_PORT + i));
+ participant.syncStart();
+ _participants.add(participant);
+ }
+ }
+
+ @AfterClass
+ public void afterClass() {
+ try {
+ for (MockParticipantManager participant : _participants) {
+ if (participant.isConnected()) {
+ participant.syncStop();
+ }
+ }
+ if (_controller != null && _controller.isConnected()) {
+ _controller.syncStop();
+ }
+ deleteCluster(_clusterName);
+ } finally {
+ if (_originalPersistWatcherProperty == null) {
+ System.clearProperty(ZkSystemPropertyKeys.ZK_PERSIST_WATCHER_ENABLED);
+ } else {
+ System.setProperty(ZkSystemPropertyKeys.ZK_PERSIST_WATCHER_ENABLED,
+ _originalPersistWatcherProperty);
+ }
+ }
+ }
+
+ @Test
+ public void testClusterConvergesWithPersistentWatches() throws Exception {
+ ZkHelixClusterVerifier verifier =
+ new
BestPossibleExternalViewVerifier.Builder(_clusterName).setZkClient(_gZkClient)
+
.setWaitTillVerify(TestHelper.DEFAULT_REBALANCE_PROCESSING_WAIT_TIME).build();
+ try {
+ boolean converged = verifier.verifyByPolling();
+ if (!converged) {
+ dumpClusterState();
+ }
+ Assert.assertTrue(converged, "Cluster did not converge in persist
watcher mode");
+ } finally {
+ verifier.close();
+ }
+
+ // Every manager created while the property is set registers persistent
watches only.
+ List<RealmAwareZkClient> clients = new ArrayList<>();
+ clients.add(_controller.getZkClient());
+ for (MockParticipantManager participant : _participants) {
+ clients.add(participant.getZkClient());
+ }
+ for (RealmAwareZkClient client : clients) {
+ Map<String, List<String>> watches = ZkTestHelper.getZkWatch(client);
+ Assert.assertFalse(watches.get("persistentWatches").isEmpty(),
+ "Expected persistent watches, got " + watches);
+ Assert.assertTrue(watches.get("dataWatches").isEmpty(), "Unexpected
one-time watches " + watches);
+ Assert.assertTrue(watches.get("childWatches").isEmpty(), "Unexpected
one-time watches " + watches);
+ Assert.assertTrue(watches.get("existWatches").isEmpty(), "Unexpected
one-time watches " + watches);
+ }
+ }
+
+ @Test
+ public void testHelixManagerPropertyOverridesSystemProperty() throws
Exception {
+ // The system property turns the mode on; an explicit client config turns
it off for this manager.
+ HelixManagerProperty property = new
HelixManagerProperty.Builder().setRealmAwareZkClientConfig(
+ new
RealmAwareZkClient.RealmAwareZkClientConfig().setUsePersistWatcher(false)).build();
+ SpectatorWithProperty spectator = new SpectatorWithProperty(_clusterName,
"spectator_override",
+ property);
+ spectator.connect();
+ try {
+ String path = PropertyPathBuilder.idealState(_clusterName, "TestDB0");
+ // A one-time watch is accepted, which a client in persist watcher mode
refuses.
+ Assert.assertNotNull(spectator.getZkClient().readData(path, new Stat(),
true),
+ "The ideal state should be readable");
+ Map<String, List<String>> watches =
ZkTestHelper.getZkWatch(spectator.getZkClient());
+ Assert.assertTrue(watches.get("dataWatches").contains(path),
+ "Expected a one-time watch on " + path + ", got " + watches);
+ Assert.assertTrue(watches.get("persistentWatches").isEmpty(),
+ "Unexpected persistent watches " + watches);
+ } finally {
+ spectator.disconnect();
+ }
+ }
+
+ /** Prints what the controller and participants see so a non-converging run
is diagnosable. */
+ private void dumpClusterState() {
+ System.out.println("=== Cluster state of " + _clusterName + " ===");
+ for (String path : new String[] {
PropertyPathBuilder.liveInstance(_clusterName),
+ PropertyPathBuilder.idealState(_clusterName),
PropertyPathBuilder.externalView(_clusterName),
+ PropertyPathBuilder.controller(_clusterName) }) {
+ dumpSubtree(path, 0);
+ }
+ for (String instance :
_gZkClient.getChildren(PropertyPathBuilder.instance(_clusterName))) {
+ dumpSubtree(PropertyPathBuilder.instanceCurrentState(_clusterName,
instance), 0);
+ dumpSubtree(PropertyPathBuilder.instanceMessage(_clusterName, instance),
0);
+ }
+ }
+
+ private void dumpSubtree(String path, int depth) {
+ if (!_gZkClient.exists(path)) {
+ System.out.println(path + " (missing)");
+ return;
+ }
+ Object data = _gZkClient.readData(path, true);
+ System.out.println(path + (data == null ? "" : " -> " + data));
+ if (depth < 3) {
+ for (String child : _gZkClient.getChildren(path)) {
+ dumpSubtree(path + "/" + child, depth + 1);
+ }
+ }
+ }
+
+ private static class SpectatorWithProperty extends ZKHelixManager {
+ SpectatorWithProperty(String clusterName, String instanceName,
HelixManagerProperty property) {
+ super(clusterName, instanceName, InstanceType.SPECTATOR, ZK_ADDR, null,
property);
+ }
+
+ RealmAwareZkClient getZkClient() {
+ return _zkclient;
+ }
+ }
+}
diff --git
a/zookeeper-api/src/main/java/org/apache/helix/zookeeper/api/client/HelixZkClient.java
b/zookeeper-api/src/main/java/org/apache/helix/zookeeper/api/client/HelixZkClient.java
index f3810a953..8d02d9ebd 100644
---
a/zookeeper-api/src/main/java/org/apache/helix/zookeeper/api/client/HelixZkClient.java
+++
b/zookeeper-api/src/main/java/org/apache/helix/zookeeper/api/client/HelixZkClient.java
@@ -153,5 +153,11 @@ public interface HelixZkClient extends RealmAwareZkClient {
this._connectInitTimeout = connectInitTimeout;
return this;
}
+
+ @Override
+ public ZkClientConfig setUsePersistWatcher(boolean usePersistWatcher) {
+ this._usePersistWatcher = usePersistWatcher;
+ return this;
+ }
}
}
diff --git
a/zookeeper-api/src/main/java/org/apache/helix/zookeeper/api/client/RealmAwareZkClient.java
b/zookeeper-api/src/main/java/org/apache/helix/zookeeper/api/client/RealmAwareZkClient.java
index 4e9d890c4..04aaecee9 100644
---
a/zookeeper-api/src/main/java/org/apache/helix/zookeeper/api/client/RealmAwareZkClient.java
+++
b/zookeeper-api/src/main/java/org/apache/helix/zookeeper/api/client/RealmAwareZkClient.java
@@ -26,6 +26,7 @@ import
org.apache.helix.msdcommon.constant.MetadataStoreRoutingConstants;
import org.apache.helix.msdcommon.datamodel.MetadataStoreRoutingData;
import org.apache.helix.msdcommon.exception.InvalidRoutingDataException;
import org.apache.helix.zookeeper.constant.RoutingDataReaderType;
+import org.apache.helix.zookeeper.constant.ZkSystemPropertyKeys;
import org.apache.helix.zookeeper.routing.RoutingDataManager;
import org.apache.helix.zookeeper.util.ZNRecordUtil;
import org.apache.helix.zookeeper.zkclient.DataUpdater;
@@ -566,6 +567,9 @@ public interface RealmAwareZkClient {
protected String _monitorKey;
protected String _monitorInstanceName = null;
protected boolean _monitorRootPathOnly = true;
+ // Watcher mode, see ZkSystemPropertyKeys#ZK_PERSIST_WATCHER_ENABLED for
the default
+ protected boolean _usePersistWatcher =
+ Boolean.getBoolean(ZkSystemPropertyKeys.ZK_PERSIST_WATCHER_ENABLED);
public RealmAwareZkClientConfig setZkSerializer(PathBasedZkSerializer
zkSerializer) {
this._zkSerializer = zkSerializer;
@@ -622,6 +626,17 @@ public interface RealmAwareZkClient {
return this;
}
+ /**
+ * Whether the client registers persistent ZooKeeper watches for its
listeners instead of
+ * one-time watches that are re-armed after every notification. In this
mode the client refuses
+ * explicit one-time watches such as {@code readData(path, stat, true)}.
Defaults to the
+ * {@link ZkSystemPropertyKeys#ZK_PERSIST_WATCHER_ENABLED} system property.
+ */
+ public RealmAwareZkClientConfig setUsePersistWatcher(boolean
usePersistWatcher) {
+ this._usePersistWatcher = usePersistWatcher;
+ return this;
+ }
+
public PathBasedZkSerializer getZkSerializer() {
if (_zkSerializer == null) {
_zkSerializer = new BasicZkSerializer(new SerializableSerializer());
@@ -653,6 +668,10 @@ public interface RealmAwareZkClient {
return _connectInitTimeout;
}
+ public boolean isUsePersistWatcher() {
+ return _usePersistWatcher;
+ }
+
/**
* Create HelixZkClient.ZkClientConfig based on RealmAwareZkClientConfig.
* @return
@@ -662,7 +681,7 @@ public interface RealmAwareZkClient {
.setMonitorType(_monitorType).setMonitorKey(_monitorKey)
.setMonitorInstanceName(_monitorInstanceName).setMonitorRootPathOnly(_monitorRootPathOnly)
.setOperationRetryTimeout(_operationRetryTimeout)
- .setConnectInitTimeout(_connectInitTimeout);
+
.setConnectInitTimeout(_connectInitTimeout).setUsePersistWatcher(_usePersistWatcher);
}
}
diff --git
a/zookeeper-api/src/main/java/org/apache/helix/zookeeper/constant/ZkSystemPropertyKeys.java
b/zookeeper-api/src/main/java/org/apache/helix/zookeeper/constant/ZkSystemPropertyKeys.java
index 97e581665..f90abfce8 100644
---
a/zookeeper-api/src/main/java/org/apache/helix/zookeeper/constant/ZkSystemPropertyKeys.java
+++
b/zookeeper-api/src/main/java/org/apache/helix/zookeeper/constant/ZkSystemPropertyKeys.java
@@ -70,6 +70,21 @@ public class ZkSystemPropertyKeys {
public static final String ZK_AUTOSYNC_ENABLED =
"zk.zkclient.autosync.enabled";
+ /**
+ * Setting this property to {@code true} makes every ZkClient that is built
from a
+ * {@code RealmAwareZkClientConfig} (HelixManager, HelixAdmin, helix-rest,
...) register
+ * <b>persistent</b> ZooKeeper watches ({@code addWatch}) for its listeners
instead of one-time
+ * watches that have to be re-armed after every notification. Persistent
watches cannot miss a
+ * change between a notification and the re-registration of the watch. In
this mode the client
+ * refuses explicit one-time watches such as {@code readData(path, stat,
true)}.
+ * <p>
+ * An explicit {@code RealmAwareZkClientConfig#setUsePersistWatcher}
overrides this property.
+ * <p>
+ * The default value is {@code false}. Requires a ZooKeeper server that
supports persistent
+ * watches (3.6+; 3.9+ recommended, see ZOOKEEPER-4466 and ZOOKEEPER-4475).
+ */
+ public static final String ZK_PERSIST_WATCHER_ENABLED =
"zk.zkclient.persistwatcher.enabled";
+
/** System property key for jute.maxbuffer */
public static final String JUTE_MAXBUFFER = "jute.maxbuffer";
diff --git
a/zookeeper-api/src/main/java/org/apache/helix/zookeeper/impl/client/DedicatedZkClient.java
b/zookeeper-api/src/main/java/org/apache/helix/zookeeper/impl/client/DedicatedZkClient.java
index e21480629..d5ef42def 100644
---
a/zookeeper-api/src/main/java/org/apache/helix/zookeeper/impl/client/DedicatedZkClient.java
+++
b/zookeeper-api/src/main/java/org/apache/helix/zookeeper/impl/client/DedicatedZkClient.java
@@ -107,7 +107,8 @@ public class DedicatedZkClient implements
RealmAwareZkClient {
_rawZkClient = new ZkClient(zkConnection, (int)
clientConfig.getConnectInitTimeout(),
clientConfig.getOperationRetryTimeout(),
clientConfig.getZkSerializer(),
clientConfig.getMonitorType(), clientConfig.getMonitorKey(),
- clientConfig.getMonitorInstanceName(),
clientConfig.isMonitorRootPathOnly());
+ clientConfig.getMonitorInstanceName(),
clientConfig.isMonitorRootPathOnly(), true,
+ clientConfig.isUsePersistWatcher());
}
@Override
diff --git
a/zookeeper-api/src/main/java/org/apache/helix/zookeeper/impl/client/FederatedZkClient.java
b/zookeeper-api/src/main/java/org/apache/helix/zookeeper/impl/client/FederatedZkClient.java
index 7943bc27a..64f81ad8d 100644
---
a/zookeeper-api/src/main/java/org/apache/helix/zookeeper/impl/client/FederatedZkClient.java
+++
b/zookeeper-api/src/main/java/org/apache/helix/zookeeper/impl/client/FederatedZkClient.java
@@ -722,7 +722,8 @@ public class FederatedZkClient implements
RealmAwareZkClient {
return new ZkClient(new ZkConnection(zkAddress), (int)
_clientConfig.getConnectInitTimeout(),
_clientConfig.getOperationRetryTimeout(), _pathBasedZkSerializer,
_clientConfig.getMonitorType(), _clientConfig.getMonitorKey(),
- _clientConfig.getMonitorInstanceName(),
_clientConfig.isMonitorRootPathOnly());
+ _clientConfig.getMonitorInstanceName(),
_clientConfig.isMonitorRootPathOnly(), true,
+ _clientConfig.isUsePersistWatcher());
}
private void checkClosedState() {
diff --git
a/zookeeper-api/src/main/java/org/apache/helix/zookeeper/impl/client/SharedZkClient.java
b/zookeeper-api/src/main/java/org/apache/helix/zookeeper/impl/client/SharedZkClient.java
index 210e82bd6..38285f5c8 100644
---
a/zookeeper-api/src/main/java/org/apache/helix/zookeeper/impl/client/SharedZkClient.java
+++
b/zookeeper-api/src/main/java/org/apache/helix/zookeeper/impl/client/SharedZkClient.java
@@ -102,6 +102,7 @@ public class SharedZkClient implements RealmAwareZkClient {
zkClientConfig.setZkSerializer(clientConfig.getZkSerializer())
.setConnectInitTimeout(clientConfig.getConnectInitTimeout())
.setOperationRetryTimeout(clientConfig.getOperationRetryTimeout())
+ .setUsePersistWatcher(clientConfig.isUsePersistWatcher())
.setMonitorInstanceName(clientConfig.getMonitorInstanceName())
.setMonitorKey(clientConfig.getMonitorKey()).setMonitorType(clientConfig.getMonitorType())
.setMonitorRootPathOnly(clientConfig.isMonitorRootPathOnly());
diff --git
a/zookeeper-api/src/main/java/org/apache/helix/zookeeper/impl/factory/DedicatedZkClientFactory.java
b/zookeeper-api/src/main/java/org/apache/helix/zookeeper/impl/factory/DedicatedZkClientFactory.java
index bbccd22e9..4d792de2f 100644
---
a/zookeeper-api/src/main/java/org/apache/helix/zookeeper/impl/factory/DedicatedZkClientFactory.java
+++
b/zookeeper-api/src/main/java/org/apache/helix/zookeeper/impl/factory/DedicatedZkClientFactory.java
@@ -62,6 +62,7 @@ public class DedicatedZkClientFactory extends
HelixZkClientFactory {
return new ZkClient(createZkConnection(connectionConfig),
(int) clientConfig.getConnectInitTimeout(),
clientConfig.getOperationRetryTimeout(),
clientConfig.getZkSerializer(), clientConfig.getMonitorType(),
clientConfig.getMonitorKey(),
- clientConfig.getMonitorInstanceName(),
clientConfig.isMonitorRootPathOnly());
+ clientConfig.getMonitorInstanceName(),
clientConfig.isMonitorRootPathOnly(), true,
+ clientConfig.isUsePersistWatcher());
}
}
diff --git
a/zookeeper-api/src/main/java/org/apache/helix/zookeeper/impl/factory/SharedZkClientFactory.java
b/zookeeper-api/src/main/java/org/apache/helix/zookeeper/impl/factory/SharedZkClientFactory.java
index 48d1bdee9..a35789c53 100644
---
a/zookeeper-api/src/main/java/org/apache/helix/zookeeper/impl/factory/SharedZkClientFactory.java
+++
b/zookeeper-api/src/main/java/org/apache/helix/zookeeper/impl/factory/SharedZkClientFactory.java
@@ -159,7 +159,7 @@ public class SharedZkClientFactory extends
HelixZkClientFactory {
super(connectionManager.getConnection(), 0,
clientConfig.getOperationRetryTimeout(),
clientConfig.getZkSerializer(), clientConfig.getMonitorType(),
clientConfig.getMonitorKey(), clientConfig.getMonitorInstanceName(),
- clientConfig.isMonitorRootPathOnly());
+ clientConfig.isMonitorRootPathOnly(), true,
clientConfig.isUsePersistWatcher());
_connectionManager = connectionManager;
// Register to the base dedicated RealmAwareZkClient
_connectionManager.registerWatcher(this);
diff --git
a/zookeeper-api/src/main/java/org/apache/helix/zookeeper/zkclient/ZkClient.java
b/zookeeper-api/src/main/java/org/apache/helix/zookeeper/zkclient/ZkClient.java
index 50703a7e0..e7dbbe4bc 100644
---
a/zookeeper-api/src/main/java/org/apache/helix/zookeeper/zkclient/ZkClient.java
+++
b/zookeeper-api/src/main/java/org/apache/helix/zookeeper/zkclient/ZkClient.java
@@ -3403,49 +3403,61 @@ public class ZkClient implements Watcher {
void run() throws KeeperException, InterruptedException;
}
- // Add a persist listener on the path.
- // Throws UnsupportedOperationException if there is already a recursive
persist listener on the
- // path because it will overwrite that recursive persist listener.
- private void addPersistListener(String path, Object listener) {
- ManipulateListener addListeners = () -> {
- if (_zkPathRecursiveWatcherTrie.hasListenerOnPath(path)) {
- throw new UnsupportedOperationException(
- "Can not subscribe PersistListener when there is an recursive
listener on path: "
- + path);
- }
- if (listener instanceof IZkChildListener) {
- addChildListener(path, (IZkChildListener) listener);
- } else if (listener instanceof IZkDataListener) {
- addDataListener(path, (IZkDataListener) listener);
- }
- };
- executeWithInPersistListenerMutex(addListeners);
+ /*
+ * Adds a persist child or data listener on the path. The two overloads
exist on purpose: a
+ * listener may implement both IZkChildListener and IZkDataListener
(CallbackHandler does), so the
+ * listener type has to come from the subscribe call that was made, not from
instanceof checks.
+ * Throws UnsupportedOperationException if there is already a recursive
persist listener on the
+ * path because it would overwrite that recursive persist listener.
+ */
+ private void addPersistListener(String path, IZkChildListener listener) {
+ executeWithInPersistListenerMutex(() -> {
+ checkNoRecursivePersistListener(path);
+ addChildListener(path, listener);
+ });
}
+ private void addPersistListener(String path, IZkDataListener listener) {
+ executeWithInPersistListenerMutex(() -> {
+ checkNoRecursivePersistListener(path);
+ addDataListener(path, listener);
+ });
+ }
- // TODO: Consider create an empty interface and let the two listeners
interface extend that
- // interface for code clean.
- // This function removes persist child or data listener.
- private void removePersistListener(String path, Object listener) {
+ private void checkNoRecursivePersistListener(String path) {
+ if (_zkPathRecursiveWatcherTrie.hasListenerOnPath(path)) {
+ throw new UnsupportedOperationException(
+ "Can not subscribe PersistListener when there is an recursive
listener on path: " + path);
+ }
+ }
- ManipulateListener removeListeners = () -> {
- try {
- if (listener instanceof IZkChildListener) {
- removeChildListener(path, (IZkChildListener) listener);
- } else if (listener instanceof IZkDataListener) {
- removeDataListener(path, (IZkDataListener) listener);
- }
- if (!hasChildOrDataListeners(path)) {
- // This will also remove persist recursive watcher on ZK. However,
there should not be an
- // persist recursive watcher installed in the first place.
- getConnection().removeWatches(path, this, WatcherType.Any);
- }
- } catch (KeeperException.NoWatcherException e) {
- LOG.warn("Persist watcher is already removed");
- }
- };
+ // Removes a persist child or data listener, and the persist watch once no
listener is left.
+ private void removePersistListener(String path, IZkChildListener listener) {
+ executeWithInPersistListenerMutex(() -> {
+ removeChildListener(path, listener);
+ removePersistWatchIfUnused(path);
+ });
+ }
- executeWithInPersistListenerMutex(removeListeners);
+ private void removePersistListener(String path, IZkDataListener listener) {
+ executeWithInPersistListenerMutex(() -> {
+ removeDataListener(path, listener);
+ removePersistWatchIfUnused(path);
+ });
+ }
+
+ private void removePersistWatchIfUnused(String path)
+ throws KeeperException, InterruptedException {
+ if (hasChildOrDataListeners(path)) {
+ return;
+ }
+ try {
+ // This will also remove persist recursive watcher on ZK. However, there
should not be an
+ // persist recursive watcher installed in the first place.
+ getConnection().removeWatches(path, this, WatcherType.Any);
+ } catch (KeeperException.NoWatcherException e) {
+ LOG.warn("Persist watcher is already removed");
+ }
}
private void executeWithInPersistListenerMutex(ManipulateListener runnable) {
diff --git
a/zookeeper-api/src/test/java/org/apache/helix/zookeeper/impl/ZkTestHelper.java
b/zookeeper-api/src/test/java/org/apache/helix/zookeeper/impl/ZkTestHelper.java
index 5624c9eca..35622d55d 100644
---
a/zookeeper-api/src/test/java/org/apache/helix/zookeeper/impl/ZkTestHelper.java
+++
b/zookeeper-api/src/test/java/org/apache/helix/zookeeper/impl/ZkTestHelper.java
@@ -378,9 +378,21 @@ public class ZkTestHelper {
field2.setAccessible(true);
Map<String, Set<Watcher>> childWatches = (Map<String, Set<Watcher>>)
field2.get(watchManager);
+ field2 = getField(watchManager.getClass(), "persistentWatches");
+ field2.setAccessible(true);
+ Map<String, Set<Watcher>> persistentWatches =
+ (Map<String, Set<Watcher>>) field2.get(watchManager);
+
+ field2 = getField(watchManager.getClass(), "persistentRecursiveWatches");
+ field2.setAccessible(true);
+ Map<String, Set<Watcher>> persistentRecursiveWatches =
+ (Map<String, Set<Watcher>>) field2.get(watchManager);
+
lists.put("dataWatches", new ArrayList<>(dataWatches.keySet()));
lists.put("existWatches", new ArrayList<>(existWatches.keySet()));
lists.put("childWatches", new ArrayList<>(childWatches.keySet()));
+ lists.put("persistentWatches", new
ArrayList<>(persistentWatches.keySet()));
+ lists.put("persistentRecursiveWatches", new
ArrayList<>(persistentRecursiveWatches.keySet()));
return lists;
}
diff --git
a/zookeeper-api/src/test/java/org/apache/helix/zookeeper/impl/client/RealmAwareZkClientFactoryTestBase.java
b/zookeeper-api/src/test/java/org/apache/helix/zookeeper/impl/client/RealmAwareZkClientFactoryTestBase.java
index bae31de40..88267ae17 100644
---
a/zookeeper-api/src/test/java/org/apache/helix/zookeeper/impl/client/RealmAwareZkClientFactoryTestBase.java
+++
b/zookeeper-api/src/test/java/org/apache/helix/zookeeper/impl/client/RealmAwareZkClientFactoryTestBase.java
@@ -21,12 +21,17 @@ package org.apache.helix.zookeeper.impl.client;
import java.io.IOException;
import java.util.NoSuchElementException;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
import org.apache.helix.msdcommon.exception.InvalidRoutingDataException;
import org.apache.helix.zookeeper.api.client.RealmAwareZkClient;
import org.apache.helix.zookeeper.datamodel.ZNRecord;
import org.apache.helix.zookeeper.datamodel.serializer.ZNRecordSerializer;
+import org.apache.helix.zookeeper.impl.TestHelper;
+import org.apache.helix.zookeeper.zkclient.IZkDataListener;
import org.apache.helix.zookeeper.zkclient.exception.ZkBadVersionException;
+import org.apache.zookeeper.data.Stat;
import org.testng.Assert;
import org.testng.annotations.AfterClass;
import org.testng.annotations.BeforeClass;
@@ -110,6 +115,58 @@ public abstract class RealmAwareZkClientFactoryTestBase
extends RealmAwareZkClie
}
}
+ /**
+ * The persist watcher mode configured through RealmAwareZkClientConfig must
reach the ZkClient
+ * the factory builds: listeners are served by persistent watches and
explicit one-time watches
+ * are refused.
+ */
+ @Test
+ public void testUsePersistWatcherConfig() throws Exception {
+ RealmAwareZkClient.RealmAwareZkConnectionConfig connectionConfig =
+ new RealmAwareZkClient.RealmAwareZkConnectionConfig.Builder()
+ .setZkRealmShardingKey(ZK_SHARDING_KEY_PREFIX).build();
+ RealmAwareZkClient.RealmAwareZkClientConfig clientConfig =
+ new
RealmAwareZkClient.RealmAwareZkClientConfig().setUsePersistWatcher(true);
+ RealmAwareZkClient client =
+ _realmAwareZkClientFactory.buildZkClient(connectionConfig,
clientConfig);
+ String path = ZK_SHARDING_KEY_PREFIX + "/" +
TestHelper.getTestMethodName();
+ try {
+ client.setZkSerializer(new ZNRecordSerializer());
+ client.createPersistent(path, true);
+
+ // A one-time watch would take over the persistent watch on older
servers, so the client
+ // refuses it in this mode.
+ try {
+ client.readData(path, new Stat(), true);
+ Assert.fail("One-time watches must be refused in persist watcher
mode");
+ } catch (IllegalArgumentException expected) {
+ // expected
+ }
+
+ // Listeners are served by a persistent watch: every change is delivered
without re-arming.
+ int changes = 5;
+ CountDownLatch latch = new CountDownLatch(changes);
+ client.subscribeDataChanges(path, new IZkDataListener() {
+ @Override
+ public void handleDataChange(String dataPath, Object data) {
+ latch.countDown();
+ }
+
+ @Override
+ public void handleDataDeleted(String dataPath) {
+ }
+ });
+ for (int i = 0; i < changes; i++) {
+ client.writeData(path, new ZNRecord("v" + i));
+ }
+ Assert.assertTrue(latch.await(10, TimeUnit.SECONDS),
+ "Expected " + changes + " data change notifications, " +
latch.getCount() + " missing");
+ } finally {
+ client.deleteRecursively(path);
+ client.close();
+ }
+ }
+
/**
* Test creating a container node.
*/
diff --git
a/zookeeper-api/src/test/java/org/apache/helix/zookeeper/impl/client/TestFederatedZkClient.java
b/zookeeper-api/src/test/java/org/apache/helix/zookeeper/impl/client/TestFederatedZkClient.java
index 3fea6a8bf..881715816 100644
---
a/zookeeper-api/src/test/java/org/apache/helix/zookeeper/impl/client/TestFederatedZkClient.java
+++
b/zookeeper-api/src/test/java/org/apache/helix/zookeeper/impl/client/TestFederatedZkClient.java
@@ -26,6 +26,7 @@ import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.NoSuchElementException;
+import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import org.apache.helix.msdcommon.constant.MetadataStoreRoutingConstants;
@@ -39,10 +40,12 @@ import org.apache.helix.zookeeper.constant.TestConstants;
import org.apache.helix.zookeeper.datamodel.ZNRecord;
import org.apache.helix.zookeeper.datamodel.serializer.ZNRecordSerializer;
import org.apache.helix.zookeeper.routing.RoutingDataManager;
+import org.apache.helix.zookeeper.zkclient.IZkDataListener;
import org.apache.helix.zookeeper.zkclient.IZkStateListener;
import org.apache.helix.zookeeper.zkclient.exception.ZkBadVersionException;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.Watcher;
+import org.apache.zookeeper.data.Stat;
import org.testng.Assert;
import org.testng.annotations.AfterClass;
import org.testng.annotations.BeforeClass;
@@ -76,6 +79,48 @@ public class TestFederatedZkClient extends
RealmAwareZkClientTestBase {
System.out.println("Ending " +
TestFederatedZkClient.class.getSimpleName());
}
+ /*
+ * The persist watcher mode configured through RealmAwareZkClientConfig must
reach the per-realm
+ * ZkClients: listeners are served by persistent watches and explicit
one-time watches are refused.
+ */
+ @Test
+ public void testUsePersistWatcherConfig() throws Exception {
+ FederatedZkClient client =
+ new FederatedZkClient(new
RealmAwareZkClient.RealmAwareZkConnectionConfig.Builder().build(),
+ new
RealmAwareZkClient.RealmAwareZkClientConfig().setUsePersistWatcher(true));
+ String path = TEST_REALM_ONE_VALID_PATH + "/persistWatcher";
+ try {
+ client.setZkSerializer(new ZNRecordSerializer());
+ client.createPersistent(path, true);
+ try {
+ client.readData(path, new Stat(), true);
+ Assert.fail("One-time watches must be refused in persist watcher
mode");
+ } catch (IllegalArgumentException expected) {
+ // expected
+ }
+ int changes = 5;
+ CountDownLatch latch = new CountDownLatch(changes);
+ client.subscribeDataChanges(path, new IZkDataListener() {
+ @Override
+ public void handleDataChange(String dataPath, Object data) {
+ latch.countDown();
+ }
+
+ @Override
+ public void handleDataDeleted(String dataPath) {
+ }
+ });
+ for (int i = 0; i < changes; i++) {
+ client.writeData(path, new ZNRecord("v" + i));
+ }
+ Assert.assertTrue(latch.await(10, TimeUnit.SECONDS),
+ "Expected " + changes + " data change notifications, " +
latch.getCount() + " missing");
+ } finally {
+ client.deleteRecursively(TEST_REALM_ONE_VALID_PATH);
+ client.close();
+ }
+ }
+
/*
* Tests that an unsupported operation should throw an
UnsupportedOperationException.
*/
diff --git
a/zookeeper-api/src/test/java/org/apache/helix/zookeeper/zkclient/TestZkClientPersistWatcher.java
b/zookeeper-api/src/test/java/org/apache/helix/zookeeper/zkclient/TestZkClientPersistWatcher.java
index c54bca1ef..6b7420628 100644
---
a/zookeeper-api/src/test/java/org/apache/helix/zookeeper/zkclient/TestZkClientPersistWatcher.java
+++
b/zookeeper-api/src/test/java/org/apache/helix/zookeeper/zkclient/TestZkClientPersistWatcher.java
@@ -24,6 +24,7 @@ import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
+import org.apache.helix.zookeeper.impl.TestHelper;
import org.apache.helix.zookeeper.impl.ZkTestBase;
import org.apache.helix.zookeeper.impl.client.ZkClient;
import org.apache.helix.zookeeper.zkclient.serialize.BasicZkSerializer;
@@ -170,6 +171,71 @@ public class TestZkClientPersistWatcher extends ZkTestBase
{
zkClient.close();
}
+ /*
+ * A listener that implements both IZkChildListener and IZkDataListener (as
CallbackHandler does)
+ * must be registered for the kind of change it subscribed to. Registering
it by instanceof made
+ * every data subscription a child subscription in persist mode, so data
changes were never
+ * delivered.
+ */
+ @Test
+ void testListenerImplementingBothInterfaces() throws Exception {
+ org.apache.helix.zookeeper.impl.client.ZkClient zkClient =
+ new
org.apache.helix.zookeeper.impl.client.ZkClient.Builder().setZkServer(ZkTestBase.ZK_ADDR)
+ .setMonitorRootPathOnly(false).setUsePersistWatcher(true).build();
+ zkClient.setZkSerializer(new BasicZkSerializer(new
SerializableSerializer()));
+ String path = "/testListenerImplementingBothInterfaces";
+ zkClient.create(path, "data", CreateMode.PERSISTENT);
+ try {
+ class BothListener implements IZkChildListener, IZkDataListener {
+ final AtomicInteger _dataChanges = new AtomicInteger();
+ final AtomicInteger _childChanges = new AtomicInteger();
+
+ @Override
+ public void handleChildChange(String parentPath, List<String>
currentChilds) {
+ _childChanges.incrementAndGet();
+ }
+
+ @Override
+ public void handleDataChange(String dataPath, Object data) {
+ _dataChanges.incrementAndGet();
+ }
+
+ @Override
+ public void handleDataDeleted(String dataPath) {
+ }
+ }
+ BothListener dataOnly = new BothListener();
+ BothListener childOnly = new BothListener();
+ zkClient.subscribeDataChanges(path, dataOnly);
+ zkClient.subscribeChildChanges(path, childOnly);
+
+ zkClient.writeData(path, "data1", -1);
+ zkClient.create(path + "/child", "c", CreateMode.PERSISTENT);
+
+ Assert.assertTrue(TestHelper.verify(() -> dataOnly._dataChanges.get() ==
1, 10_000L),
+ "The data listener should get the data change");
+ Assert.assertTrue(TestHelper.verify(() -> childOnly._childChanges.get()
== 1, 10_000L),
+ "The child listener should get the child change");
+ Assert.assertEquals(dataOnly._childChanges.get(), 0,
+ "The data listener must not be registered as a child listener");
+ Assert.assertEquals(childOnly._dataChanges.get(), 0,
+ "The child listener must not be registered as a data listener");
+
+ // Unsubscribing one listener kind keeps the other one, and its watch,
working.
+ zkClient.unsubscribeDataChanges(path, dataOnly);
+ zkClient.writeData(path, "data2", -1);
+ zkClient.create(path + "/child2", "c", CreateMode.PERSISTENT);
+ Assert.assertTrue(TestHelper.verify(() -> childOnly._childChanges.get()
== 2, 10_000L),
+ "The child listener should still get child changes");
+ Assert.assertEquals(dataOnly._dataChanges.get(), 1,
+ "The unsubscribed data listener must not get further data changes");
+ zkClient.unsubscribeChildChanges(path, childOnly);
+ } finally {
+ zkClient.deleteRecursively(path);
+ zkClient.close();
+ }
+ }
+
@Test
void testSubscribeOneTimeChangeWhenUsingPersistWatcher() {
org.apache.helix.zookeeper.impl.client.ZkClient.Builder builder =
