From ed630b9d67474990cf5575ae16edde2bee569742 Mon Sep 17 00:00:00 2001 From: Masahiro Sakamoto Date: Wed, 29 Jul 2020 20:02:25 +0900 Subject: [PATCH 1/3] Fix deadlock in pulsar proxy --- .../server/util/ZookeeperCacheLoader.java | 20 ++++++++++++++----- 1 file changed, 15 insertions(+), 5 deletions(-) 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..748f4258ef9cf 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 @@ -18,12 +18,16 @@ */ package org.apache.pulsar.proxy.server.util; +import io.netty.util.concurrent.DefaultThreadFactory; + import java.io.Closeable; import java.io.IOException; import java.util.ArrayList; import java.util.Collections; import java.util.List; import java.util.Set; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; import java.util.concurrent.TimeUnit; import org.apache.bookkeeper.common.util.OrderedScheduler; @@ -55,6 +59,8 @@ public class ZookeeperCacheLoader implements Closeable { private final OrderedScheduler orderedExecutor = OrderedScheduler.newSchedulerBuilder().numThreads(8) .name("pulsar-proxy-ordered-cache").build(); + private final ExecutorService brokerListUpdater = Executors + .newSingleThreadExecutor(new DefaultThreadFactory("pulsar-broker-list-updater")); public static final String LOADBALANCE_BROKERS_ROOT = "/loadbalance/brokers"; @@ -78,11 +84,14 @@ 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); - } + // Run in a separate thread to avoid deadlocks + brokerListUpdater.execute(() -> { + try { + updateBrokerList(brokerNodes); + } catch (Exception e) { + log.warn("Error updating broker info after broker list changed.", e); + } + }); }); // Do initial fetch of brokers list @@ -110,6 +119,7 @@ public void close() throws IOException { throw new IOException(e); } orderedExecutor.shutdown(); + brokerListUpdater.shutdownNow(); } private void updateBrokerList(Set brokerNodes) throws Exception { From f2c5475c10882fdadf687ad3322558bb5b8b2742 Mon Sep 17 00:00:00 2001 From: Masahiro Sakamoto Date: Thu, 30 Jul 2020 16:22:30 +0900 Subject: [PATCH 2/3] Update broker info asynchronously --- .../service/web/ZookeeperCacheLoader.java | 59 ++++++++++++---- pulsar-proxy/pom.xml | 6 ++ .../server/util/ZookeeperCacheLoader.java | 68 +++++++++++++------ 3 files changed, 99 insertions(+), 34 deletions(-) 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..f32c785173f2f 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 getAvailableBrokers() { @@ -133,13 +138,43 @@ public void close() throws IOException { orderedExecutor.shutdown(); } - private void updateBrokerList(Set brokerNodes) throws Exception { - List availableBrokers = new ArrayList<>(brokerNodes.size()); + private CompletableFuture updateBrokerList(Set brokerNodes) { + CompletableFuture future = new CompletableFuture<>(); + + if (brokerNodes.isEmpty()) { + availableBrokers = new ArrayList<>(); + future.complete(null); + return future; + } + + List>> 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 newAvailableBrokers = new ArrayList<>(brokerNodes.size()); + + for (CompletableFuture> loadReportFuture : loadReportFutureList) { + try { + LoadManagerReport loadReport = loadReportFuture.get().get(); + if (loadReport != null) { + newAvailableBrokers.add(loadReport); + } + } 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 @@ ${project.version} + + ${project.groupId} + pulsar-common + ${project.version} + + org.apache.commons commons-lang3 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 748f4258ef9cf..b2bdee34c64d3 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 @@ -18,19 +18,18 @@ */ package org.apache.pulsar.proxy.server.util; -import io.netty.util.concurrent.DefaultThreadFactory; - import java.io.Closeable; import java.io.IOException; import java.util.ArrayList; import java.util.Collections; import java.util.List; +import java.util.Optional; import java.util.Set; -import java.util.concurrent.ExecutorService; -import java.util.concurrent.Executors; +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; @@ -59,8 +58,6 @@ public class ZookeeperCacheLoader implements Closeable { private final OrderedScheduler orderedExecutor = OrderedScheduler.newSchedulerBuilder().numThreads(8) .name("pulsar-proxy-ordered-cache").build(); - private final ExecutorService brokerListUpdater = Executors - .newSingleThreadExecutor(new DefaultThreadFactory("pulsar-broker-list-updater")); public static final String LOADBALANCE_BROKERS_ROOT = "/loadbalance/brokers"; @@ -72,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(localZkCache) { @Override @@ -84,21 +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) -> { - // Run in a separate thread to avoid deadlocks - brokerListUpdater.execute(() -> { - 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); } } @@ -119,16 +114,45 @@ public void close() throws IOException { throw new IOException(e); } orderedExecutor.shutdown(); - brokerListUpdater.shutdownNow(); } - private void updateBrokerList(Set brokerNodes) throws Exception { - List availableBrokers = new ArrayList<>(brokerNodes.size()); + private CompletableFuture updateBrokerList(Set brokerNodes) { + CompletableFuture future = new CompletableFuture<>(); + + if (brokerNodes.isEmpty()) { + availableBrokers = new ArrayList<>(); + future.complete(null); + return future; + } + + List>> 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 newAvailableBrokers = new ArrayList<>(brokerNodes.size()); + + for (CompletableFuture> loadReportFuture : loadReportFutureList) { + try { + LoadManagerReport loadReport = loadReportFuture.get().get(); + if (loadReport != null) { + newAvailableBrokers.add(loadReport); + } + } 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); From 361512f30fdb2d003f0b17838c966f4e17f33877 Mon Sep 17 00:00:00 2001 From: Masahiro Sakamoto Date: Tue, 4 Aug 2020 21:17:12 +0900 Subject: [PATCH 3/3] Make loadReport Optional --- .../pulsar/discovery/service/web/ZookeeperCacheLoader.java | 6 +++--- .../pulsar/proxy/server/util/ZookeeperCacheLoader.java | 6 +++--- 2 files changed, 6 insertions(+), 6 deletions(-) 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 f32c785173f2f..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 @@ -157,9 +157,9 @@ private CompletableFuture updateBrokerList(Set brokerNodes) { for (CompletableFuture> loadReportFuture : loadReportFutureList) { try { - LoadManagerReport loadReport = loadReportFuture.get().get(); - if (loadReport != null) { - newAvailableBrokers.add(loadReport); + Optional loadReport = loadReportFuture.get(); + if (loadReport.isPresent()) { + newAvailableBrokers.add(loadReport.get()); } } catch (Exception e) { future.completeExceptionally(e); 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 b2bdee34c64d3..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 @@ -135,9 +135,9 @@ private CompletableFuture updateBrokerList(Set brokerNodes) { for (CompletableFuture> loadReportFuture : loadReportFutureList) { try { - LoadManagerReport loadReport = loadReportFuture.get().get(); - if (loadReport != null) { - newAvailableBrokers.add(loadReport); + Optional loadReport = loadReportFuture.get(); + if (loadReport.isPresent()) { + newAvailableBrokers.add(loadReport.get()); } } catch (Exception e) { future.completeExceptionally(e);