diff --git a/pulsar-discovery-service/src/main/java/org/apache/pulsar/discovery/service/web/ZookeeperCacheLoader.java b/pulsar-discovery-service/src/main/java/org/apache/pulsar/discovery/service/web/ZookeeperCacheLoader.java
index 56332e7316fd4..e1a2cd7927bdc 100644
--- a/pulsar-discovery-service/src/main/java/org/apache/pulsar/discovery/service/web/ZookeeperCacheLoader.java
+++ b/pulsar-discovery-service/src/main/java/org/apache/pulsar/discovery/service/web/ZookeeperCacheLoader.java
@@ -22,10 +22,13 @@
 import java.io.IOException;
 import java.util.ArrayList;
 import java.util.List;
+import java.util.Optional;
 import java.util.Set;
+import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.TimeUnit;
 
 import org.apache.bookkeeper.common.util.OrderedScheduler;
+import org.apache.pulsar.common.util.FutureUtil;
 import org.apache.pulsar.common.util.ObjectMapperFactory;
 import org.apache.pulsar.policies.data.loadbalancer.LoadManagerReport;
 import org.apache.pulsar.zookeeper.LocalZooKeeperCache;
@@ -80,8 +83,9 @@ public void setWatcher(ZooKeeperSessionWatcher watcher) {
             }
         });
 
