diff --git a/meta-client/src/main/java/org/apache/helix/metaclient/impl/zk/ZkMetaClient.java b/meta-client/src/main/java/org/apache/helix/metaclient/impl/zk/ZkMetaClient.java
index 7941deb..4398e8e 100644
--- a/meta-client/src/main/java/org/apache/helix/metaclient/impl/zk/ZkMetaClient.java
+++ b/meta-client/src/main/java/org/apache/helix/metaclient/impl/zk/ZkMetaClient.java
@@ -56,6 +56,7 @@
 import org.apache.helix.zookeeper.zkclient.IZkStateListener;
 import org.apache.helix.zookeeper.zkclient.ZkConnection;
 import org.apache.helix.zookeeper.zkclient.exception.ZkException;
+import org.apache.helix.zookeeper.zkclient.exception.ZkInterruptedException;
 import org.apache.zookeeper.CreateMode;
 import org.apache.zookeeper.KeeperException;
 import org.apache.zookeeper.Watcher;
@@ -332,15 +333,13 @@
     return true;
   }
 
-  // TODO: add impl and remove UnimplementedException
   @Override
   public boolean subscribeChildChanges(String key, ChildChangeListener listener, boolean skipWatchingNonExistNode) {
-    try {
-      _zkClient.subscribePersistRecursiveWatcher(key, new ChildListenerAdapter(listener));
-    } catch (KeeperException.UnimplementedException e) {
-      LOG.error(e.getLocalizedMessage());
+    if (skipWatchingNonExistNode && exists(key) == null) {
+      return false;
     }
-    return false;
+    _zkClient.subscribePersistRecursiveListener(key, new ChildListenerAdapter(listener));
+    return true;
   }
 
   @Override
@@ -356,11 +355,7 @@
   // TODO: add impl and remove UnimplementedException
   @Override
   public void unsubscribeChildChanges(String key, ChildChangeListener listener) {
-    try{
-    _zkClient.unsubscribePersistRecursiveWatcher(key, new ChildListenerAdapter(listener));
-    } catch (KeeperException.UnimplementedException e) {
-      LOG.error(e.getLocalizedMessage());
-    }
+    _zkClient.unsubscribePersistRecursiveListener(key, new ChildListenerAdapter(listener));
   }
 
   @Override
diff --git a/meta-client/src/main/java/org/apache/helix/metaclient/impl/zk/adapter/ChildListenerAdapter.java b/meta-client/src/main/java/org/apache/helix/metaclient/impl/zk/adapter/ChildListenerAdapter.java
index e7d8856..19c1d64 100644
--- a/meta-client/src/main/java/org/apache/helix/metaclient/impl/zk/adapter/ChildListenerAdapter.java
+++ b/meta-client/src/main/java/org/apache/helix/metaclient/impl/zk/adapter/ChildListenerAdapter.java
@@ -39,7 +39,7 @@
   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;
+      case NodeDataChanged: return ChildChangeListener.ChangeType.ENTRY_DATA_CHANGE;
       case NodeDeleted: return ChildChangeListener.ChangeType.ENTRY_DELETED;
       default: throw new IllegalArgumentException("EventType " + eventType + " is not supported.");
     }
diff --git a/meta-client/src/test/java/org/apache/helix/metaclient/impl/zk/TestZkMetaClient.java b/meta-client/src/test/java/org/apache/helix/metaclient/impl/zk/TestZkMetaClient.java
index cbefd21..69724c9 100644
--- a/meta-client/src/test/java/org/apache/helix/metaclient/impl/zk/TestZkMetaClient.java
+++ b/meta-client/src/test/java/org/apache/helix/metaclient/impl/zk/TestZkMetaClient.java
@@ -19,6 +19,7 @@
  * under the License.
  */
 
+import org.apache.helix.metaclient.api.ChildChangeListener;
 import org.apache.helix.metaclient.api.DataUpdater;
 import org.apache.helix.metaclient.api.MetaClientInterface;
 import org.apache.helix.metaclient.exception.MetaClientException;
