diff --git a/helix-common/src/main/java/org/apache/helix/SystemPropertyKeys.java b/helix-common/src/main/java/org/apache/helix/SystemPropertyKeys.java index 1f00a73d9d..9b1152a2d1 100644 --- a/helix-common/src/main/java/org/apache/helix/SystemPropertyKeys.java +++ b/helix-common/src/main/java/org/apache/helix/SystemPropertyKeys.java @@ -97,4 +97,15 @@ public class SystemPropertyKeys { // Stage thread pool size for parallel stage execution public static final String STAGE_THREAD_POOL_SIZE = "helix.stage.threadpool.size"; + + // When enabled, the controller subscribes a single PERSISTENT_RECURSIVE ZooKeeper watch per + // participant current-state subtree (CURRENTSTATES and TASKCURRENTSTATES) instead of one child watch + // + one data watch per partition. This collapses the per-handoff watch footprint and the cold-start + // subscribe cost from O(N*M) to O(1) re-arm. Requires the controller's ZkClient to run with + // usePersistWatcher=true (wired automatically when this flag is set). Off by default for backward + // compatibility. The recursive watch is intentionally scoped to current-state/task-current-state; + // all other change types (e.g. CUSTOMIZEDSTATES) keep per-node watches, which operate correctly in + // persist mode and continue to drive customized-view aggregation. + public static final String PARTICIPANT_STATE_PERSIST_RECURSIVE_WATCH_ENABLED = + "helix.controller.participantState.persistRecursiveWatch.enabled"; } diff --git a/helix-core/src/main/java/org/apache/helix/manager/zk/CallbackHandler.java b/helix-core/src/main/java/org/apache/helix/manager/zk/CallbackHandler.java index c2809620e3..16b0077846 100644 --- a/helix-core/src/main/java/org/apache/helix/manager/zk/CallbackHandler.java +++ b/helix-core/src/main/java/org/apache/helix/manager/zk/CallbackHandler.java @@ -36,6 +36,7 @@ import org.apache.helix.HelixDataAccessor; import org.apache.helix.HelixException; import org.apache.helix.HelixManager; +import org.apache.helix.InstanceType; import org.apache.helix.HelixProperty; import org.apache.helix.NotificationContext; import org.apache.helix.NotificationContext.Type; @@ -79,6 +80,7 @@ import org.apache.helix.zookeeper.datamodel.ZNRecord; import org.apache.helix.zookeeper.zkclient.IZkChildListener; import org.apache.helix.zookeeper.zkclient.IZkDataListener; +import org.apache.helix.zookeeper.zkclient.RecursivePersistListener; import org.apache.helix.zookeeper.zkclient.annotation.PreFetchChangedData; import org.apache.helix.zookeeper.zkclient.exception.ZkNoNodeException; import org.apache.zookeeper.Watcher.Event.EventType; @@ -105,7 +107,7 @@ import static org.apache.helix.HelixConstants.ChangeType.TASK_CURRENT_STATE; @PreFetchChangedData(enabled = false) -public class CallbackHandler implements IZkChildListener, IZkDataListener { +public class CallbackHandler implements IZkChildListener, IZkDataListener, RecursivePersistListener { private static Logger logger = LoggerFactory.getLogger(CallbackHandler.class); private static final AtomicLong CALLBACK_HANDLER_UID = new AtomicLong(); @@ -138,8 +140,23 @@ public class CallbackHandler implements IZkChildListener, IZkDataListener { private boolean _watchChild = true; // Whether we should subscribe to the child znode's data // change. + // When true (controller CURRENT_STATE/TASK_CURRENT_STATE handler + flag enabled + + // persist-watcher client), this handler installs ONE PERSISTENT_RECURSIVE watch covering its whole + // subtree instead of one child watch plus one data watch per partition. + private final boolean _useRecursivePersistWatch; + // Whether the single recursive watch is currently installed on the server. A PERSISTENT_RECURSIVE + // watch is installed once per session and never re-armed, so it must be (re)installed only on INIT + // and removed exactly once on reset. Guards against re-subscribing on every callback / on FINALIZE, + // and makes reset's unsubscribe accurate. Volatile: read/written from the ZkEventThread and resets. + private volatile boolean _recursiveWatchInstalled = false; + // Set if the underlying zk client does not support persistent recursive watches (should not happen + // when the flag is on, but kept defensive): once set, the handler permanently uses per-node watches. + private volatile boolean _recursiveWatchUnsupported = false; + // indicated whether this CallbackHandler is ready to serve event callback from ZkClient. - private boolean _ready = false; + // Volatile: written under the manager monitor (init/reset) and read on the ZkEventThread + // (handleZNodeChange / handleChildChange / handleDataChange / enqueueTask). + private volatile boolean _ready = false; /** * maintain the expected notification types @@ -186,6 +203,16 @@ public CallbackHandler(HelixManager manager, RealmAwareZkClient client, Property parseListenerProperties(); + _useRecursivePersistWatch = + Boolean.getBoolean(SystemPropertyKeys.PARTICIPANT_STATE_PERSIST_RECURSIVE_WATCH_ENABLED) + && (_changeType == CURRENT_STATE || _changeType == TASK_CURRENT_STATE) + // Only the controller runs its ZkClient in persist-watcher mode (see ZKHelixManager). + // Gate on instance type too so a spectator/participant current-state handler sharing the + // JVM does not attempt a recursive subscribe on a non-persist client (which would throw, + // log a warning, and fall back to per-node watches). + && (_manager.getInstanceType() == InstanceType.CONTROLLER + || _manager.getInstanceType() == InstanceType.CONTROLLER_PARTICIPANT); + init(); } @@ -563,6 +590,31 @@ private void subscribeForChanges(NotificationContext.Type callbackType, String p _uid, path, callbackType, _eventTypes, _listener, watchChild); long start = System.currentTimeMillis(); + + if (_useRecursivePersistWatch && !_recursiveWatchUnsupported) { + // A PERSISTENT_RECURSIVE watch covers all data + child changes under the entire subtree and is + // NEVER re-armed. Install it once, on INIT only: CALLBACK re-entries (handleZNodeChange sets + // isChildChange=true) and FINALIZE (reset) must NOT re-subscribe, otherwise (a) we would issue a + // redundant addWatch on every event and (b) reset() -> invoke(FINALIZE) would re-install the + // watch right after unsubscribing it, leaking it. For all these change types we then skip the + // per-node child/data subscribe loop below. + if (callbackType == Type.INIT && !_recursiveWatchInstalled) { + try { + _zkClient.subscribePersistRecursiveListener(path, this); + _recursiveWatchInstalled = true; + logger.info("CallbackHandler {} installed ONE persistent recursive watch on path: {} " + + "(replaces per-child data watches for change type {})", _uid, path, _changeType); + } catch (UnsupportedOperationException e) { + logger.warn("CallbackHandler {} persist recursive watch unsupported by zk client; falling " + + "back to per-node watches on path: {}", _uid, path); + _recursiveWatchUnsupported = true; + } + } + if (!_recursiveWatchUnsupported) { + return; + } + // else: client does not support it -> fall through to the per-node subscribe below. + } if (_eventTypes.contains(EventType.NodeDataChanged) || _eventTypes.contains(EventType.NodeCreated) || _eventTypes.contains(EventType.NodeDeleted)) { @@ -752,6 +804,37 @@ public void handleChildChange(String parentPath, List currentChilds) { } } + @Override + public void handleZNodeChange(String dataPath, EventType eventType) { + // A single PERSISTENT_RECURSIVE watch fired for a change to some node anywhere under _path. + // Route it to the same CALLBACK path that the per-node child/data watches would have used so the + // controller pipeline reacts identically (it re-reads current state regardless of which node). + try { + updateNotificationTime(System.nanoTime()); + // Match the watched node itself or a node strictly under it. Respect the '/' segment boundary + // so a sibling whose name merely extends _path (e.g. .../session1 vs .../session11) cannot be + // accepted. (The recursive-watcher trie already scopes delivery to the exact subtree; this + // guard is defense-in-depth in case dispatch ever changes.) + if (dataPath != null && (dataPath.equals(_path) || dataPath.startsWith(_path + "/"))) { + if (!isReady()) { + logger.info("CallbackHandler {} is in reset state; skip recursive {} event on path: {}", + _uid, eventType, dataPath); + return; + } + NotificationContext changeContext = new NotificationContext(_manager); + changeContext.setType(NotificationContext.Type.CALLBACK); + changeContext.setPathChanged(dataPath); + changeContext.setChangeType(_changeType); + changeContext.setIsChildChange(true); + enqueueTask(changeContext); + } + } catch (Exception e) { + String msg = "exception in handling recursive znode-change. path: " + dataPath + ", listener: " + + _listener; + ZKExceptionHandler.getInstance().handle(msg, e); + } + } + /** * Invoke the listener for the last time so that the listener could clean up resources */ @@ -764,6 +847,19 @@ public void reset(boolean isShutdown) { logger.info("Resetting CallbackHandler: {}. Is resetting for shutdown: {}.", _uid, isShutdown); try { _ready = false; + if (_recursiveWatchInstalled) { + // The recursive watch is persistent, so it must be explicitly removed (one call) on reset / + // session change, otherwise it would leak. Only attempt removal if we actually installed it; + // init() re-installs the single watch on the next INIT. Cleared even if removal throws (e.g. + // session already expired) so a subsequent INIT re-installs cleanly. + try { + _zkClient.unsubscribePersistRecursiveListener(_path, this); + } catch (Exception e) { + logger.warn("CallbackHandler {} failed to unsubscribe recursive watch on path: {}, {}", + _uid, _path, e.toString()); + } + _recursiveWatchInstalled = false; + } CallbackEventExecutor callbackExecutor = _batchCallbackExecutorRef.get(); if (callbackExecutor != null) { if (isShutdown) { 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 2044dbcd79..25263600ba 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,12 @@ private void createLiveInstance() { _clusterName); Stat stat = new Stat(); - ZNRecord record = _zkclient.readData(liveInstancePath, stat, true); + // watch=false: this read only checks whether a stale live-instance node still exists; no + // listener consumes a watch here. A one-shot (watch=true) read would throw under a + // persist-watcher client (validateNativeZkWatcherType), which a CONTROLLER_PARTICIPANT uses + // when the persist-recursive watch flag is on -> it would abort new-session handling on the + // duplicate-live-instance race (fast controller restart). + ZNRecord record = _zkclient.readData(liveInstancePath, stat, false); 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 32d6eec621..f3f52cf7d5 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 @@ -1522,6 +1522,14 @@ private RealmAwareZkClient createSingleRealmZkClient() { .setMonitorInstanceName(_instanceName) .setMonitorRootPathOnly(isMonitorRootPathOnly()); + // When the persist-recursive participant-state watch is enabled, the controller's client must run + // in persistent-watcher mode so CallbackHandler can install one recursive watch per subtree. + if (Boolean.getBoolean(SystemPropertyKeys.PARTICIPANT_STATE_PERSIST_RECURSIVE_WATCH_ENABLED) + && (_instanceType == InstanceType.CONTROLLER + || _instanceType == InstanceType.CONTROLLER_PARTICIPANT)) { + clientConfig.setUsePersistWatcher(true); + } + if (_instanceType == InstanceType.ADMINISTRATOR) { return resolveZkClient(SharedZkClientFactory.getInstance(), _realmAwareZkConnectionConfig, clientConfig); diff --git a/helix-core/src/test/java/org/apache/helix/integration/TestCurrentStatePersistRecursiveWatch.java b/helix-core/src/test/java/org/apache/helix/integration/TestCurrentStatePersistRecursiveWatch.java new file mode 100644 index 0000000000..4398ae1d15 --- /dev/null +++ b/helix-core/src/test/java/org/apache/helix/integration/TestCurrentStatePersistRecursiveWatch.java @@ -0,0 +1,324 @@ +package org.apache.helix.integration; + +/* + * 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.lang.reflect.Field; +import java.util.Date; + +import org.apache.helix.SystemPropertyKeys; +import org.apache.helix.TestHelper; +import org.apache.helix.ZkTestHelper; +import org.apache.helix.ZkUnitTestBase; +import org.apache.helix.integration.manager.ClusterControllerManager; +import org.apache.helix.integration.manager.MockParticipantManager; +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.zkclient.util.ZkPathRecursiveWatcherTrie; +import org.testng.Assert; +import org.testng.annotations.AfterClass; +import org.testng.annotations.BeforeClass; +import org.testng.annotations.Test; + +/** + * End-to-end test for the persist-recursive current-state watch + * ({@link SystemPropertyKeys#PARTICIPANT_STATE_PERSIST_RECURSIVE_WATCH_ENABLED}). + * + * With the flag enabled, the controller's ZkClient is built with usePersistWatcher=true and each + * CURRENT_STATE / TASK_CURRENT_STATE CallbackHandler installs ONE PERSISTENT_RECURSIVE watch on the + * participant's CURRENTSTATES subtree instead of one child watch plus one data watch per partition. + * + * The controller can only compute a correct ExternalView if it actually receives the participants' + * current-state changes. So a green {@link BestPossibleExternalViewVerifier} both at steady state and + * after an ongoing current-state change (a participant failure that forces masters to move) proves the + * single recursive watch delivers initial AND incremental current-state events correctly. + */ +public class TestCurrentStatePersistRecursiveWatch extends ZkUnitTestBase { + + @BeforeClass + public void beforeClass() { + System.setProperty(SystemPropertyKeys.PARTICIPANT_STATE_PERSIST_RECURSIVE_WATCH_ENABLED, "true"); + } + + @AfterClass + public void afterClass() { + System.clearProperty(SystemPropertyKeys.PARTICIPANT_STATE_PERSIST_RECURSIVE_WATCH_ENABLED); + } + + @Test + public void testControllerConvergesWithRecursiveCurrentStateWatch() throws Exception { + String className = TestHelper.getTestClassName(); + String methodName = TestHelper.getTestMethodName(); + String clusterName = className + "_" + methodName; + final int n = 3; + + System.out.println("START " + clusterName + " at " + new Date(System.currentTimeMillis())); + + TestHelper.setupCluster(clusterName, ZK_ADDR, 12918, // participant port + "localhost", // participant name prefix + "TestDB", // resource name prefix + 1, // resources + 8, // partitions per resource + n, // number of nodes + 3, // replicas + "MasterSlave", true); // do rebalance + + MockParticipantManager[] participants = new MockParticipantManager[n]; + for (int i = 0; i < n; i++) { + String instanceName = "localhost_" + (12918 + i); + participants[i] = new MockParticipantManager(ZK_ADDR, clusterName, instanceName); + participants[i].syncStart(); + } + + // The controller's ZkClient is built with usePersistWatcher=true because the flag is set, so its + // CURRENT_STATE CallbackHandlers use the single recursive watch. + ClusterControllerManager controller = + new ClusterControllerManager(ZK_ADDR, clusterName, "controller_0"); + controller.syncStart(); + + ZkHelixClusterVerifier verifier = + new BestPossibleExternalViewVerifier.Builder(clusterName).setZkClient(_gZkClient) + .setWaitTillVerify(TestHelper.DEFAULT_REBALANCE_PROCESSING_WAIT_TIME).build(); + + // Steady state: controller must have observed every participant's current state through the + // recursive watch to make ExternalView == BestPossible. + Assert.assertTrue(verifier.verifyByPolling(), + "Cluster did not converge at steady state with the recursive current-state watch"); + + // Ongoing current-state changes: drop one participant; the masters it hosted must move, which the + // controller can only carry out by observing OFFLINE->SLAVE->MASTER current-state transitions on + // the surviving participants (delivered by the recursive watch). Re-convergence proves incremental + // current-state events are delivered. + participants[0].syncStop(); + Assert.assertTrue(verifier.verifyByPolling(), + "Cluster did not re-converge after a participant failure; recursive current-state watch did " + + "not deliver incremental events"); + + // Cleanup. + controller.syncStop(); + for (int i = 0; i < n; i++) { + if (participants[i].isConnected()) { + participants[i].syncStop(); + } + } + deleteCluster(clusterName); + System.out.println("END " + clusterName + " at " + new Date(System.currentTimeMillis())); + } + + /** + * Regression test for the watch lifecycle: with the flag on, the controller must (a) use a small, + * O(participants) number of CURRENTSTATES watches (one recursive watch per participant subtree, not + * one per partition), and (b) REMOVE the recursive watch when a participant departs (reset()), with + * no leak. The latter specifically guards against re-subscribing the persistent watch on every + * callback / re-installing it on FINALIZE during reset. + */ + @Test + public void testRecursiveWatchRemovedOnParticipantDeparture() throws Exception { + String className = TestHelper.getTestClassName(); + String methodName = TestHelper.getTestMethodName(); + String clusterName = className + "_" + methodName; + final int n = 3; + + System.out.println("START " + clusterName + " at " + new Date(System.currentTimeMillis())); + + TestHelper.setupCluster(clusterName, ZK_ADDR, 12918, "localhost", "TestDB", + 1, // resources + 8, // partitions per resource + n, // nodes + 3, // replicas + "MasterSlave", true); + + MockParticipantManager[] participants = new MockParticipantManager[n]; + for (int i = 0; i < n; i++) { + participants[i] = new MockParticipantManager(ZK_ADDR, clusterName, "localhost_" + (12918 + i)); + participants[i].syncStart(); + } + ClusterControllerManager controller = + new ClusterControllerManager(ZK_ADDR, clusterName, "controller_0"); + controller.syncStart(); + + ZkHelixClusterVerifier verifier = + new BestPossibleExternalViewVerifier.Builder(clusterName).setZkClient(_gZkClient) + .setWaitTillVerify(TestHelper.DEFAULT_REBALANCE_PROCESSING_WAIT_TIME).build(); + Assert.assertTrue(verifier.verifyByPolling(), "Cluster did not converge"); + + // The CURRENTSTATES subtree of the participant whose session we will expire. Use the OLD session id + // because expireSession reconnects the participant under a NEW session (the controller will watch + // the new session's subtree; only the OLD one must be torn down). + String oldSession = participants[0].getSessionId(); + String oldCsPath = + "/" + clusterName + "/INSTANCES/localhost_12918/CURRENTSTATES/" + oldSession; + + // Precondition: the controller's recursive-watch trie holds a listener for the old session's + // CURRENTSTATES subtree. (A wchp server-side dump cannot prove the leak: after expiry the old + // CURRENTSTATES node is deleted, and a persistent watch re-armed on a deleted path is not reported. + // The client-side trie is the definitive location where the leaked handler is retained.) + Assert.assertTrue(countControllerRecursiveListeners(controller, oldCsPath) >= 1, + "precondition: controller should hold a recursive watch on the participant's CURRENTSTATES " + + "before expiry"); + + // Expire the participant session: the old session leaves LIVEINSTANCES -> controller removeListener + // -> reset() must remove the persistent recursive watch on the OLD session's CURRENTSTATES subtree. + // The buggy version re-installed the watch via invoke(FINALIZE) right after reset unsubscribed it, + // permanently leaking one CallbackHandler (retained by the recursive trie) + one server-side watch + // on the dead path. This fails on that bug and passes once reset tears the watch down. + ZkTestHelper.expireSession(participants[0].getZkClient()); + + boolean removed = TestHelper.verify( + () -> countControllerRecursiveListeners(controller, oldCsPath) == 0, + TestHelper.WAIT_DURATION); + Assert.assertTrue(removed, + "Controller leaked a recursive watch on the expired session's CURRENTSTATES subtree (" + + oldCsPath + ") -- the CallbackHandler was re-subscribed on FINALIZE and orphaned in the " + + "recursive-watch trie"); + + controller.syncStop(); + for (int i = 0; i < n; i++) { + if (participants[i].isConnected()) { + participants[i].syncStop(); + } + } + deleteCluster(clusterName); + System.out.println("END " + clusterName + " at " + new Date(System.currentTimeMillis())); + } + + /** + * Regression test for the CONTROLLER's own ZK-session loss (the production failover / ZK-blip case): + * on a new session the controller must TEAR DOWN the recursive watches from the expired session and + * RE-INSTALL them on the new session, and resume delivering current-state changes. This exercises + * the reset()-then-init() re-arm path on the controller side (vs. the participant-departure test). + */ + @Test + public void testRecursiveWatchReArmedOnControllerSessionExpiry() throws Exception { + String className = TestHelper.getTestClassName(); + String methodName = TestHelper.getTestMethodName(); + String clusterName = className + "_" + methodName; + final int n = 3; + + System.out.println("START " + clusterName + " at " + new Date(System.currentTimeMillis())); + + TestHelper.setupCluster(clusterName, ZK_ADDR, 12918, "localhost", "TestDB", + 1, // resources + 8, // partitions per resource + n, // nodes + 3, // replicas + "MasterSlave", true); + + MockParticipantManager[] participants = new MockParticipantManager[n]; + for (int i = 0; i < n; i++) { + participants[i] = new MockParticipantManager(ZK_ADDR, clusterName, "localhost_" + (12918 + i)); + participants[i].syncStart(); + } + ClusterControllerManager controller = + new ClusterControllerManager(ZK_ADDR, clusterName, "controller_0"); + controller.syncStart(); + + ZkHelixClusterVerifier verifier = + new BestPossibleExternalViewVerifier.Builder(clusterName).setZkClient(_gZkClient) + .setWaitTillVerify(TestHelper.DEFAULT_REBALANCE_PROCESSING_WAIT_TIME).build(); + Assert.assertTrue(verifier.verifyByPolling(), "Cluster did not converge"); + + // A participant's CURRENTSTATES subtree the controller watches. The participant session is NOT + // expired here (only the controller's session is), so this path is stable across the controller's + // new session -- the controller must re-install its recursive watch on this same path. + String participantSession = participants[0].getSessionId(); + String csPath = + "/" + clusterName + "/INSTANCES/localhost_12918/CURRENTSTATES/" + participantSession; + Assert.assertTrue(countControllerRecursiveListeners(controller, csPath) >= 1, + "precondition: controller should hold a recursive watch on the participant's CURRENTSTATES"); + + // Expire the CONTROLLER's own ZK session: handleNewSession -> resetHandlers (remove old-session + // recursive watches) -> initHandlers (re-install on the new session). The watch must come back. + String oldControllerSession = controller.getSessionId(); + ZkTestHelper.expireSession(controller.getZkClient()); + + // Confirm a genuinely NEW controller session (so this exercises real server-side watch loss, not a + // stale client-side trie entry): a persistent-recursive watch is dropped by the server on expiry. + boolean newSession = TestHelper.verify( + () -> controller.isConnected() && !oldControllerSession.equals(controller.getSessionId()), + TestHelper.WAIT_DURATION); + Assert.assertTrue(newSession, "controller did not establish a new session after expiry"); + + boolean reArmed = TestHelper.verify( + () -> countControllerRecursiveListeners(controller, csPath) >= 1, TestHelper.WAIT_DURATION); + Assert.assertTrue(reArmed, + "Controller did not re-install its recursive current-state watch after its own session " + + "expiry (path " + csPath + "); reset()/init() re-arm on the new session is broken"); + + // Functional proof the re-armed watch actually DELIVERS on the new session (the test runs with no + // periodic rebalance, so the only way the controller observes the post-failure current-state + // transitions is the recursive watch firing). A participant failure forces master movement, which + // the controller can only carry out by observing OFFLINE->SLAVE->MASTER transitions. Re-convergence + // proves incremental current-state events are delivered after the controller's session expiry. + participants[1].syncStop(); + Assert.assertTrue(verifier.verifyByPolling(), + "Cluster did not re-converge after controller session expiry + participant failure; the " + + "re-armed recursive watch did not deliver incremental current-state events"); + + controller.syncStop(); + for (int i = 0; i < n; i++) { + if (participants[i].isConnected()) { + participants[i].syncStop(); + } + } + deleteCluster(clusterName); + System.out.println("END " + clusterName + " at " + new Date(System.currentTimeMillis())); + } + + /** + * Number of recursive-watch listeners the controller's client-side trie holds for {@code path}. + * Reaches the raw {@code ZkClient} that owns the {@link ZkPathRecursiveWatcherTrie} via reflection + * (the same reflection-into-zkclient-internals pattern used by {@code ZkTestHelper#getZkWatch}). + */ + private int countControllerRecursiveListeners(ClusterControllerManager controller, String path) + throws Exception { + RealmAwareZkClient zkClient = controller.getZkClient(); + // In single-realm mode getZkClient() is the raw ZkClient (the trie is on its superclass); in + // realm-aware mode it is a DedicatedZkClient that wraps the raw client in _rawZkClient. Handle both. + Object trieOwner = zkClient; + try { + getFieldValue(zkClient, "_zkPathRecursiveWatcherTrie"); + } catch (NoSuchFieldException notRaw) { + trieOwner = getFieldValue(zkClient, "_rawZkClient"); + } + Object usePersist = getFieldValue(trieOwner, "_usePersistWatcher"); + if (!Boolean.TRUE.equals(usePersist)) { + throw new IllegalStateException("controller ZkClient was not built with usePersistWatcher=true; " + + "the persist-recursive watch is not active (usePersistWatcher=" + usePersist + ")"); + } + ZkPathRecursiveWatcherTrie trie = + (ZkPathRecursiveWatcherTrie) getFieldValue(trieOwner, "_zkPathRecursiveWatcherTrie"); + return trie.getAllRecursiveListeners(path).size(); + } + + private static Object getFieldValue(Object target, String fieldName) throws Exception { + Class clazz = target.getClass(); + while (clazz != null) { + try { + Field field = clazz.getDeclaredField(fieldName); + field.setAccessible(true); + return field.get(target); + } catch (NoSuchFieldException e) { + clazz = clazz.getSuperclass(); + } + } + throw new NoSuchFieldException(fieldName + " not found on " + target.getClass()); + } +} 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 a491f0de92..7e7a393532 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 @@ -164,5 +164,11 @@ public ZkClientConfig setConnectInitTimeout(long connectInitTimeout) { 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 751cc98af9..868552dbc7 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 @@ -32,6 +32,7 @@ import org.apache.helix.zookeeper.zkclient.IZkChildListener; import org.apache.helix.zookeeper.zkclient.IZkDataListener; import org.apache.helix.zookeeper.zkclient.IZkStateListener; +import org.apache.helix.zookeeper.zkclient.RecursivePersistListener; import org.apache.helix.zookeeper.zkclient.callback.ZkAsyncCallbacks; import org.apache.helix.zookeeper.zkclient.exception.ZkTimeoutException; import org.apache.helix.zookeeper.zkclient.serialize.BasicZkSerializer; @@ -130,6 +131,32 @@ boolean subscribeDataChanges(String path, IZkDataListener listener, void unsubscribeAll(); + /** + * Subscribe a single PERSISTENT_RECURSIVE watch (ZooKeeper 3.6+) that covers all data and child + * changes under the entire subtree rooted at {@code path}. Unlike the per-node child/data watches, + * this installs ONE server-side watch for the whole subtree and does not need to be re-armed after + * each event. The owning client must be built with {@code usePersistWatcher=true}. + * + * Default implementation throws {@link UnsupportedOperationException}; only clients that support a + * persistent watcher (e.g. the dedicated single-realm client) override it. + * + * @param path the subtree root to watch + * @param listener invoked for every add/remove/data change anywhere under {@code path} + */ + default void subscribePersistRecursiveListener(String path, RecursivePersistListener listener) { + throw new UnsupportedOperationException( + "subscribePersistRecursiveListener is not supported by this RealmAwareZkClient implementation"); + } + + /** + * Remove a recursive persistent watch previously installed via + * {@link #subscribePersistRecursiveListener(String, RecursivePersistListener)}. + */ + default void unsubscribePersistRecursiveListener(String path, RecursivePersistListener listener) { + throw new UnsupportedOperationException( + "unsubscribePersistRecursiveListener is not supported by this RealmAwareZkClient implementation"); + } + // data access void createPersistent(String path); @@ -472,6 +499,9 @@ class RealmAwareZkClientConfig { protected String _monitorKey; protected String _monitorInstanceName = null; protected boolean _monitorRootPathOnly = true; + // When true, the client registers PERSISTENT / PERSISTENT_RECURSIVE watches instead of one-shot + // watches, enabling subscribePersistRecursiveListener. Off by default for backward compatibility. + protected boolean _usePersistWatcher = false; public RealmAwareZkClientConfig setZkSerializer(PathBasedZkSerializer zkSerializer) { this._zkSerializer = zkSerializer; @@ -518,6 +548,15 @@ public RealmAwareZkClientConfig setMonitorRootPathOnly(Boolean monitorRootPathOn return this; } + public RealmAwareZkClientConfig setUsePersistWatcher(boolean usePersistWatcher) { + this._usePersistWatcher = usePersistWatcher; + return this; + } + + public boolean isUsePersistWatcher() { + return _usePersistWatcher; + } + public RealmAwareZkClientConfig setOperationRetryTimeout(Long operationRetryTimeout) { this._operationRetryTimeout = operationRetryTimeout; return this; @@ -568,7 +607,8 @@ public HelixZkClient.ZkClientConfig createHelixZkClientConfig() { .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/impl/client/DedicatedZkClient.java b/zookeeper-api/src/main/java/org/apache/helix/zookeeper/impl/client/DedicatedZkClient.java index 26e3efb1ee..a4d243ca2c 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 @@ -38,6 +38,7 @@ import org.apache.helix.zookeeper.zkclient.ZkConnection; import org.apache.helix.zookeeper.zkclient.callback.ZkAsyncCallbacks; import org.apache.helix.zookeeper.zkclient.IZkStateListener; +import org.apache.helix.zookeeper.zkclient.RecursivePersistListener; import org.apache.helix.zookeeper.zkclient.serialize.PathBasedZkSerializer; import org.apache.helix.zookeeper.zkclient.serialize.ZkSerializer; import org.apache.zookeeper.CreateMode; @@ -107,7 +108,8 @@ public DedicatedZkClient(RealmAwareZkClient.RealmAwareZkConnectionConfig connect _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 @@ -161,6 +163,18 @@ public void unsubscribeAll() { _rawZkClient.unsubscribeAll(); } + @Override + public void subscribePersistRecursiveListener(String path, RecursivePersistListener listener) { + checkIfPathContainsShardingKey(path); + _rawZkClient.subscribePersistRecursiveListener(path, listener); + } + + @Override + public void unsubscribePersistRecursiveListener(String path, RecursivePersistListener listener) { + checkIfPathContainsShardingKey(path); + _rawZkClient.unsubscribePersistRecursiveListener(path, listener); + } + @Override public void createPersistent(String path) { createPersistent(path, false); 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 bbccd22e9e..4d792de2f4 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 HelixZkClient buildZkClient(HelixZkClient.ZkConnectionConfig connectionCo 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/zkclient/ZkClient.java b/zookeeper-api/src/main/java/org/apache/helix/zookeeper/zkclient/ZkClient.java index 597a369a00..e022774591 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 @@ -3103,58 +3103,79 @@ interface ManipulateListener { // 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); + // A persist listener may implement BOTH IZkChildListener and IZkDataListener (Helix's + // CallbackHandler does). The caller therefore selects the target listener map explicitly through + // the overload it invokes; the kind MUST NOT be inferred via `instanceof`, otherwise a data + // subscription on such a dual-interface listener would be misrouted into the child-listener map + // and its NodeDataChanged events would be silently dropped. + private void addPersistListener(String path, IZkChildListener listener) { + executeWithInPersistListenerMutex(() -> { + checkNoRecursiveListenerOnPath(path); + addChildListener(path, listener); + }); } + private void addPersistListener(String path, IZkDataListener listener) { + executeWithInPersistListenerMutex(() -> { + checkNoRecursiveListenerOnPath(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 checkNoRecursiveListenerOnPath(String path) { + if (_zkPathRecursiveWatcherTrie.hasListenerOnPath(path)) { + throw new UnsupportedOperationException( + "Can not subscribe PersistListener when there is an recursive listener on path: " + path); + } + } - ManipulateListener removeListeners = () -> { + // This function removes a persist child listener. See addPersistListener for why the listener + // kind is selected by overload rather than inferred from the runtime type. + private void removePersistListener(String path, IZkChildListener listener) { + executeWithInPersistListenerMutex(() -> { + removeChildListener(path, listener); + removePersistWatchIfNoListeners(path); + }); + } + + // This function removes a persist data listener. + private void removePersistListener(String path, IZkDataListener listener) { + executeWithInPersistListenerMutex(() -> { + removeDataListener(path, listener); + removePersistWatchIfNoListeners(path); + }); + } + + private void removePersistWatchIfNoListeners(String path) + throws KeeperException, InterruptedException { + if (!hasChildOrDataListeners(path)) { + // This will also remove persist recursive watcher on ZK. However, there should not be a + // persist recursive watcher installed in the first place. 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); - } + getConnection().removeWatches(path, this, WatcherType.Any); } catch (KeeperException.NoWatcherException e) { LOG.warn("Persist watcher is already removed"); } - }; - - executeWithInPersistListenerMutex(removeListeners); + } } private void executeWithInPersistListenerMutex(ManipulateListener runnable) { + boolean locked = false; try { _persistListenerMutex.lockInterruptibly(); + locked = true; runnable.run(); } catch (KeeperException.NoWatcherException e) { LOG.warn("Persist watcher is already removed"); } catch (KeeperException | InterruptedException ex) { throw new ZkException(ex); } finally { - _persistListenerMutex.unlock(); + // Only unlock if we actually acquired the lock: an interrupted lockInterruptibly() does NOT + // hold it, and unlocking an unowned ReentrantLock would throw IllegalMonitorStateException, + // masking the original exception and swallowing the interrupt. + if (locked) { + _persistListenerMutex.unlock(); + } } } 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 c54bca1ef6..58b47ce9f5 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 @@ -70,6 +70,67 @@ public void handleDataDeleted(String dataPath) throws Exception { zkClient.close(); } + /* + * Regression test for the persist-watcher listener-routing bug. A listener that implements BOTH + * IZkChildListener and IZkDataListener (as Helix's CallbackHandler does) must still receive + * NodeDataChanged callbacks when it subscribes for data changes under persist-watcher mode. + * Previously addPersistListener inferred the listener kind via `instanceof` and matched + * IZkChildListener first, so a data subscription on such a listener was misrouted into the + * child-listener map and its data-change events were silently dropped. + */ + @Test + void testDualInterfaceListenerDataChangeUnderPersistWatcher() throws Exception { + org.apache.helix.zookeeper.impl.client.ZkClient.Builder builder = + new org.apache.helix.zookeeper.impl.client.ZkClient.Builder(); + builder.setZkServer(ZkTestBase.ZK_ADDR).setMonitorRootPathOnly(false).setUsePersistWatcher(true); + org.apache.helix.zookeeper.impl.client.ZkClient zkClient = builder.build(); + zkClient.setZkSerializer(new BasicZkSerializer(new SerializableSerializer())); + + int count = 50; + String path = "/testDualInterfaceListenerDataChange"; + CountDownLatch dataLatch = new CountDownLatch(count); + CountDownLatch childLatch = new CountDownLatch(count); + + // A single listener implementing BOTH interfaces, mirroring Helix's CallbackHandler. + class DualListener implements IZkDataListener, IZkChildListener { + @Override + public void handleDataChange(String dataPath, Object data) { + dataLatch.countDown(); + } + + @Override + public void handleDataDeleted(String dataPath) { + } + + @Override + public void handleChildChange(String parentPath, List currentChilds) { + childLatch.countDown(); + } + } + DualListener dualListener = new DualListener(); + + zkClient.create(path, "data", CreateMode.PERSISTENT); + zkClient.subscribeDataChanges(path, dualListener); + zkClient.subscribeChildChanges(path, dualListener); + + // Data updates must reach handleDataChange (the regression that this test guards). + for (int i = 0; i < count; ++i) { + zkClient.writeData(path, "data" + i, -1); + } + Assert.assertTrue(dataLatch.await(15000, TimeUnit.MILLISECONDS), + "Data-change events were not delivered to a dual-interface listener under persist mode"); + + // Child changes must also reach handleChildChange for the same listener. + for (int i = 0; i < count; ++i) { + zkClient.create(path + "/c" + i, "data", CreateMode.PERSISTENT); + } + Assert.assertTrue(childLatch.await(15000, TimeUnit.MILLISECONDS), + "Child-change events were not delivered to a dual-interface listener under persist mode"); + + zkClient.deleteRecursively(path); + zkClient.close(); + } + @Test(dependsOnMethods = "testZkClientDataChange") void testZkClientChildChange() throws Exception { org.apache.helix.zookeeper.impl.client.ZkClient.Builder builder =