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 e1a2cd7927bdc..85f440f32ecc3 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,13 +22,10 @@ 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; @@ -51,7 +48,6 @@ public class ZookeeperCacheLoader implements Closeable { private final ZooKeeperCache localZkCache; private final LocalZooKeeperConnectionService localZkConnectionSvc; - private final ZooKeeperDataCache brokerInfo; private final ZooKeeperChildrenCache availableBrokersCache; private volatile List availableBrokers; @@ -68,7 +64,7 @@ public class ZookeeperCacheLoader implements Closeable { * @throws Exception */ public ZookeeperCacheLoader(ZooKeeperClientFactory zkClientFactory, String zookeeperServers, - int zookeeperSessionTimeoutMs) throws Exception { + int zookeeperSessionTimeoutMs) throws Exception { localZkConnectionSvc = new LocalZooKeeperConnectionService(zkClientFactory, zookeeperServers, zookeeperSessionTimeoutMs); localZkConnectionSvc.start(new ZookeeperSessionExpiredHandler() { @@ -83,9 +79,8 @@ public void setWatcher(ZooKeeperSessionWatcher watcher) { } }); - int zkOperationTimeoutSeconds = (int) TimeUnit.MILLISECONDS.toSeconds(zookeeperSessionTimeoutMs); - this.localZkCache = new LocalZooKeeperCache(localZkConnectionSvc.getLocalZooKeeper(), zkOperationTimeoutSeconds, - this.orderedExecutor); + this.localZkCache = new LocalZooKeeperCache(localZkConnectionSvc.getLocalZooKeeper(), + (int) TimeUnit.MILLISECONDS.toSeconds(zookeeperSessionTimeoutMs), this.orderedExecutor); localZkConnectionSvc.start(new ZookeeperSessionExpiredHandler() { @Override public void onSessionExpired() { @@ -102,25 +97,17 @@ public void setWatcher(ZooKeeperSessionWatcher watcher) { } }); - this.brokerInfo = new ZooKeeperDataCache(localZkCache) { - @Override - public LoadManagerReport deserialize(String key, byte[] content) throws Exception { - return ObjectMapperFactory.getThreadLocal().readValue(content, LoadManagerReport.class); - } - }; - this.availableBrokersCache = new ZooKeeperChildrenCache(getLocalZkCache(), LOADBALANCE_BROKERS_ROOT); this.availableBrokersCache.registerListener((path, brokerNodes, stat) -> { - 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; - }); + try { + updateBrokerList(brokerNodes); + } catch (Exception e) { + log.warn("Error updating broker info after broker list changed.", e); + } }); // Do initial fetch of brokers list - updateBrokerList(availableBrokersCache.get()).get(zkOperationTimeoutSeconds, TimeUnit.SECONDS); + updateBrokerList(availableBrokersCache.get()); } public List getAvailableBrokers() { @@ -138,43 +125,15 @@ public void close() throws IOException { orderedExecutor.shutdown(); } - private CompletableFuture updateBrokerList(Set brokerNodes) { - CompletableFuture future = new CompletableFuture<>(); - - if (brokerNodes.isEmpty()) { - availableBrokers = new ArrayList<>(); - future.complete(null); - return future; - } - - List>> loadReportFutureList = new ArrayList<>(); + private void updateBrokerList(Set brokerNodes) throws Exception { + List availableBrokers = new ArrayList<>(brokerNodes.size()); for (String broker : brokerNodes) { - loadReportFutureList.add(brokerInfo.getAsync(LOADBALANCE_BROKERS_ROOT + '/' + broker)); + byte[] data = getLocalZkCache().getZooKeeper().getData(LOADBALANCE_BROKERS_ROOT + '/' + broker, false, null); + LoadManagerReport loadManagerReport = ObjectMapperFactory.getThreadLocal().readValue(data, LoadManagerReport.class); + availableBrokers.add(loadManagerReport); } - FutureUtil.waitForAll(loadReportFutureList).thenRun(() -> { - List newAvailableBrokers = new ArrayList<>(brokerNodes.size()); - - for (CompletableFuture> loadReportFuture : loadReportFutureList) { - try { - Optional 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; + this.availableBrokers = availableBrokers; } private static final Logger log = LoggerFactory.getLogger(ZookeeperCacheLoader.class); 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 7280ec896cdc6..30176b3d9f5f4 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,13 +23,10 @@ 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; @@ -51,9 +48,7 @@ public class ZookeeperCacheLoader implements Closeable { private final ZooKeeper zkClient; private final ZooKeeperCache localZkCache; - private final ZooKeeperDataCache brokerInfo; private final ZooKeeperChildrenCache availableBrokersCache; - private volatile List availableBrokers; private final OrderedScheduler orderedExecutor = OrderedScheduler.newSchedulerBuilder().numThreads(8) @@ -69,31 +64,23 @@ public class ZookeeperCacheLoader implements Closeable { */ public ZookeeperCacheLoader(ZooKeeperClientFactory factory, String zookeeperServers, int zookeeperSessionTimeoutMs) throws Exception { this.zkClient = factory.create(zookeeperServers, SessionType.AllowReadOnly, zookeeperSessionTimeoutMs).get(); - int zkOperationTimeoutSeconds = (int) TimeUnit.MILLISECONDS.toSeconds(zookeeperSessionTimeoutMs); - this.localZkCache = new LocalZooKeeperCache(zkClient, zkOperationTimeoutSeconds, this.orderedExecutor); - - this.brokerInfo = new ZooKeeperDataCache(localZkCache) { - @Override - public LoadManagerReport deserialize(String key, byte[] content) throws Exception { - return ObjectMapperFactory.getThreadLocal().readValue(content, LoadManagerReport.class); - } - }; + this.localZkCache = new LocalZooKeeperCache(zkClient, + (int) TimeUnit.MILLISECONDS.toSeconds(zookeeperSessionTimeoutMs), this.orderedExecutor); this.availableBrokersCache = new ZooKeeperChildrenCache(getLocalZkCache(), LOADBALANCE_BROKERS_ROOT); this.availableBrokersCache.registerListener((path, brokerNodes, stat) -> { - 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; - }); + try { + updateBrokerList(brokerNodes); + } catch (Exception e) { + log.warn("Error updating broker info after broker list changed.", e); + } }); // Do initial fetch of brokers list try { - updateBrokerList(availableBrokersCache.get()).get(zkOperationTimeoutSeconds, TimeUnit.SECONDS); + updateBrokerList(availableBrokersCache.get()); } catch (NoNodeException nne) { // can happen if no broker started yet - updateBrokerList(Collections.emptySet()).get(zkOperationTimeoutSeconds, TimeUnit.SECONDS); + updateBrokerList(Collections.emptySet()); } } @@ -116,43 +103,15 @@ public void close() throws IOException { orderedExecutor.shutdown(); } - private CompletableFuture updateBrokerList(Set brokerNodes) { - CompletableFuture future = new CompletableFuture<>(); - - if (brokerNodes.isEmpty()) { - availableBrokers = new ArrayList<>(); - future.complete(null); - return future; - } - - List>> loadReportFutureList = new ArrayList<>(); + private void updateBrokerList(Set brokerNodes) throws Exception { + List availableBrokers = new ArrayList<>(brokerNodes.size()); for (String broker : brokerNodes) { - loadReportFutureList.add(brokerInfo.getAsync(LOADBALANCE_BROKERS_ROOT + '/' + broker)); + byte[] data = getLocalZkCache().getZooKeeper().getData(LOADBALANCE_BROKERS_ROOT + '/' + broker, false, null); + LoadManagerReport loadManagerReport = ObjectMapperFactory.getThreadLocal().readValue(data, LoadManagerReport.class); + availableBrokers.add(loadManagerReport); } - FutureUtil.waitForAll(loadReportFutureList).thenRun(() -> { - List newAvailableBrokers = new ArrayList<>(brokerNodes.size()); - - for (CompletableFuture> loadReportFuture : loadReportFutureList) { - try { - Optional 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; + this.availableBrokers = availableBrokers; } private static final Logger log = LoggerFactory.getLogger(ZookeeperCacheLoader.class);