Add persist-recursive watch for controller current-state (flag-gated)

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";
 }

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<String> 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) {

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

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);