@@ -55,6 +56,7 @@
 
   private static final String TRANSACTION_TEST_PARENT_PATH = "/transactionOpTestPath";
   private static final String TEST_INVALID_PATH = "/_invalid/a/b/c";
+  private static final int DEFAULT_LISTENER_WAIT_TIMEOUT = 5000;
 
   private final Object _syncObject = new Object();
 
@@ -316,7 +318,7 @@
         }
       }
       zkMetaClient.set(basePath + "_1", testData, -1);
-      Assert.assertTrue(countDownLatch.await(5000, TimeUnit.MILLISECONDS));
+      Assert.assertTrue(countDownLatch.await(DEFAULT_LISTENER_WAIT_TIMEOUT, TimeUnit.MILLISECONDS));
       Assert.assertTrue(dataExpected.get());
     }
   }
@@ -347,7 +349,7 @@
       Assert.assertEquals(watchers.get("persistentWatches").get(0), basePath);
       Assert.assertEquals(watchers.get("childWatches").size(), 0);
       Assert.assertEquals(watchers.get("dataWatches").size(), 0);
-      Assert.assertTrue(countDownLatch.await(5000, TimeUnit.MILLISECONDS));
+      Assert.assertTrue(countDownLatch.await(DEFAULT_LISTENER_WAIT_TIMEOUT, TimeUnit.MILLISECONDS));
 
       zkMetaClient.unsubscribeDirectChildChange(basePath, listener);
       // verify that no listener is registered on any path
