Skip to content

Commit e61e1ed

Browse files
LZD-PratyushBhattCopilot
LZD-PratyushBhatt
and
Copilot
committed
Add persist-recursive watch for controller current-state (flag-gated)
The controller cold-rebuilds every participant CURRENTSTATES subtree with one child watch plus one data watch per partition (O(N*M)) and re-registers the whole set on every leadership handoff / session change. On large super-clusters this is 10^5-10^6 watches re-armed per handoff and the dominant cause of multi-minute MissingTopState SLA breaches. Behind a new flag (helix.controller.currentState.persistRecursiveWatch.enabled, default off), the controller's ZkClient runs with usePersistWatcher=true and each CURRENT_STATE / TASK_CURRENT_STATE CallbackHandler installs ONE ZooKeeper 3.6+ PERSISTENT_RECURSIVE watch over the participant subtree instead of the per-child watch loop, collapsing the watch footprint and per-handoff re-subscribe cost from O(N*M) to O(1). Measured on the real Helix ZkClient: 50k current-state children => 51,000 watches / ~7s subscribe -> 2 watches / ~2ms. - zookeeper-api: expose subscribe/unsubscribePersistRecursiveListener on RealmAwareZkClient (default-throwing; DedicatedZkClient delegates to its raw client) + usePersistWatcher in RealmAwareZkClientConfig, plumbed into the dedicated client. - helix-common: new SystemPropertyKeys.CURRENT_STATE_PERSIST_RECURSIVE_WATCH_ENABLED. - helix-core: ZKHelixManager builds the controller client in persist mode when enabled; CallbackHandler installs/handles (handleZNodeChange)/removes the recursive watch for current-state, falling back to per-node watches if unsupported. - test: TestCurrentStatePersistRecursiveWatch converges at steady state and re-converges after a participant failure with the flag on. Regression (flag off): TestZkCallbackHandlerLeak unchanged. Design doc: docs/design/001-current-state-persist-recursive-watch.md. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
1 parent 3685c4b commit e61e1ed

6 files changed

Lines changed: 253 additions & 2 deletions

File tree

‎helix-common/src/main/java/org/apache/helix/SystemPropertyKeys.java‎

Lines changed: 8 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -97,4 +97,12 @@ public class SystemPropertyKeys {
9797

9898
// Stage thread pool size for parallel stage execution
9999
public static final String STAGE_THREAD_POOL_SIZE = "helix.stage.threadpool.size";
100+
101+
// When enabled, the controller subscribes a single PERSISTENT_RECURSIVE ZooKeeper watch per
102+
// participant CURRENTSTATES subtree instead of one child watch + one data watch per partition.
103+
// This collapses the per-handoff watch footprint and the cold-start subscribe cost from O(N*M)
104+
// to O(1) re-arm. Requires the controller's ZkClient to run with usePersistWatcher=true (wired
105+
// automatically when this flag is set). Off by default for backward compatibility.
106+
public static final String CURRENT_STATE_PERSIST_RECURSIVE_WATCH_ENABLED =
107+
"helix.controller.currentState.persistRecursiveWatch.enabled";
100108
}

‎helix-core/src/main/java/org/apache/helix/manager/zk/CallbackHandler.java‎

Lines changed: 63 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -79,6 +79,7 @@
7979
import org.apache.helix.zookeeper.datamodel.ZNRecord;
8080
import org.apache.helix.zookeeper.zkclient.IZkChildListener;
8181
import org.apache.helix.zookeeper.zkclient.IZkDataListener;
82+
import org.apache.helix.zookeeper.zkclient.RecursivePersistListener;
8283
import org.apache.helix.zookeeper.zkclient.annotation.PreFetchChangedData;
8384
import org.apache.helix.zookeeper.zkclient.exception.ZkNoNodeException;
8485
import org.apache.zookeeper.Watcher.Event.EventType;
@@ -105,7 +106,7 @@
105106
import static org.apache.helix.HelixConstants.ChangeType.TASK_CURRENT_STATE;
106107

