Skip to content

fix bug: https://github.com/apache/pulsar/issues/7706 #7774

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
xuesongxs wants to merge 1 commit into from
Closed

fix bug: https://github.com/apache/pulsar/issues/7706 #7774

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
71 changes: 15 additions & 56 deletions ...y-service/src/main/java/org/apache/pulsar/discovery/service/web/ZookeeperCacheLoader.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand All @@ -51,7 +48,6 @@ public class ZookeeperCacheLoader implements Closeable {
private final ZooKeeperCache localZkCache;
private final LocalZooKeeperConnectionService localZkConnectionSvc;

private final ZooKeeperDataCache<LoadManagerReport> brokerInfo;
private final ZooKeeperChildrenCache availableBrokersCache;

private volatile List<LoadManagerReport> availableBrokers;
Expand All @@ -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() {
Expand All @@ -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() {
Expand All @@ -102,25 +97,17 @@ public void setWatcher(ZooKeeperSessionWatcher watcher) {
}
});

this.brokerInfo = new ZooKeeperDataCache<LoadManagerReport>(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<LoadManagerReport> getAvailableBrokers() {
Expand All @@ -138,43 +125,15 @@ public void close() throws IOException {
orderedExecutor.shutdown();
}

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<>();
private void updateBrokerList(Set<String> brokerNodes) throws Exception {
List<LoadManagerReport> 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<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;
this.availableBrokers = availableBrokers;
}

private static final Logger log = LoggerFactory.getLogger(ZookeeperCacheLoader.class);
Expand Down
71 changes: 15 additions & 56 deletions pulsar-proxy/src/main/java/org/apache/pulsar/proxy/server/util/ZookeeperCacheLoader.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand All @@ -51,9 +48,7 @@ public class ZookeeperCacheLoader implements Closeable {

private final ZooKeeper zkClient;
private final ZooKeeperCache localZkCache;
private final ZooKeeperDataCache<LoadManagerReport> brokerInfo;
private final ZooKeeperChildrenCache availableBrokersCache;

private volatile List<LoadManagerReport> availableBrokers;

private final OrderedScheduler orderedExecutor = OrderedScheduler.newSchedulerBuilder().numThreads(8)
Expand All @@ -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<LoadManagerReport>(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());
}
}

Expand All @@ -116,43 +103,15 @@ public void close() throws IOException {
orderedExecutor.shutdown();
}

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<>();
private void updateBrokerList(Set<String> brokerNodes) throws Exception {
List<LoadManagerReport> 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<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;
this.availableBrokers = availableBrokers;
}

private static final Logger log = LoggerFactory.getLogger(ZookeeperCacheLoader.class);
Expand Down