@@ -362,7 +364,6 @@
   public void testDataChangeListener() throws Exception {
     final String basePath = "/TestZkMetaClient_testDataChangeListener";
     final int count = 200;
-    final int[] get_count = {0};
     try (ZkMetaClient<String> zkMetaClient = createZkMetaClient()) {
       zkMetaClient.connect();
       CountDownLatch countDownLatch = new CountDownLatch(count);
@@ -372,7 +373,6 @@
         public void handleDataChange(String key, Object data, ChangeType changeType)
             throws Exception {
           if(changeType == ENTRY_UPDATE) {
-            get_count[0]++;
             countDownLatch.countDown();
           }
         }
@@ -391,13 +391,84 @@
       for (int i=0; i<200; ++i) {
         zkMetaClient.set(basePath, "data7" + i, -1);
       }
-      Assert.assertTrue(countDownLatch.await(5000, TimeUnit.MILLISECONDS));
+      Assert.assertTrue(countDownLatch.await(DEFAULT_LISTENER_WAIT_TIMEOUT, TimeUnit.MILLISECONDS));
 
 
       zkMetaClient.unsubscribeDataChange(basePath, listener);
       // verify that no listener is registered on any path
       watchers = TestUtil.getZkWatch(zkMetaClient.getZkClient());
+      Assert.assertEquals(watchers.get("persistentWatches").size(), 0);
+      Assert.assertEquals(watchers.get("childWatches").size(), 0);
+      Assert.assertEquals(watchers.get("dataWatches").size(), 0);
+
+    }
+  }
+
+  @Test
+  public void testChildChangeListener() throws Exception {
+    final String basePath = "/TestZkMetaClient_testChildChangeListener";
+    final int count = 100;
+    try (ZkMetaClient<String> zkMetaClient = createZkMetaClient()) {
+      zkMetaClient.connect();
+      CountDownLatch countDownLatch = new CountDownLatch(count*4);
+      ChildChangeListener listener = new ChildChangeListener() {
+
+        @Override
+        public void handleChildChange(String changedPath, ChangeType changeType) throws Exception {
+          countDownLatch.countDown();
+
+        }
+      };
+      zkMetaClient.create(basePath, "");
+      Assert.assertTrue(
+          zkMetaClient.subscribeChildChanges(basePath, listener, false)
+      );
+
+      DataChangeListener dummyDataListener = new DataChangeListener() {
+        @Override
+        public void handleDataChange(String key, Object data, ChangeType changeType)
+            throws Exception {
+        }
+      };
+      try {
+        zkMetaClient.subscribeDataChange(basePath, dummyDataListener, false);
+        Assert.fail("subscribeDataChange should throw exception");
+      } catch (UnsupportedOperationException ex) {
+        // we are expecting a UnsupportedOperationException, continue with test.
+      }
+
+      DirectChildChangeListener dummyCldListener = new DirectChildChangeListener() {
+        @Override
+        public void handleDirectChildChange(String key) throws Exception {
+
+        }
+      };
+      try {
+        zkMetaClient.subscribeDirectChildChange(basePath, dummyCldListener, false);
+      } catch ( Exception ex) {
+        Assert.assertEquals(ex.getClass().getName(), "java.lang.UnsupportedOperationException");
+      }
+
+      // Verify no one time watcher is registered. Only one persist listener is registered.
+      Map<String, List<String>> watchers = TestUtil.getZkWatch(zkMetaClient.getZkClient());
+      Assert.assertEquals(watchers.get("persistentRecursiveWatches").size(), 1);
+      Assert.assertEquals(watchers.get("persistentRecursiveWatches").get(0), basePath);
+      Assert.assertEquals(watchers.get("persistentWatches").size(), 0);
+      Assert.assertEquals(watchers.get("childWatches").size(), 0);
+      Assert.assertEquals(watchers.get("dataWatches").size(), 0);
+
+      for (int i=0; i<count; ++i) {
+        zkMetaClient.set(basePath, "data7" + i, -1);
+        zkMetaClient.create(basePath+"/c1_" +i , "datat");
+        zkMetaClient.create(basePath+"/c1_" +i + "/c2", "datat");
+        zkMetaClient.delete(basePath+"/c1_" +i + "/c2");
+      }
+      Assert.assertTrue(countDownLatch.await(5000, TimeUnit.MILLISECONDS));
+
+      zkMetaClient.unsubscribeChildChanges(basePath, listener);
+      // verify that no listener is registered on any path
       watchers = TestUtil.getZkWatch(zkMetaClient.getZkClient());
+      Assert.assertEquals(watchers.get("persistentRecursiveWatches").size(), 0);
       Assert.assertEquals(watchers.get("persistentWatches").size(), 0);
       Assert.assertEquals(watchers.get("childWatches").size(), 0);
       Assert.assertEquals(watchers.get("dataWatches").size(), 0);
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 13dd510..064f6b4 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
@@ -343,17 +343,54 @@
    * 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);
+      }
+      // subscribe a PERSISTENT_RECURSIVE listener on path. It throws exception if not successful
+      retryUntilConnected(() -> {
+        getConnection().addWatch(path, ZkClient.this, AddWatchMode.PERSISTENT_RECURSIVE);
+        return null;
+      });
+
+      _zkPathRecursiveWatcherTrie.addRecursiveListener(path, recursivePersistListener);
+    };
+
+    executeWithInPersistListenerMutex(addListener);
   }
 