107108
@PreFetchChangedData(enabled = false)
108-
public class CallbackHandler implements IZkChildListener, IZkDataListener {
109+
public class CallbackHandler implements IZkChildListener, IZkDataListener, RecursivePersistListener {
109110
private static Logger logger = LoggerFactory.getLogger(CallbackHandler.class);
110111
private static final AtomicLong CALLBACK_HANDLER_UID = new AtomicLong();
111112

@@ -138,6 +139,11 @@ public class CallbackHandler implements IZkChildListener, IZkDataListener {
138139
private boolean _watchChild = true; // Whether we should subscribe to the child znode's data
139140
// change.
140141

142+
// When true (controller CURRENT_STATE/TASK_CURRENT_STATE handler + flag enabled + persist-watcher
143+
// client), this handler installs ONE PERSISTENT_RECURSIVE watch covering its whole subtree instead
144+
// of one child watch plus one data watch per partition.
145+
private final boolean _useRecursivePersistWatch;
146+
141147
// indicated whether this CallbackHandler is ready to serve event callback from ZkClient.
142148
private boolean _ready = false;
143149

@@ -186,6 +192,10 @@ public CallbackHandler(HelixManager manager, RealmAwareZkClient client, Property
186192

187193
parseListenerProperties();
188194

195+
_useRecursivePersistWatch =
196+
Boolean.getBoolean(SystemPropertyKeys.CURRENT_STATE_PERSIST_RECURSIVE_WATCH_ENABLED)
197+
&& (_changeType == CURRENT_STATE || _changeType == TASK_CURRENT_STATE);
198+
189199
init();
190200
}
191201

@@ -563,6 +573,21 @@ private void subscribeForChanges(NotificationContext.Type callbackType, String p
563573
_uid, path, callbackType, _eventTypes, _listener, watchChild);
564574

565575
long start = System.currentTimeMillis();
576+
577+
if (_useRecursivePersistWatch) {
578+
try {
579+
// One PERSISTENT_RECURSIVE watch covers all data + child changes under the entire subtree,
580+
// replacing the per-child subscribeChildChange + subscribeDataChange loop below. It is also
581+
// not re-armed per event, so the per-handoff re-subscribe cost drops from O(N*M) to O(1).
582+
_zkClient.subscribePersistRecursiveListener(path, this);
583+
logger.info("CallbackHandler {} installed ONE persistent recursive watch on path: {} "
584+
+ "(replaces per-child data watches for change type {})", _uid, path, _changeType);
585+
return;
586+
} catch (UnsupportedOperationException e) {
587+
logger.warn("CallbackHandler {} persist recursive watch unsupported by zk client; falling "
588+
+ "back to per-node watches on path: {}", _uid, path);
589+
}
590+
}
566591
if (_eventTypes.contains(EventType.NodeDataChanged)
567592
|| _eventTypes.contains(EventType.NodeCreated)
568593
|| _eventTypes.contains(EventType.NodeDeleted)) {
@@ -752,6 +777,33 @@ public void handleChildChange(String parentPath, List<String> currentChilds) {
752777
}
753778
}
754779

780+
@Override
781+
public void handleZNodeChange(String dataPath, EventType eventType) {
782+
// A single PERSISTENT_RECURSIVE watch fired for a change to some node anywhere under _path.
783+
// Route it to the same CALLBACK path that the per-node child/data watches would have used so the
784+
// controller pipeline reacts identically (it re-reads current state regardless of which node).
785+
try {
786+
updateNotificationTime(System.nanoTime());
787+
if (dataPath != null && dataPath.startsWith(_path)) {
788+
if (!isReady()) {
789+
logger.info("CallbackHandler {} is in reset state; skip recursive {} event on path: {}",
790+
_uid, eventType, dataPath);
791+
return;
792+
}
793+
NotificationContext changeContext = new NotificationContext(_manager);
794+
changeContext.setType(NotificationContext.Type.CALLBACK);
795+
changeContext.setPathChanged(dataPath);
796+
changeContext.setChangeType(_changeType);
797+
changeContext.setIsChildChange(true);
798+
enqueueTask(changeContext);
799+
}
800+
} catch (Exception e) {
801+
String msg = "exception in handling recursive znode-change. path: " + dataPath + ", listener: "
802+
+ _listener;
803+
ZKExceptionHandler.getInstance().handle(msg, e);
804+
}
805+
}
806+
755807
/**
756808
* Invoke the listener for the last time so that the listener could clean up resources
757809
*/
@@ -764,6 +816,16 @@ public void reset(boolean isShutdown) {
764816
logger.info("Resetting CallbackHandler: {}. Is resetting for shutdown: {}.", _uid, isShutdown);
765817
try {
766818
_ready = false;
819+
if (_useRecursivePersistWatch) {
820+
// The recursive watch is persistent, so it must be explicitly removed (one call) on reset /
821+
// session change, otherwise it would leak. init() re-installs the single watch afterward.
822+
try {
823+
_zkClient.unsubscribePersistRecursiveListener(_path, this);
824+
} catch (Exception e) {
825+
logger.warn("CallbackHandler {} failed to unsubscribe recursive watch on path: {}, {}",
826+
_uid, _path, e.toString());
827+
}
828+
}
767829
CallbackEventExecutor callbackExecutor = _batchCallbackExecutorRef.get();
768830
if (callbackExecutor != null) {
769831
if (isShutdown) {

‎helix-core/src/main/java/org/apache/helix/manager/zk/ZKHelixManager.java‎

Lines changed: 8 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1522,6 +1522,14 @@ private RealmAwareZkClient createSingleRealmZkClient() {
15221522
.setMonitorInstanceName(_instanceName)
15231523
.setMonitorRootPathOnly(isMonitorRootPathOnly());
15241524

1525+
// When the persist-recursive current-state watch is enabled, the controller's client must run
1526+
// in persistent-watcher mode so CallbackHandler can install one recursive watch per subtree.
1527+
if (Boolean.getBoolean(SystemPropertyKeys.CURRENT_STATE_PERSIST_RECURSIVE_WATCH_ENABLED)
1528+
&& (_instanceType == InstanceType.CONTROLLER
1529+
|| _instanceType == InstanceType.CONTROLLER_PARTICIPANT)) {
1530+
clientConfig.setUsePersistWatcher(true);
1531+
}
1532+
15251533
if (_instanceType == InstanceType.ADMINISTRATOR) {
15261534
return resolveZkClient(SharedZkClientFactory.getInstance(), _realmAwareZkConnectionConfig,
15271535
clientConfig);

‎helix-core/src/test/java/org/apache/helix/integration/TestCurrentStatePersistRecursiveWatch.java‎

Lines changed: 120 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,120 @@
1+
package org.apache.helix.integration;
2+
3+
/*
4+
* Licensed to the Apache Software Foundation (ASF) under one
5+
* or more contributor license agreements. See the NOTICE file
6+
* distributed with this work for additional information
7+
* regarding copyright ownership. The ASF licenses this file
8+
* to you under the Apache License, Version 2.0 (the
9+
* "License"); you may not use this file except in compliance
10+
* with the License. You may obtain a copy of the License at
11+
*
12+
* http://www.apache.org/licenses/LICENSE-2.0
13+
*
14+
* Unless required by applicable law or agreed to in writing,
15+
* software distributed under the License is distributed on an
16+
* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
17+
* KIND, either express or implied. See the License for the
18+
* specific language governing permissions and limitations
19+
* under the License.
20+
*/
21+
22+
import java.util.Date;
23+
24+
import org.apache.helix.SystemPropertyKeys;
25+
import org.apache.helix.TestHelper;
26+
import org.apache.helix.ZkUnitTestBase;
27+
import org.apache.helix.integration.manager.ClusterControllerManager;
28+
import org.apache.helix.integration.manager.MockParticipantManager;
29+
import org.apache.helix.tools.ClusterVerifiers.BestPossibleExternalViewVerifier;
30+
import org.apache.helix.tools.ClusterVerifiers.ZkHelixClusterVerifier;
31+
import org.testng.Assert;
32+
import org.testng.annotations.AfterClass;
33+
import org.testng.annotations.BeforeClass;
34+
import org.testng.annotations.Test;
35+
36+
/**
37+
* End-to-end test for the persist-recursive current-state watch
38+
* ({@link SystemPropertyKeys#CURRENT_STATE_PERSIST_RECURSIVE_WATCH_ENABLED}).
39+
*
40+
* With the flag enabled, the controller's ZkClient is built with usePersistWatcher=true and each
41+
* CURRENT_STATE / TASK_CURRENT_STATE CallbackHandler installs ONE PERSISTENT_RECURSIVE watch on the
42+
* participant's CURRENTSTATES subtree instead of one child watch plus one data watch per partition.
43+
*
44+
* The controller can only compute a correct ExternalView if it actually receives the participants'
45+
* current-state changes. So a green {@link BestPossibleExternalViewVerifier} both at steady state and
46+
* after an ongoing current-state change (a participant failure that forces masters to move) proves the
47+
* single recursive watch delivers initial AND incremental current-state events correctly.
48+
*/
49+
public class TestCurrentStatePersistRecursiveWatch extends ZkUnitTestBase {
50+
51+
@BeforeClass
52+
public void beforeClass() {
53+
System.setProperty(SystemPropertyKeys.CURRENT_STATE_PERSIST_RECURSIVE_WATCH_ENABLED, "true");
54+
}
55+
56+
@AfterClass
57+
public void afterClass() {
58+
System.clearProperty(SystemPropertyKeys.CURRENT_STATE_PERSIST_RECURSIVE_WATCH_ENABLED);
59+
}
60+
61+
@Test
62+
public void testControllerConvergesWithRecursiveCurrentStateWatch() throws Exception {
63+
String className = TestHelper.getTestClassName();
64+
String methodName = TestHelper.getTestMethodName();
65+
String clusterName = className + "_" + methodName;
66+
final int n = 3;
67+
68+
System.out.println("START " + clusterName + " at " + new Date(System.currentTimeMillis()));
69+
70+
TestHelper.setupCluster(clusterName, ZK_ADDR, 12918, // participant port
71+
"localhost", // participant name prefix
72+
"TestDB", // resource name prefix
73+
1, // resources
74+
8, // partitions per resource
75+
n, // number of nodes
76+
3, // replicas
77+
"MasterSlave", true); // do rebalance
78+
79+
MockParticipantManager[] participants = new MockParticipantManager[n];
80+
for (int i = 0; i < n; i++) {
81+
String instanceName = "localhost_" + (12918 + i);
82+
participants[i] = new MockParticipantManager(ZK_ADDR, clusterName, instanceName);
83+
participants[i].syncStart();
84+
}
85+
86+
// The controller's ZkClient is built with usePersistWatcher=true because the flag is set, so its
87+
// CURRENT_STATE CallbackHandlers use the single recursive watch.
88+
ClusterControllerManager controller =
89+
new ClusterControllerManager(ZK_ADDR, clusterName, "controller_0");
90+
controller.syncStart();
91+
92+
ZkHelixClusterVerifier verifier =
93+
new BestPossibleExternalViewVerifier.Builder(clusterName).setZkClient(_gZkClient)
94+
.setWaitTillVerify(TestHelper.DEFAULT_REBALANCE_PROCESSING_WAIT_TIME).build();
95+
96+
// Steady state: controller must have observed every participant's current state through the
97+
// recursive watch to make ExternalView == BestPossible.
98+
Assert.assertTrue(verifier.verifyByPolling(),
99+
"Cluster did not converge at steady state with the recursive current-state watch");
100+
101+
// Ongoing current-state changes: drop one participant; the masters it hosted must move, which the
102+
// controller can only carry out by observing OFFLINE->SLAVE->MASTER current-state transitions on
103+
// the surviving participants (delivered by the recursive watch). Re-convergence proves incremental
104+
// current-state events are delivered.
105+
participants[0].syncStop();
106+
Assert.assertTrue(verifier.verifyByPolling(),
107+
"Cluster did not re-converge after a participant failure; recursive current-state watch did "
108+
+ "not deliver incremental events");
109+
110+
// Cleanup.
111+
controller.syncStop();
112+
for (int i = 0; i < n; i++) {
113+
if (participants[i].isConnected()) {
114+
participants[i].syncStop();
115+
}
116+
}
117+
deleteCluster(clusterName);
118+
System.out.println("END " + clusterName + " at " + new Date(System.currentTimeMillis()));
119+
}
120+
}

‎zookeeper-api/src/main/java/org/apache/helix/zookeeper/api/client/RealmAwareZkClient.java‎

Lines changed: 39 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -32,6 +32,7 @@
3232
import org.apache.helix.zookeeper.zkclient.IZkChildListener;
3333
import org.apache.helix.zookeeper.zkclient.IZkDataListener;
3434
import org.apache.helix.zookeeper.zkclient.IZkStateListener;
35+
import org.apache.helix.zookeeper.zkclient.RecursivePersistListener;
3536
import org.apache.helix.zookeeper.zkclient.callback.ZkAsyncCallbacks;
3637
import org.apache.helix.zookeeper.zkclient.exception.ZkTimeoutException;
3738
import org.apache.helix.zookeeper.zkclient.serialize.BasicZkSerializer;
@@ -130,6 +131,32 @@ boolean subscribeDataChanges(String path, IZkDataListener listener,
130131

131132
void unsubscribeAll();
132133

134+
/**
135+
* Subscribe a single PERSISTENT_RECURSIVE watch (ZooKeeper 3.6+) that covers all data and child
136+
* changes under the entire subtree rooted at {@code path}. Unlike the per-node child/data watches,
137+
* this installs ONE server-side watch for the whole subtree and does not need to be re-armed after
138+
* each event. The owning client must be built with {@code usePersistWatcher=true}.
139+
*
140+
* Default implementation throws {@link UnsupportedOperationException}; only clients that support a
141+
* persistent watcher (e.g. the dedicated single-realm client) override it.
142+
*
143+
* @param path the subtree root to watch
144+
* @param listener invoked for every add/remove/data change anywhere under {@code path}
145+
*/
146+
default void subscribePersistRecursiveListener(String path, RecursivePersistListener listener) {
147+
throw new UnsupportedOperationException(
148+
"subscribePersistRecursiveListener is not supported by this RealmAwareZkClient implementation");
149+
}
150+
151+
/**
152+
* Remove a recursive persistent watch previously installed via
153+
* {@link #subscribePersistRecursiveListener(String, RecursivePersistListener)}.
154+
*/
155+
default void unsubscribePersistRecursiveListener(String path, RecursivePersistListener listener) {
156+
throw new UnsupportedOperationException(
157+
"unsubscribePersistRecursiveListener is not supported by this RealmAwareZkClient implementation");
158+
}
159+
133160
// data access
134161
void createPersistent(String path);
135162

@@ -472,6 +499,9 @@ class RealmAwareZkClientConfig {
472499
protected String _monitorKey;
473500
protected String _monitorInstanceName = null;
474501
protected boolean _monitorRootPathOnly = true;
502+
// When true, the client registers PERSISTENT / PERSISTENT_RECURSIVE watches instead of one-shot
503+
// watches, enabling subscribePersistRecursiveListener. Off by default for backward compatibility.
504+
protected boolean _usePersistWatcher = false;
475505

476506
public RealmAwareZkClientConfig setZkSerializer(PathBasedZkSerializer zkSerializer) {
477507
this._zkSerializer = zkSerializer;
@@ -518,6 +548,15 @@ public RealmAwareZkClientConfig setMonitorRootPathOnly(Boolean monitorRootPathOn
518548
return this;
519549
}
520550

551+
public RealmAwareZkClientConfig setUsePersistWatcher(boolean usePersistWatcher) {
552+
this._usePersistWatcher = usePersistWatcher;
553+
return this;
554+
}
555+
556+
public boolean isUsePersistWatcher() {
557+
return _usePersistWatcher;
558+
}
559+
521560
public RealmAwareZkClientConfig setOperationRetryTimeout(Long operationRetryTimeout) {
522561
this._operationRetryTimeout = operationRetryTimeout;
523562
return this;

‎zookeeper-api/src/main/java/org/apache/helix/zookeeper/impl/client/DedicatedZkClient.java‎

Lines changed: 15 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -38,6 +38,7 @@
3838
import org.apache.helix.zookeeper.zkclient.ZkConnection;
3939
import org.apache.helix.zookeeper.zkclient.callback.ZkAsyncCallbacks;
4040
import org.apache.helix.zookeeper.zkclient.IZkStateListener;
41+
import org.apache.helix.zookeeper.zkclient.RecursivePersistListener;
4142
import org.apache.helix.zookeeper.zkclient.serialize.PathBasedZkSerializer;
4243
import org.apache.helix.zookeeper.zkclient.serialize.ZkSerializer;
4344
import org.apache.zookeeper.CreateMode;
@@ -107,7 +108,8 @@ public DedicatedZkClient(RealmAwareZkClient.RealmAwareZkConnectionConfig connect
107108
_rawZkClient = new ZkClient(zkConnection, (int) clientConfig.getConnectInitTimeout(),
108109
clientConfig.getOperationRetryTimeout(), clientConfig.getZkSerializer(),
109110
clientConfig.getMonitorType(), clientConfig.getMonitorKey(),
110-
clientConfig.getMonitorInstanceName(), clientConfig.isMonitorRootPathOnly());
111+
clientConfig.getMonitorInstanceName(), clientConfig.isMonitorRootPathOnly(), true,
112+
clientConfig.isUsePersistWatcher());
111113
}
112114

113115
@Override
@@ -161,6 +163,18 @@ public void unsubscribeAll() {
161163
_rawZkClient.unsubscribeAll();
162164
}
163165

166+
@Override
167+
public void subscribePersistRecursiveListener(String path, RecursivePersistListener listener) {
168+
checkIfPathContainsShardingKey(path);
169+
_rawZkClient.subscribePersistRecursiveListener(path, listener);
170+
}
171+
172+
@Override
173+
public void unsubscribePersistRecursiveListener(String path, RecursivePersistListener listener) {
174+
checkIfPathContainsShardingKey(path);
175+
_rawZkClient.unsubscribePersistRecursiveListener(path, listener);
176+
}
177+
164178
@Override
165179
public void createPersistent(String path) {
166180
createPersistent(path, false);

0 commit comments

Comments
 (0)