[GitHub] [helix] rahulrane50 commented on a diff in pull request #2506: ZkClient add recursive persist listener implementation

["rahulrane50 (via GitHub)" <gi...@apache.org>]

rahulrane50 commented on code in PR #2506:
URL: https://github.com/apache/helix/pull/2506#discussion_r1204914109


##########
meta-client/src/main/java/org/apache/helix/metaclient/impl/zk/ZkMetaClient.java:
##########
@@ -335,12 +335,11 @@ public boolean subscribeStateChanges(ConnectStateChangeListener listener) {
   // TODO: add impl and remove UnimplementedException

Review Comment:
   nit. i think we can remove this TODO now.



##########
zookeeper-api/src/main/java/org/apache/helix/zookeeper/zkclient/ZkClient.java:
##########
@@ -343,17 +343,48 @@ public void subscribeDataChanges(String path, IZkDataListener listener) {
    * Subscribe RecursivePersistListener for a particular path. User can only subscribe when
    * `_usePersistWatcher` is set to true and there is no pre-existing watcher on the path.
    */
-  // TODO: Add impl and remove exception
-  public boolean subscribePersistRecursiveWatcher(String path,
-      RecursivePersistListener recursivePersistListener)
-      throws KeeperException.UnimplementedException {
-    throw new KeeperException.UnimplementedException();
+  public void subscribePersistRecursiveListener(String path,
+      RecursivePersistListener recursivePersistListener) {
+    if (!_usePersistWatcher) {
+      throw new UnsupportedOperationException(
+          "Can not subscribe PersistRecursiveWatcher. Persist listener is not enabled.");
+    }
+
+    ManipulateListener addListener = () -> {
+      if (hasChildOrDataListeners(path)) {
+        throw new UnsupportedOperationException(
+            "Can not subscribe PersistRecursiveWatcher. There is an existing listener on " + path);
+      }
+      _zkPathRecursiveWatcherTrie.addRecursiveListener(path, recursivePersistListener);
+      getConnection().addWatch(path, ZkClient.this, AddWatchMode.PERSISTENT_RECURSIVE);

Review Comment:
   Curious, what happens if we were not able to add watch on zk connection but update trie registry with listener. I'm guessing we do infinitely retry for addWatch operation but curious since "subscribePersistRecursiveListener" not an atomic operation now. 



##########
zookeeper-api/src/main/java/org/apache/helix/zookeeper/zkclient/ZkClient.java:
##########
@@ -1807,6 +1838,23 @@ private void processDataOrChildChange(WatchedEvent event, long notificationTime)
         fireDataChangedEvents(event.getPath(), listeners, OptionalLong.of(notificationTime),
             pathExists, event.getType());
       }
+
+      // fire change event for persist recursive listener
+      if (_usePersistWatcher) {
+        Set<RecursivePersistListener> recListeners =
+            _zkPathRecursiveWatcherTrie.getAllRecursiveListeners(path);
+        if (recListeners != null && !recListeners.isEmpty()) {
+          for (final RecursivePersistListener listener : recListeners) {

Review Comment:
   Can we start all zk event threads here in parallel?



##########
zookeeper-api/src/main/java/org/apache/helix/zookeeper/zkclient/util/ZkPathRecursiveWatcherTrie.java:
##########
@@ -224,6 +223,30 @@ public void removeRecursiveListener(final String path, RecursivePersistListener
     }
   }
 
+  /**
+   *
+   * @param path
+   * @return
+   */
+  public Boolean hasListenerOnPath(String path) {
+    Objects.requireNonNull(path, "Path cannot be null");
+
+    final List<String> pathComponents = split(path);
+    Set<RecursivePersistListener> result = new HashSet<>();

Review Comment:
   what is this result used for?



##########
zookeeper-api/src/main/java/org/apache/helix/zookeeper/zkclient/ZkClient.java:
##########
@@ -3030,8 +3078,16 @@ interface ManipulateListener {
     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)) {

Review Comment:
   Curious, we already check that in subscribing listeners right?



##########
zookeeper-api/src/main/java/org/apache/helix/zookeeper/zkclient/ZkClient.java:
##########
@@ -1807,6 +1838,23 @@ private void processDataOrChildChange(WatchedEvent event, long notificationTime)
         fireDataChangedEvents(event.getPath(), listeners, OptionalLong.of(notificationTime),
             pathExists, event.getType());
       }
+
+      // fire change event for persist recursive listener
+      if (_usePersistWatcher) {
+        Set<RecursivePersistListener> recListeners =
+            _zkPathRecursiveWatcherTrie.getAllRecursiveListeners(path);
+        if (recListeners != null && !recListeners.isEmpty()) {

Review Comment:
   can recListeners even be null? I see that we always initialize it in getter method. 



##########
meta-client/src/main/java/org/apache/helix/metaclient/impl/zk/adapter/ChildListenerAdapter.java:
##########
@@ -39,7 +39,7 @@ public ChildListenerAdapter(ChildChangeListener listener) {
   private static ChildChangeListener.ChangeType convertType(Watcher.Event.EventType eventType) {
     switch (eventType) {
       case NodeCreated: return ChildChangeListener.ChangeType.ENTRY_CREATED;
-      case NodeChildrenChanged: return ChildChangeListener.ChangeType.ENTRY_DATA_CHANGE;

Review Comment:
   Is my understanding correct that persistent and recursive watcher here and event "NodeChildrenChanged" in helix manager (callback handler events) are totally different? What I'm trying to understand is after persistent recursive watcher change we would get "NodeDataChanged" event instead of "NodeChildrenChanged" event but i can see in helix manager we still fire events of type "NodeChildrenChanged".



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: reviews-unsubscribe@helix.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@helix.apache.org
For additional commands, e-mail: reviews-help@helix.apache.org