-        this.localZkCache = new LocalZooKeeperCache(localZkConnectionSvc.getLocalZooKeeper(),
-                (int) TimeUnit.MILLISECONDS.toSeconds(zookeeperSessionTimeoutMs), this.orderedExecutor);
+        int zkOperationTimeoutSeconds = (int) TimeUnit.MILLISECONDS.toSeconds(zookeeperSessionTimeoutMs);
+        this.localZkCache = new LocalZooKeeperCache(localZkConnectionSvc.getLocalZooKeeper(), zkOperationTimeoutSeconds,
+                this.orderedExecutor);
         localZkConnectionSvc.start(new ZookeeperSessionExpiredHandler() {
             @Override
             public void onSessionExpired() {
@@ -107,15 +111,16 @@ public LoadManagerReport deserialize(String key, byte[] content) throws Exceptio
 
         this.availableBrokersCache = new ZooKeeperChildrenCache(getLocalZkCache(), LOADBALANCE_BROKERS_ROOT);
         this.availableBrokersCache.registerListener((path, brokerNodes, stat) -> {
-            try {
-                updateBrokerList(brokerNodes);
-            } catch (Exception e) {
-                log.warn("Error updating broker info after broker list changed.", e);
-            }
+            updateBrokerList(brokerNodes).thenRun(() -> {
+                log.info("Successfully updated broker info {}", brokerNodes);
+            }).exceptionally(ex -> {
+                log.warn("Error updating broker info after broker list changed", ex);
+                return null;
+            });
         });
 
         // Do initial fetch of brokers list
-        updateBrokerList(availableBrokersCache.get());
+        updateBrokerList(availableBrokersCache.get()).get(zkOperationTimeoutSeconds, TimeUnit.SECONDS);
     }
 
     public List<LoadManagerReport> getAvailableBrokers() {
@@ -133,13 +138,43 @@ public void close() throws IOException {
         orderedExecutor.shutdown();
     }
 
-    private void updateBrokerList(Set<String> brokerNodes) throws Exception {
-        List<LoadManagerReport> availableBrokers = new ArrayList<>(brokerNodes.size());
+    private CompletableFuture<Void> updateBrokerList(Set<String> brokerNodes) {
+        CompletableFuture<Void> future = new CompletableFuture<>();
+
+        if (brokerNodes.isEmpty()) {
+            availableBrokers = new ArrayList<>();
+            future.complete(null);
+            return future;
+        }
+
+        List<CompletableFuture<Optional<LoadManagerReport>>> loadReportFutureList = new ArrayList<>();
         for (String broker : brokerNodes) {
-            availableBrokers.add(brokerInfo.get(LOADBALANCE_BROKERS_ROOT + '/' + broker).get());
+            loadReportFutureList.add(brokerInfo.getAsync(LOADBALANCE_BROKERS_ROOT + '/' + broker));
         }
 
-        this.availableBrokers = availableBrokers;
+        FutureUtil.waitForAll(loadReportFutureList).thenRun(() -> {
+            List<LoadManagerReport> newAvailableBrokers = new ArrayList<>(brokerNodes.size());
+
+            for (CompletableFuture<Optional<LoadManagerReport>> loadReportFuture : loadReportFutureList) {
+                try {
+                    Optional<LoadManagerReport> loadReport = loadReportFuture.get();
+                    if (loadReport.isPresent()) {
+                        newAvailableBrokers.add(loadReport.get());
+                    }
+                } catch (Exception e) {
+                    future.completeExceptionally(e);
+                    return;
+                }
+            }
+
+            availableBrokers = newAvailableBrokers;
+            future.complete(null);
+        }).exceptionally(ex -> {
+            future.completeExceptionally(ex);
+            return null;
+        });
+
+        return future;
     }
 
     private static final Logger log = LoggerFactory.getLogger(ZookeeperCacheLoader.class);
diff --git a/pulsar-proxy/pom.xml b/pulsar-proxy/pom.xml
index 3800ca20ca3a2..c5f4ecb216669 100644
--- a/pulsar-proxy/pom.xml
+++ b/pulsar-proxy/pom.xml
@@ -49,6 +49,12 @@
       <version>${project.version}</version>
     </dependency>
 
+    <dependency>
+      <groupId>${project.groupId}</groupId>
+      <artifactId>pulsar-common</artifactId>
+      <version>${project.version}</version>
+    </dependency>
+
     <dependency>
       <groupId>org.apache.commons</groupId>
       <artifactId>commons-lang3</artifactId>
diff --git a/pulsar-proxy/src/main/java/org/apache/pulsar/proxy/server/util/ZookeeperCacheLoader.java b/pulsar-proxy/src/main/java/org/apache/pulsar/proxy/server/util/ZookeeperCacheLoader.java
index 82dd42c5e19f6..7280ec896cdc6 100644
--- a/pulsar-proxy/src/main/java/org/apache/pulsar/proxy/server/util/ZookeeperCacheLoader.java
+++ b/pulsar-proxy/src/main/java/org/apache/pulsar/proxy/server/util/ZookeeperCacheLoader.java
@@ -23,10 +23,13 @@
 import java.util.ArrayList;
 import java.util.Collections;
 import java.util.List;
+import java.util.Optional;
 import java.util.Set;
+import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.TimeUnit;
 
 import org.apache.bookkeeper.common.util.OrderedScheduler;
+import org.apache.pulsar.common.util.FutureUtil;
 import org.apache.pulsar.common.util.ObjectMapperFactory;
 import org.apache.pulsar.policies.data.loadbalancer.LoadManagerReport;
 import org.apache.pulsar.zookeeper.LocalZooKeeperCache;
@@ -66,8 +69,8 @@ public class ZookeeperCacheLoader implements Closeable {
      */
     public ZookeeperCacheLoader(ZooKeeperClientFactory factory, String zookeeperServers, int zookeeperSessionTimeoutMs) throws Exception {
         this.zkClient = factory.create(zookeeperServers, SessionType.AllowReadOnly, zookeeperSessionTimeoutMs).get();
-        this.localZkCache = new LocalZooKeeperCache(zkClient,
-                (int) TimeUnit.MILLISECONDS.toSeconds(zookeeperSessionTimeoutMs), this.orderedExecutor);
+        int zkOperationTimeoutSeconds = (int) TimeUnit.MILLISECONDS.toSeconds(zookeeperSessionTimeoutMs);
+        this.localZkCache = new LocalZooKeeperCache(zkClient, zkOperationTimeoutSeconds, this.orderedExecutor);
 
         this.brokerInfo = new ZooKeeperDataCache<LoadManagerReport>(localZkCache) {
             @Override
@@ -78,18 +81,19 @@ public LoadManagerReport deserialize(String key, byte[] content) throws Exceptio
 
         this.availableBrokersCache = new ZooKeeperChildrenCache(getLocalZkCache(), LOADBALANCE_BROKERS_ROOT);
         this.availableBrokersCache.registerListener((path, brokerNodes, stat) -> {
-            try {
-                updateBrokerList(brokerNodes);
-            } catch (Exception e) {
-                log.warn("Error updating broker info after broker list changed.", e);
-            }
+            updateBrokerList(brokerNodes).thenRun(() -> {
+                log.info("Successfully updated broker info {}", brokerNodes);
+            }).exceptionally(ex -> {
+                log.warn("Error updating broker info after broker list changed", ex);
+                return null;
+            });
         });
 
         // Do initial fetch of brokers list
         try {
-            updateBrokerList(availableBrokersCache.get());
+            updateBrokerList(availableBrokersCache.get()).get(zkOperationTimeoutSeconds, TimeUnit.SECONDS);
         } catch (NoNodeException nne) { // can happen if no broker started yet
-            updateBrokerList(Collections.emptySet());
+            updateBrokerList(Collections.emptySet()).get(zkOperationTimeoutSeconds, TimeUnit.SECONDS);
         }
     }
 
@@ -112,13 +116,43 @@ public void close() throws IOException {
         orderedExecutor.shutdown();
     }
 
-    private void updateBrokerList(Set<String> brokerNodes) throws Exception {
-        List<LoadManagerReport> availableBrokers = new ArrayList<>(brokerNodes.size());
+    private CompletableFuture<Void> updateBrokerList(Set<String> brokerNodes) {
+        CompletableFuture<Void> future = new CompletableFuture<>();
+
+        if (brokerNodes.isEmpty()) {
+            availableBrokers = new ArrayList<>();
+            future.complete(null);
+            return future;
+        }
+
+        List<CompletableFuture<Optional<LoadManagerReport>>> loadReportFutureList = new ArrayList<>();
         for (String broker : brokerNodes) {
-            availableBrokers.add(brokerInfo.get(LOADBALANCE_BROKERS_ROOT + '/' + broker).get());
+            loadReportFutureList.add(brokerInfo.getAsync(LOADBALANCE_BROKERS_ROOT + '/' + broker));
         }
 
-        this.availableBrokers = availableBrokers;
+        FutureUtil.waitForAll(loadReportFutureList).thenRun(() -> {
+            List<LoadManagerReport> newAvailableBrokers = new ArrayList<>(brokerNodes.size());
+
+            for (CompletableFuture<Optional<LoadManagerReport>> loadReportFuture : loadReportFutureList) {
+                try {
+                    Optional<LoadManagerReport> loadReport = loadReportFuture.get();
+                    if (loadReport.isPresent()) {
+                        newAvailableBrokers.add(loadReport.get());
+                    }
+                } catch (Exception e) {
+                    future.completeExceptionally(e);
+                    return;
+                }
+            }
+
+            availableBrokers = newAvailableBrokers;
+            future.complete(null);
+        }).exceptionally(ex -> {
+            future.completeExceptionally(ex);
+            return null;
+        });
+
+        return future;
     }
 
     private static final Logger log = LoggerFactory.getLogger(ZookeeperCacheLoader.class);