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/10/22 20:22:44 UTC
(pulsar) 02/04: [fix][broker] Don't continue loading a topic after its load has timed out (#20540)
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a commit to branch branch-5.0
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit d8edc2bf8eb0882374a3d88833fa009871f8cfa7
Author: Lari Hotari <lh...@users.noreply.github.com>
AuthorDate: Thu Oct 22 22:46:20 2026 +0300
[fix][broker] Don't continue loading a topic after its load has timed out (#20540)
(cherry picked from commit aec5da342cb53fb05bff48f04ef6fb0eea6ae2b3)
---
.../pulsar/broker/service/BrokerService.java | 76 +++++++++++++-
.../pulsar/broker/service/BrokerServiceTest.java | 109 +++++++++++++++++++++
2 files changed, 182 insertions(+), 3 deletions(-)
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java
index 804405b90f1a..bc21c0ccd6a8 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java
@@ -381,6 +381,8 @@ public class BrokerService implements Closeable {
private final TopicEventsDispatcher topicEventsDispatcher = new TopicEventsDispatcher();
private volatile boolean unloaded = false;
+ // set when the broker starts shutting down, so that topic loads waiting for a permit are failed
+ private volatile boolean pendingTopicLoadsRejected = false;
// semaphore for limiting the concurrency of ledger deletion at broker level,
// thus all managed ledgers sharing the same semaphore
@@ -1237,6 +1239,9 @@ public class BrokerService implements Closeable {
if (unloaded) {
return;
}
+ // unloading a bundle waits for the topic futures cached for it, including the ones of the topic loads that
+ // wait for a permit, which could otherwise only complete when the topic load times out
+ rejectPendingTopicLoads();
try {
log.info("Unloading namespace-bundles...");
// make broker-node unavailable from the cluster
@@ -2227,6 +2232,10 @@ public class BrokerService implements Closeable {
} else {
pendingTopicLoadingQueue.add(ImmutablePair.of(context, queueTrace));
log.debug().attr("topic", topic).log("topic-loading for added into pending queue");
+ if (pendingTopicLoadsRejected) {
+ // the broker started shutting down concurrently, after the queue had been drained
+ failPendingTopicLoads();
+ }
}
}).exceptionally(ex -> {
failTopicFuture(topic, topicFuture, ex);
@@ -2424,6 +2433,17 @@ public class BrokerService implements Closeable {
// took the bundle over from, must not open, and so recover, the new owner's ledger.
context.trace(TopicLoadingTracePoints.OWNERSHIP_BEFORE_OPEN_ML, checkTopicNsOwnership(topic))
.thenRun(() -> {
+ if (topicFuture.isDone()) {
+ // The load has timed out: opening, and so recovering, the managed ledger now would only be
+ // followed by closing it again. The timed-out future stays cached until the pending load
+ // actions finish, so remove it here to let a new load start.
+ log.debug()
+ .attr("topic", topic)
+ .attr("latency", context.getSnapshot().description())
+ .log("Skip opening the managed ledger since the topic load has already completed");
+ pulsar.getExecutor().execute(() -> topics.remove(topic, topicFuture));
+ return;
+ }
final LatencyTracer.TracePoint openMlTracePoint = context.startTrace(TopicLoadingTracePoints.OPEN_ML);
managedLedgerFactory.asyncOpen(topicName.getPersistenceNamingEncoding(), managedLedgerConfig,
new OpenLedgerCallback() {
@@ -4154,14 +4174,16 @@ public class BrokerService implements Closeable {
* permit if it was successful to acquire it.
*/
private void createPendingLoadTopic() {
- Pair<TopicLoadingContext, LatencyTracer.TracePoint> contextAndPointPair = pendingTopicLoadingQueue.poll();
+ if (pendingTopicLoadsRejected) {
+ failPendingTopicLoads();
+ return;
+ }
+ Pair<TopicLoadingContext, LatencyTracer.TracePoint> contextAndPointPair = pollPendingTopicLoad();
if (contextAndPointPair == null) {
return;
}
TopicLoadingContext topicLoadingContext = contextAndPointPair.getLeft();
- LatencyTracer.TracePoint queuedPoint = contextAndPointPair.getRight();
- topicLoadingContext.finishTrace(queuedPoint, null);
final String topic = topicLoadingContext.getTopicName().toString();
// ServiceUnitNotReadyException is classified as bundle_unloading when the topic future completes.
topicLoadingContext.trace(TopicLoadingTracePoints.OWNERSHIP, checkTopicNsOwnership(topic)).thenRun(() -> {
@@ -4187,6 +4209,54 @@ public class BrokerService implements Closeable {
});
}
+ /**
+ * Polls the next pending topic load whose topic future is not yet completed. A load that timed out while
+ * it was queued is dropped instead of being started: its timed-out future would otherwise stay cached until
+ * the managed ledger had been opened and closed again.
+ */
+ private Pair<TopicLoadingContext, LatencyTracer.TracePoint> pollPendingTopicLoad() {
+ Pair<TopicLoadingContext, LatencyTracer.TracePoint> contextAndPointPair;
+ while ((contextAndPointPair = pendingTopicLoadingQueue.poll()) != null) {
+ TopicLoadingContext topicLoadingContext = contextAndPointPair.getLeft();
+ topicLoadingContext.finishTrace(contextAndPointPair.getRight(), null);
+ CompletableFuture<Optional<Topic>> topicFuture = topicLoadingContext.getTopicFuture();
+ if (!topicFuture.isDone()) {
+ return contextAndPointPair;
+ }
+ String topic = topicLoadingContext.getTopicName().toString();
+ log.debug()
+ .attr("topic", topic)
+ .log("Skip loading the pending topic since its topic load has already completed");
+ pulsar.getExecutor().execute(() -> topics.remove(topic, topicFuture));
+ }
+ return null;
+ }
+
+ /**
+ * Fails the pending topic loads, and the ones queued later, when the broker is shutting down, so that they
+ * don't wait for the topic load timeout and clients can look the topics up again.
+ */
+ private void rejectPendingTopicLoads() {
+ pendingTopicLoadsRejected = true;
+ failPendingTopicLoads();
+ }
+
+ private void failPendingTopicLoads() {
+ Pair<TopicLoadingContext, LatencyTracer.TracePoint> contextAndPointPair;
+ while ((contextAndPointPair = pendingTopicLoadingQueue.poll()) != null) {
+ TopicLoadingContext topicLoadingContext = contextAndPointPair.getLeft();
+ topicLoadingContext.finishTrace(contextAndPointPair.getRight(), null);
+ String topic = topicLoadingContext.getTopicName().toString();
+ CompletableFuture<Optional<Topic>> topicFuture = topicLoadingContext.getTopicFuture();
+ topicFuture.completeExceptionally(new ServiceUnitNotReadyException("Broker is shutting down"));
+ try {
+ pulsar.getExecutor().execute(() -> topics.remove(topic, topicFuture));
+ } catch (RejectedExecutionException e) {
+ // the broker has already been closed
+ }
+ }
+ }
+
public CompletableFuture<PartitionedTopicMetadata> fetchPartitionedTopicMetadataCheckAllowAutoCreationAsync(
TopicName topicName) {
return fetchPartitionedTopicMetadataCheckAllowAutoCreationAsync(topicName, ALLOW_NEW_TOPIC);
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/BrokerServiceTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/BrokerServiceTest.java
index b5be3d059553..26f1f07c8e45 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/BrokerServiceTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/BrokerServiceTest.java
@@ -24,11 +24,16 @@ import static org.apache.pulsar.common.naming.SystemTopicNames.TRANSACTION_COORD
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.ArgumentMatchers.argThat;
+import static org.mockito.ArgumentMatchers.eq;
import static org.mockito.Mockito.any;
+import static org.mockito.Mockito.doAnswer;
import static org.mockito.Mockito.doNothing;
import static org.mockito.Mockito.doReturn;
import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
import static org.mockito.Mockito.spy;
+import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
import static org.testng.Assert.assertEquals;
import static org.testng.Assert.assertFalse;
@@ -67,6 +72,7 @@ import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
+import java.util.concurrent.Semaphore;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicBoolean;
@@ -77,6 +83,7 @@ import org.apache.bookkeeper.client.api.ReadHandle;
import org.apache.bookkeeper.mledger.LedgerOffloader;
import org.apache.bookkeeper.mledger.ManagedLedgerConfig;
import org.apache.bookkeeper.mledger.ManagedLedgerException;
+import org.apache.bookkeeper.mledger.ManagedLedgerFactory;
import org.apache.bookkeeper.mledger.impl.ManagedCursorImpl;
import org.apache.bookkeeper.mledger.impl.ManagedLedgerFactoryImpl;
import org.apache.bookkeeper.mledger.impl.ManagedLedgerImpl;
@@ -289,6 +296,108 @@ public class BrokerServiceTest extends BrokerTestBase {
.containsKey(topicName.getPersistenceNamingEncoding()));
}
+ @Test
+ public void testTimedOutTopicLoadDoesNotOpenTheManagedLedger() throws Exception {
+ TopicName topicName = TopicName.get("persistent://prop/ns-abc/timed-out-load-" + UUID.randomUUID());
+ BrokerService brokerService = spy(pulsar.getBrokerService());
+ CompletableFuture<Void> ownershipCheckStarted = new CompletableFuture<>();
+ CompletableFuture<Void> ownershipCheck = new CompletableFuture<>();
+ doAnswer(invocation -> {
+ ownershipCheckStarted.complete(null);
+ return ownershipCheck;
+ }).when(brokerService).checkTopicNsOwnership(topicName.toString());
+ ManagedLedgerFactory managedLedgerFactory = mock(ManagedLedgerFactory.class);
+ doReturn(managedLedgerFactory).when(brokerService).getManagedLedgerFactoryForTopic(eq(topicName), any());
+ TopicLoadingContext context = new TopicLoadingContext(topicName, true, new CompletableFuture<>(),
+ mock(PulsarStats.class));
+ CompletableFuture<Optional<Topic>> topicFuture = context.getTopicFuture();
+ brokerService.getTopics().put(topicName.toString(), topicFuture);
+
+ brokerService.createPersistentTopic0(context);
+ ownershipCheckStarted.get(10, TimeUnit.SECONDS);
+ // the load times out while it waits for the ownership check before opening the managed ledger
+ topicFuture.completeExceptionally(new TimeoutException("Failed to load topic within timeout"));
+ ownershipCheck.complete(null);
+
+ // the timed-out future no longer blocks a new load of the topic
+ Awaitility.await().untilAsserted(() -> assertNull(brokerService.getTopics().get(topicName.toString())));
+ verify(managedLedgerFactory, never()).asyncOpen(anyString(), any(), any(), any(), any());
+ }
+
+ @Test
+ public void testPendingTopicLoadThatTimedOutIsNotStarted() throws Exception {
+ BrokerService brokerService = spy(pulsar.getBrokerService());
+ Semaphore topicLoadSemaphore = brokerService.topicLoadRequestSemaphore.get();
+ TopicName queuedTopicName = TopicName.get("persistent://prop/ns-abc/queued-load-" + UUID.randomUUID());
+ TopicName otherTopicName = TopicName.get("persistent://prop/ns-abc/other-load-" + UUID.randomUUID());
+ TopicLoadingContext queuedContext = new TopicLoadingContext(queuedTopicName, true,
+ new CompletableFuture<>(), mock(PulsarStats.class));
+ TopicLoadingContext otherContext = new TopicLoadingContext(otherTopicName, true,
+ new CompletableFuture<>(), mock(PulsarStats.class));
+ CompletableFuture<Optional<Topic>> queuedFuture = queuedContext.getTopicFuture();
+ brokerService.getTopics().put(queuedTopicName.toString(), queuedFuture);
+ brokerService.getTopics().put(otherTopicName.toString(), otherContext.getTopicFuture());
+ doReturn(CompletableFuture.completedFuture(null)).when(brokerService)
+ .checkTopicNsOwnership(queuedTopicName.toString());
+ // acquire the ownership of the bundle so that the other topic can be loaded
+ admin.lookups().lookupTopic(otherTopicName.toString());
+ try {
+ // no permits are left, so the load is queued
+ brokerService.topicLoadRequestSemaphore.set(new Semaphore(0));
+ brokerService.loadOrCreatePersistentTopic(queuedContext);
+ // the queued load times out while it waits for a permit
+ queuedFuture.completeExceptionally(new TimeoutException("Failed to load topic within timeout"));
+
+ // the next load to finish processes the queue
+ brokerService.topicLoadRequestSemaphore.set(new Semaphore(1));
+ brokerService.loadOrCreatePersistentTopic(otherContext).get(10, TimeUnit.SECONDS);
+
+ Awaitility.await().untilAsserted(
+ () -> assertNull(brokerService.getTopics().get(queuedTopicName.toString())));
+ verify(brokerService, never()).createPersistentTopic0(argThat(c -> c == queuedContext));
+ admin.topics().delete(otherTopicName.toString(), true);
+ } finally {
+ brokerService.topicLoadRequestSemaphore.set(topicLoadSemaphore);
+ }
+ }
+
+ @Test
+ public void testPendingTopicLoadsFailWhenTheBrokerShutsDown() throws Exception {
+ BrokerService brokerService = spy(pulsar.getBrokerService());
+ Semaphore topicLoadSemaphore = brokerService.topicLoadRequestSemaphore.get();
+ long namespaceBundleUnloadingTimeoutMs = conf.getNamespaceBundleUnloadingTimeoutMs();
+ TopicName topicName = TopicName.get("persistent://prop/ns-abc/shutdown-load-" + UUID.randomUUID());
+ TopicLoadingContext context = new TopicLoadingContext(topicName, true, new CompletableFuture<>(),
+ mock(PulsarStats.class));
+ CompletableFuture<Optional<Topic>> topicFuture = context.getTopicFuture();
+ brokerService.getTopics().put(topicName.toString(), topicFuture);
+ doReturn(CompletableFuture.completedFuture(null)).when(brokerService)
+ .checkTopicNsOwnership(topicName.toString());
+ // acquire the ownership of the bundle so that the graceful unload unloads it
+ admin.lookups().lookupTopic(topicName.toString());
+ try {
+ // no permits are left, so the load is queued
+ brokerService.topicLoadRequestSemaphore.set(new Semaphore(0));
+ brokerService.loadOrCreatePersistentTopic(context);
+ assertFalse(topicFuture.isDone());
+
+ // without failing the queued load first, unloading the bundle would wait for this timeout
+ conf.setNamespaceBundleUnloadingTimeoutMs(TimeUnit.SECONDS.toMillis(60));
+ long startNanos = System.nanoTime();
+ pulsar.getBrokerService().unloadNamespaceBundlesGracefully();
+ assertThat(System.nanoTime() - startNanos).isLessThan(TimeUnit.SECONDS.toNanos(30));
+
+ assertTrue(topicFuture.isDone());
+ assertThatThrownBy(topicFuture::get)
+ .hasCauseInstanceOf(BrokerServiceException.ServiceUnitNotReadyException.class)
+ .hasMessageContaining("Broker is shutting down");
+ } finally {
+ brokerService.topicLoadRequestSemaphore.set(topicLoadSemaphore);
+ conf.setNamespaceBundleUnloadingTimeoutMs(namespaceBundleUnloadingTimeoutMs);
+ resetState();
+ }
+ }
+
@Test
public void testNonStageTraceIsPending() {
TopicLoadingContext context = new TopicLoadingContext(TopicName.get("persistent://public/default/test"), true,