-  public boolean unsubscribePersistRecursiveWatcher(String path,
-      RecursivePersistListener recursivePersistListener)
-      throws KeeperException.UnimplementedException {
-    throw new KeeperException.UnimplementedException();
+  public void unsubscribePersistRecursiveListener(String path,
+      RecursivePersistListener recursivePersistListener) {
+    if (!_usePersistWatcher) {
+      throw new UnsupportedOperationException(
+          "Can not subscribe PersistRecursiveWatcher. Persist listener is not enabled.");
+    }
+
+    // unsubscribe from ZK if this is the only recursive persist listener on this path.
+    ManipulateListener removeListeners = () -> {
+      _zkPathRecursiveWatcherTrie.removeRecursiveListener(path, recursivePersistListener);
+      if (_zkPathRecursiveWatcherTrie.hasListenerOnPath(path)) {
+        return;
+      }
+      try {
+        // We are not checking if there is a persist listener registered on the path because
+        // we do not allow subscribe a persist listener on the same path of persist recursive
+        // listener as of now.
+        getConnection().removeWatches(path, this, WatcherType.Any);
+      } catch (KeeperException.NoWatcherException e) {
+        LOG.warn("Persist watcher is already removed on path: {}", path);
+      }
+    };
+    executeWithInPersistListenerMutex(removeListeners);
+
   }
 
   private boolean isPrefetchEnabled(IZkDataListener dataListener) {
@@ -1372,7 +1409,7 @@
    * are deleted before the last page is fetched. The upstream caller should be able to handle this.
    */
   public List<String> getChildren(String path) {
-    return getChildren(path, (!_usePersistWatcher) && hasListeners(path));
+    return getChildren(path, (!_usePersistWatcher) && hasChildOrDataListeners(path));
   }
 
   protected List<String> getChildren(final String path, final boolean watch) {
@@ -1437,7 +1474,7 @@
   }
 
   public boolean exists(final String path) {
-    return exists(path, hasListeners(path));
+    return exists(path, hasChildOrDataListeners(path));
   }
 
   protected boolean exists(final String path, final boolean watch) {
@@ -1724,7 +1761,7 @@
     }
   }
 
-  private boolean hasListeners(String path) {
+  private boolean hasChildOrDataListeners(String path) {
     Set<IZkDataListenerEntry> dataListeners = _dataListener.get(path);
     if (dataListeners != null && dataListeners.size() > 0) {
       return true;
@@ -1807,6 +1844,23 @@
         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.isEmpty()) {
+          for (final RecursivePersistListener listener : recListeners) {
+            _eventThread.send(
+                new ZkEventThread.ZkEvent("Data of " + path + " changed sent to " + listener) {
+                  @Override
+                  public void run() throws Exception {
+                    listener.handleZNodeChange(path, event.getType());
+                  }
+                });
+          }
+        }
+      }
     }
   }
 
