Population of cache

apache · Sep 27, 2023 · d1f01df · d1f01df
1 parent 7355b82
commit d1f01df
Showing 1 changed file with 21 additions and 12 deletions.
diff --git a/meta-client/src/main/java/org/apache/helix/metaclient/impl/zk/ZkMetaClientCache.java b/meta-client/src/main/java/org/apache/helix/metaclient/impl/zk/ZkMetaClientCache.java
@@ -29,8 +29,11 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import java.util.ArrayDeque;
 import java.util.ArrayList;
 import java.util.List;
+import java.util.Queue;
+
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.ExecutorService;
@@ -111,18 +114,26 @@ public int countDirectChildren(final String key) {
 
     private void populateAllCache() {
         // TODO: Concurrently populate children and data cache.
-        if (_cacheData) {
-            try {
-                List<String> children = _cacheClient.getChildren(_rootEntry);
-                for (String child : children) {
-                    String childPath = _rootEntry + "/" + child;
-                    T dataRecord = _cacheClient.readData(childPath, true);
-                    getDataCacheMap().put(childPath, dataRecord);
-                }
-            } catch (Exception e) {
-                // Ignore
+        if (!_cacheClient.exists(_rootEntry)) {
+            LOG.warn("Root entry: {} does not exist.", _rootEntry);
+            // Let the other threads know that the cache is populated.
+            _initializedCache.countDown();
+            return;
+        }
+
+        Queue<String> queue = new ArrayDeque<>();
+        queue.add(_rootEntry);
+
+        while (!queue.isEmpty()) {
+            String node = queue.poll();
+            if (_cacheData) {
+                T dataRecord = _cacheClient.readData(node, true);
+                _dataCacheMap.put(node, dataRecord);
             }
+            queue.addAll(_cacheClient.getChildren(node));
         }
+        // Let the other threads know that the cache is populated.
+        _initializedCache.countDown();
     }
 
     private class CacheUpdateRunnable implements Runnable {
@@ -196,7 +207,5 @@ public void connect() {
         executor = Executors.newSingleThreadExecutor();
         _cacheClient.subscribePersistRecursiveListener(_rootEntry, new ChildListenerAdapter(_eventListener));
         populateAllCache();
-        // Notify the latch that cache is populated.
-        _initializedCache.countDown();
     }
 }