You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by lh...@apache.org on 2026/06/22 12:03:43 UTC
(pulsar) 13/17: [fix][proxy] Avoid blocking the proxy IO thread on a cold broker cache (#26052)
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a commit to branch branch-4.0
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit e192e2b7de5d2c6f6da8ffda966be616ccdf2486
Author: Matteo Merli <mm...@apache.org>
AuthorDate: Thu Jun 18 04:20:44 2026 -0700
[fix][proxy] Avoid blocking the proxy IO thread on a cold broker cache (#26052)
(cherry picked from commit 1d7ae01479cc225b9f588c6727d6c5d2e713a0e3)
---
.../broker/resources/MetadataStoreCacheLoader.java | 45 +++++++------
.../resources/MetadataStoreCacheLoaderTest.java | 76 ++++++++++++++++++++++
2 files changed, 98 insertions(+), 23 deletions(-)
diff --git a/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/resources/MetadataStoreCacheLoader.java b/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/resources/MetadataStoreCacheLoader.java
index 29451148da44..7a0a41282824 100644
--- a/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/resources/MetadataStoreCacheLoader.java
+++ b/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/resources/MetadataStoreCacheLoader.java
@@ -25,7 +25,7 @@ import java.util.List;
import java.util.Optional;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.TimeUnit;
-import java.util.function.Supplier;
+import java.util.concurrent.atomic.AtomicBoolean;
import org.apache.bookkeeper.common.util.OrderedScheduler;
import org.apache.commons.collections4.CollectionUtils;
import org.apache.pulsar.common.util.FutureUtil;
@@ -46,6 +46,7 @@ public class MetadataStoreCacheLoader implements Closeable {
private volatile List<LoadManagerReport> availableBrokers;
private final FutureUtil.Sequencer<Void> sequencer;
+ private final AtomicBoolean refreshInProgress = new AtomicBoolean(false);
private final OrderedScheduler orderedExecutor = OrderedScheduler.newSchedulerBuilder().numThreads(8)
.name("pulsar-metadata-cache-loader-ordered-cache").build();
@@ -65,40 +66,38 @@ public class MetadataStoreCacheLoader implements Closeable {
* @throws Exception
*/
public void init() throws Exception {
- Supplier<CompletableFuture<Void>> tryUpdate = () -> {
- return loadReportResources.getChildrenAsync(LOADBALANCE_BROKERS_ROOT)
- .thenComposeAsync(brokerNodes -> {
- return 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;
- });
- };
loadReportResources.getStore().registerListener((n) -> {
if (LOADBALANCE_BROKERS_ROOT.equals(n.getPath()) && NotificationType.ChildrenChanged.equals(n.getType())) {
- sequencer.sequential(tryUpdate);
+ sequencer.sequential(this::reloadBrokers);
}
});
if (loadReportResources.getStore() instanceof MetadataStoreExtended) {
((MetadataStoreExtended) loadReportResources.getStore()).registerSessionListener(sessionEvent ->
- sequencer.sequential(tryUpdate));
+ sequencer.sequential(this::reloadBrokers));
}
// Do initial fetch of brokers list
- tryUpdate.get().get(operationTimeoutMs, TimeUnit.MILLISECONDS);
+ reloadBrokers().get(operationTimeoutMs, TimeUnit.MILLISECONDS);
+ }
+
+ private CompletableFuture<Void> reloadBrokers() {
+ return loadReportResources.getChildrenAsync(LOADBALANCE_BROKERS_ROOT)
+ .thenComposeAsync(brokerNodes -> 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;
+ });
}
public List<LoadManagerReport> getAvailableBrokers() {
- if (CollectionUtils.isEmpty(availableBrokers)) {
- try {
- updateBrokerList(loadReportResources.getChildren(LOADBALANCE_BROKERS_ROOT));
- } catch (Exception e) {
- log.warn("Error updating broker from zookeeper.", e);
- }
+ List<LoadManagerReport> brokers = availableBrokers;
+ if (CollectionUtils.isEmpty(brokers) && refreshInProgress.compareAndSet(false, true)) {
+ // Avoid blocking the caller (which may be a Netty IO thread): refresh the cache in the
+ // background and return the current snapshot. The cache is otherwise kept up to date by the
+ // metadata-store listener, so an empty snapshot means there are no active brokers.
+ sequencer.sequential(this::reloadBrokers).whenComplete((__, ex) -> refreshInProgress.set(false));
}
- return availableBrokers;
+ return brokers == null ? new ArrayList<>() : brokers;
}
@Override
diff --git a/pulsar-broker-common/src/test/java/org/apache/pulsar/broker/resources/MetadataStoreCacheLoaderTest.java b/pulsar-broker-common/src/test/java/org/apache/pulsar/broker/resources/MetadataStoreCacheLoaderTest.java
new file mode 100644
index 000000000000..e8d4d28153c9
--- /dev/null
+++ b/pulsar-broker-common/src/test/java/org/apache/pulsar/broker/resources/MetadataStoreCacheLoaderTest.java
@@ -0,0 +1,76 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.broker.resources;
+
+import static org.apache.pulsar.broker.resources.MetadataStoreCacheLoader.LOADBALANCE_BROKERS_ROOT;
+import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.assertTrue;
+import java.util.List;
+import java.util.Optional;
+import java.util.concurrent.CompletableFuture;
+import lombok.Cleanup;
+import org.apache.pulsar.metadata.api.MetadataStore;
+import org.apache.pulsar.policies.data.loadbalancer.LoadManagerReport;
+import org.testng.annotations.Test;
+
+public class MetadataStoreCacheLoaderTest {
+
+ private MetadataStoreCacheLoader newCacheLoader(LoadManagerReportResources loadReportResources) throws Exception {
+ MetadataStore store = mock(MetadataStore.class);
+ PulsarResources pulsarResources = mock(PulsarResources.class);
+ when(pulsarResources.getLoadReportResources()).thenReturn(loadReportResources);
+ when(loadReportResources.getStore()).thenReturn(store);
+ return new MetadataStoreCacheLoader(pulsarResources, 5000);
+ }
+
+ @Test
+ public void testGetAvailableBrokersServesCacheWithoutBlocking() throws Exception {
+ LoadManagerReportResources loadReportResources = mock(LoadManagerReportResources.class);
+ LoadManagerReport report = mock(LoadManagerReport.class);
+ when(loadReportResources.getChildrenAsync(LOADBALANCE_BROKERS_ROOT))
+ .thenReturn(CompletableFuture.completedFuture(List.of("broker-1")));
+ when(loadReportResources.getAsync(LOADBALANCE_BROKERS_ROOT + "/broker-1"))
+ .thenReturn(CompletableFuture.completedFuture(Optional.of(report)));
+
+ @Cleanup
+ MetadataStoreCacheLoader loader = newCacheLoader(loadReportResources);
+
+ assertEquals(loader.getAvailableBrokers(), List.of(report));
+ // The blocking, synchronous getChildren(...) must never be used: it could stall a Netty IO thread.
+ verify(loadReportResources, never()).getChildren(anyString());
+ }
+
+ @Test
+ public void testGetAvailableBrokersDoesNotBlockOnEmptyCache() throws Exception {
+ LoadManagerReportResources loadReportResources = mock(LoadManagerReportResources.class);
+ when(loadReportResources.getChildrenAsync(LOADBALANCE_BROKERS_ROOT))
+ .thenReturn(CompletableFuture.completedFuture(List.of()));
+
+ @Cleanup
+ MetadataStoreCacheLoader loader = newCacheLoader(loadReportResources);
+
+ assertTrue(loader.getAvailableBrokers().isEmpty());
+ verify(loadReportResources, never()).getChildren(anyString());
+ }
+}