@@ -1873,7 +1927,7 @@
           public void run() throws Exception {
             if (!pathStatRecord.pathChecked()) {
               Stat stat = null;
-              if (_usePersistWatcher || !pathExists || !hasListeners(path)) {
+              if (_usePersistWatcher || !pathExists || !hasChildOrDataListeners(path)) {
                 // will not install listener using exists call
                 stat = getStat(path, false);
               } else {
@@ -2178,7 +2232,7 @@
 
   @SuppressWarnings("unchecked")
   public <T extends Object> T readData(String path, Stat stat) {
-    return (T) readData(path, stat, hasListeners(path));
+    return (T) readData(path, stat, hasChildOrDataListeners(path));
   }
 
   @SuppressWarnings("unchecked")
@@ -3030,8 +3084,16 @@
     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) {
@@ -3044,6 +3106,7 @@
 
   // 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) {
 
     ManipulateListener removeListeners = () -> {
@@ -3053,8 +3116,9 @@
         } else if (listener instanceof IZkDataListener) {
           removeDataListener(path, (IZkDataListener) listener);
         }
-        if (!hasListeners(path)) {
-          // TODO: update hasListeners logic when recursive persist listener is added
+        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) {
diff --git a/zookeeper-api/src/main/java/org/apache/helix/zookeeper/zkclient/util/ZkPathRecursiveWatcherTrie.java b/zookeeper-api/src/main/java/org/apache/helix/zookeeper/zkclient/util/ZkPathRecursiveWatcherTrie.java
index 9adcb3a..41a9dd1 100644
--- a/zookeeper-api/src/main/java/org/apache/helix/zookeeper/zkclient/util/ZkPathRecursiveWatcherTrie.java
+++ b/zookeeper-api/src/main/java/org/apache/helix/zookeeper/zkclient/util/ZkPathRecursiveWatcherTrie.java
@@ -164,7 +164,6 @@
         result.addAll(cur.getRecursiveListeners());
       }
     }
-
     return result;
   }
 
@@ -225,6 +224,28 @@
   }
 
   /**
+   * Return if there is listener on a particular path
+   * @param path
+   * @return
+   */
+  public boolean hasListenerOnPath(String path) {
+    Objects.requireNonNull(path, "Path cannot be null");
+
+    final List<String> pathComponents = split(path);
+    TrieNode cur;
+    synchronized (this) {
+      cur = _rootNode;
+      for (final String element : pathComponents) {
+        cur = cur.getChild(element);
+        if (cur == null) {
+          break;
+        }
+      }
+    }
+    return cur != null && !cur.getRecursiveListeners().isEmpty();
+  }
+
+  /**
    * Clear all nodes in the trie.
    */
   public synchronized void clear() {
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 b841bcd..76f5352 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
@@ -116,39 +116,62 @@
     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(false);
+        .setUsePersistWatcher(true);
     org.apache.helix.zookeeper.impl.client.ZkClient zkClient = builder.build();
     zkClient.setZkSerializer(new BasicZkSerializer(new SerializableSerializer()));
     int count = 100;
     final AtomicInteger[] event_count = {new AtomicInteger(0)};
     final AtomicInteger[] event_count2 = {new AtomicInteger(0)};
-    CountDownLatch countDownLatch1 = new CountDownLatch(count);
-    CountDownLatch countDownLatch2 = new CountDownLatch(count/2);
-    String path = "/base/testZkClientChildChange";
+    // for each iteration, we will edit a node, create a child, create a grand child, and
+    // delete child. Expect 4 event per iteration. -> total event should be count*4
+    CountDownLatch countDownLatch1 = new CountDownLatch(count*4);
+    CountDownLatch countDownLatch2 = new CountDownLatch(count);
+    String path = "/testZkClientPersistRecursiveChange";
     RecursivePersistListener rcListener = new RecursivePersistListener() {
       @Override
       public void handleZNodeChange(String dataPath, Watcher.Event.EventType eventType)
           throws Exception {
         countDownLatch1.countDown();
         event_count[0].incrementAndGet() ;
-        System.out.println("rcListener count " + event_count[0]);
       }
     };
+    zkClient.create(path, "datat", CreateMode.PERSISTENT);
+    zkClient.subscribePersistRecursiveListener(path, rcListener);
+    for (int i=0; i<count; ++i) {
+      zkClient.writeData(path, "data7" + i, -1);
+      zkClient.create(path+"/c1_" +i , "datat", CreateMode.PERSISTENT);
+      zkClient.create(path+"/c1_" +i + "/c2", "datat", CreateMode.PERSISTENT);
+      zkClient.delete(path+"/c1_" +i + "/c2");
+    }
+    Assert.assertTrue(countDownLatch1.await(50000000, TimeUnit.MILLISECONDS));
+
+    // subscribe a persist child watch, it should throw exception
     IZkChildListener childListener2 = new IZkChildListener() {
       @Override
       public void handleChildChange(String parentPath, List<String> currentChilds)
           throws Exception {
         countDownLatch2.countDown();
         event_count2[0].incrementAndGet();
-        System.out.println("childListener2 count " + event_count2[0]);
       }
     };
     try {
-      zkClient.subscribePersistRecursiveWatcher(path, rcListener);
-    } catch (KeeperException.UnimplementedException e) {
-      e.printStackTrace();
+      zkClient.subscribeChildChanges(path, childListener2, false);
+    } catch ( Exception ex) {
+      Assert.assertEquals(ex.getClass().getName(), "java.lang.UnsupportedOperationException");
     }
 
+    // unsubscribe recursive persist watcher, and subscribe persist watcher should success.
+    zkClient.unsubscribePersistRecursiveListener(path, rcListener);
+    zkClient.subscribeChildChanges(path, childListener2, false);
+    // we should only get 100 event since only 100 direct child change.
+    for (int i=0; i<count; ++i) {
+      zkClient.writeData(path, "data7" + i, -1);
+      zkClient.create(path+"/c2_" +i , "datat", CreateMode.PERSISTENT);
+      zkClient.create(path+"/c2_" +i + "/c3", "datat", CreateMode.PERSISTENT);
+      zkClient.delete(path+"/c2_" +i + "/c3");
+    }
+    Assert.assertTrue(countDownLatch2.await(50000000, TimeUnit.MILLISECONDS));
+
     zkClient.close();
   }