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:34 UTC
(pulsar) 04/17: [fix][broker] Don't let a stuck or aborted topic policies cache init make a namespace's topics unloadable (#26025)
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 d42ac7a2d8d09ea4d47450293724700ffa0e2d63
Author: Lari Hotari <lh...@users.noreply.github.com>
AuthorDate: Sun Jun 14 16:22:58 2026 +0300
[fix][broker] Don't let a stuck or aborted topic policies cache init make a namespace's topics unloadable (#26025)
(cherry picked from commit 24f2270ec21906608429d05f86ec7347e9bf0edf)
---
conf/broker.conf | 7 ++
conf/standalone.conf | 7 ++
.../apache/pulsar/broker/ServiceConfiguration.java | 13 ++
.../SystemTopicBasedTopicPoliciesService.java | 135 ++++++++++++++++++++-
.../SystemTopicBasedTopicPoliciesServiceTest.java | 121 ++++++++++++++++--
5 files changed, 271 insertions(+), 12 deletions(-)
diff --git a/conf/broker.conf b/conf/broker.conf
index 503a7a0e11f5..72a5f50876fa 100644
--- a/conf/broker.conf
+++ b/conf/broker.conf
@@ -769,6 +769,13 @@ systemTopicSchemaCompatibilityStrategy=ALWAYS_COMPATIBLE
# Please enable the system topic first.
topicLevelPoliciesEnabled=true
+# Amount of seconds to timeout initializing the topic policies cache of a namespace (reading the namespace's
+# __change_events system topic to the end). Topic loading waits for this initialization, so if the system-topic
+# reader gets stuck, this bounds the wait: the broker fails the initialization, closes the stuck reader and clears
+# the cached state so loading the namespace's topics can be retried with a fresh reader instead of hanging until the
+# broker is restarted. Set to 0 or a negative value to disable the timeout (not recommended).
+topicPoliciesCacheInitTimeoutSeconds=60
+
# If a topic remains fenced for this number of seconds, it will be closed forcefully.
# If it is set to 0 or a negative number, the fenced topic will not be closed.
topicFencingTimeoutSeconds=0
diff --git a/conf/standalone.conf b/conf/standalone.conf
index f9c7ccb658f3..59fcbdf2aabc 100644
--- a/conf/standalone.conf
+++ b/conf/standalone.conf
@@ -579,6 +579,13 @@ systemTopicSchemaCompatibilityStrategy=ALWAYS_COMPATIBLE
# Please enable the system topic first.
topicLevelPoliciesEnabled=true
+# Amount of seconds to timeout initializing the topic policies cache of a namespace (reading the namespace's
+# __change_events system topic to the end). Topic loading waits for this initialization, so if the system-topic
+# reader gets stuck, this bounds the wait: the broker fails the initialization, closes the stuck reader and clears
+# the cached state so loading the namespace's topics can be retried with a fresh reader instead of hanging until the
+# broker is restarted. Set to 0 or a negative value to disable the timeout (not recommended).
+topicPoliciesCacheInitTimeoutSeconds=60
+
# If a topic remains fenced for this number of seconds, it will be closed forcefully.
# If it is set to 0 or a negative number, the fenced topic will not be closed.
topicFencingTimeoutSeconds=0
diff --git a/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/ServiceConfiguration.java b/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/ServiceConfiguration.java
index 192f1b3ad4c6..4b4920bcc0be 100644
--- a/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/ServiceConfiguration.java
+++ b/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/ServiceConfiguration.java
@@ -552,6 +552,19 @@ public class ServiceConfiguration implements PulsarConfiguration {
)
private long topicLoadTimeoutSeconds = 60;
+ @FieldContext(
+ category = CATEGORY_SERVER,
+ dynamic = true,
+ doc = "Amount of seconds to timeout initializing the topic policies cache of a namespace (reading the "
+ + "namespace's __change_events system topic to the end). Topic loading waits for this "
+ + "initialization, so if the system-topic reader gets stuck (for example after __change_events is "
+ + "unloaded and the reconnected reader stops making progress), this bounds the wait: the broker "
+ + "fails the initialization, closes the stuck reader and clears the cached state so that loading "
+ + "the namespace's topics can be retried with a fresh reader instead of hanging until the broker "
+ + "is restarted. Set to 0 or a negative value to disable the timeout (not recommended)."
+ )
+ private long topicPoliciesCacheInitTimeoutSeconds = 60;
+
@FieldContext(
category = CATEGORY_SERVER,
doc = "Whether we should enable metadata operations batching"
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/SystemTopicBasedTopicPoliciesService.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/SystemTopicBasedTopicPoliciesService.java
index f8a5d5936610..7e1b453b35b8 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/SystemTopicBasedTopicPoliciesService.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/SystemTopicBasedTopicPoliciesService.java
@@ -24,6 +24,7 @@ import com.github.benmanes.caffeine.cache.AsyncLoadingCache;
import com.github.benmanes.caffeine.cache.Caffeine;
import com.google.common.annotations.VisibleForTesting;
import com.google.common.collect.Sets;
+import io.opentelemetry.api.metrics.LongCounter;
import java.util.ArrayList;
import java.util.HashSet;
import java.util.List;
@@ -34,7 +35,9 @@ import java.util.PriorityQueue;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.Consumer;
@@ -78,10 +81,14 @@ import org.slf4j.LoggerFactory;
*/
public class SystemTopicBasedTopicPoliciesService implements TopicPoliciesService {
+ public static final String TOPIC_POLICIES_CACHE_INIT_TIMEOUT_METRIC_NAME =
+ "pulsar.broker.topic.policies.cache.init.timeout.count";
+
private final PulsarService pulsarService;
private final HashSet<String> localCluster;
private final String clusterName;
private final AtomicBoolean closed = new AtomicBoolean(false);
+ private final LongCounter policyCacheInitTimeoutCounter;
private final ConcurrentInitializer<NamespaceEventsSystemTopicFactory>
namespaceEventsSystemTopicFactoryLazyInitializer = new LazyInitializer<>() {
@@ -125,6 +132,13 @@ public class SystemTopicBasedTopicPoliciesService implements TopicPoliciesServic
this.pulsarService = pulsarService;
this.clusterName = pulsarService.getConfiguration().getClusterName();
this.localCluster = Sets.newHashSet(clusterName);
+ this.policyCacheInitTimeoutCounter = pulsarService.getOpenTelemetry().getMeter()
+ .counterBuilder(TOPIC_POLICIES_CACHE_INIT_TIMEOUT_METRIC_NAME)
+ .setDescription("The number of times initializing a namespace's topic policies cache timed out "
+ + "because the __change_events system-topic reader was stuck. Each occurrence closes the "
+ + "stuck reader and clears the cached state so topic loading can be retried.")
+ .setUnit("{timeout}")
+ .build();
this.writerCaches = Caffeine.newBuilder()
.expireAfterAccess(5, TimeUnit.MINUTES)
.removalListener((namespaceName, writer, cause) -> {
@@ -583,6 +597,12 @@ public class SystemTopicBasedTopicPoliciesService implements TopicPoliciesServic
CompletableFuture<Void> existingFuture =
policyCacheInitMap.putIfAbsent(namespace, initNamespacePolicyFuture);
if (existingFuture == null) {
+ // Topic loading waits on this future, so a system-topic reader that gets stuck (e.g. after
+ // __change_events is unloaded and the reconnected reader stops making progress) would pin the
+ // policy cache for the whole namespace until the broker restarts (issue #25294). Bound the
+ // initialization so it fails fast and cleans up, letting topic loading retry with a fresh
+ // reader.
+ scheduleInitPolicesCacheTimeout(namespace, initNamespacePolicyFuture);
final CompletableFuture<SystemTopicClient.Reader<PulsarEvent>> readerCompletableFuture =
newReader(namespace);
readerCompletableFuture
@@ -597,14 +617,19 @@ public class SystemTopicBasedTopicPoliciesService implements TopicPoliciesServic
return null;
}).exceptionally(ex -> {
try {
+ // Identity-guarded cleanup: a concurrent timeout cleanup or a namespace unload
+ // may already have dropped this future and let a retry install a fresh
+ // future/reader, so clean up by namespace key here would clobber that newer
+ // attempt. Tear down state only while this future still owns the namespace.
if (readerCompletableFuture.isCompletedExceptionally()) {
log.error("[{}] Failed to create reader on __change_events topic",
namespace, ex);
initNamespacePolicyFuture.completeExceptionally(ex);
- cleanPoliciesCacheInitMap(namespace, true);
+ cleanupFailedPolicyCacheInit(namespace, initNamespacePolicyFuture, true);
} else {
initNamespacePolicyFuture.completeExceptionally(ex);
- cleanPoliciesCacheInitMap(namespace, isAlreadyClosedException(ex));
+ cleanupFailedPolicyCacheInit(namespace, initNamespacePolicyFuture,
+ isAlreadyClosedException(ex));
}
} catch (Throwable cleanupEx) {
// Adding this catch to avoid break callback chain
@@ -685,6 +710,86 @@ public class SystemTopicBasedTopicPoliciesService implements TopicPoliciesServic
});
}
+ /**
+ * Bound the topic-policies cache initialization for {@code namespace} so a stuck {@code __change_events} reader
+ * cannot pin the cache (and therefore topic loading) for the whole namespace indefinitely (issue #25294). If the
+ * timeout wins the race to complete {@code initNamespacePolicyFuture}, the cached state is cleared and the stuck
+ * reader is closed so a subsequent load retries from scratch rather than waiting until the broker restarts.
+ */
+ private void scheduleInitPolicesCacheTimeout(@NonNull NamespaceName namespace,
+ @NonNull CompletableFuture<Void> initNamespacePolicyFuture) {
+ long timeoutSeconds = pulsarService.getConfiguration().getTopicPoliciesCacheInitTimeoutSeconds();
+ if (timeoutSeconds <= 0) {
+ return;
+ }
+ final ScheduledFuture<?> timeoutTask = pulsarService.getExecutor().schedule(() -> {
+ TimeoutException timeoutException = new TimeoutException(String.format(
+ "Timed out after %d seconds initializing the topic policies cache for namespace %s; the "
+ + "__change_events reader did not reach the end of the topic", timeoutSeconds, namespace));
+ if (initNamespacePolicyFuture.completeExceptionally(timeoutException)) {
+ policyCacheInitTimeoutCounter.add(1);
+ log.error("[{}] Timed out initializing the topic policies cache after {} seconds; closing the stuck "
+ + "__change_events reader so the namespace can be loaded again", namespace, timeoutSeconds);
+ try {
+ cleanupFailedPolicyCacheInit(namespace, initNamespacePolicyFuture, true);
+ } catch (Throwable cleanupEx) {
+ log.error("[{}] Failed to clean up the topic policies cache after init timeout", namespace,
+ cleanupEx);
+ }
+ }
+ }, timeoutSeconds, TimeUnit.SECONDS);
+ // Cancel the timeout once initialization finishes (successfully or not) so we don't leak scheduled tasks.
+ initNamespacePolicyFuture.whenComplete((__, ex) -> timeoutTask.cancel(false));
+ }
+
+ /**
+ * Identity-guarded cleanup for an initialization that failed (it timed out, the {@code __change_events} reader
+ * could not be created, or reading the topic threw). Unlike {@link #cleanPoliciesCacheInitMap}, which
+ * removes/closes by namespace key unconditionally, this only tears down state that still belongs to
+ * {@code initFuture}. By the time the failure is observed, a concurrent retry — or a namespace-bundle unload that
+ * left the init future orphaned — may already own the namespace with a fresh future and reader; removing by key
+ * would drop that newer future and close its reader, pinning the namespace again. Guarding on identity ensures a
+ * late failure never clobbers a newer initialization.
+ *
+ * @param closeReader when {@code true}, also clears the cached policies and closes the reader that belongs to this
+ * initialization; when {@code false}, only the init future is dropped, leaving the reader cached
+ * for the retry to reuse (mirrors the transient read-error path of
+ * {@link #cleanPoliciesCacheInitMap}).
+ */
+ @VisibleForTesting
+ void cleanupFailedPolicyCacheInit(@NonNull NamespaceName namespace,
+ @NonNull CompletableFuture<Void> initFuture, boolean closeReader) {
+ // Capture the reader before dropping the init future so we only close the reader that belongs to this
+ // initialization, never one a concurrent retry creates immediately afterwards.
+ CompletableFuture<SystemTopicClient.Reader<PulsarEvent>> readerFuture =
+ closeReader ? readerCaches.get(namespace) : null;
+ if (!policyCacheInitMap.remove(namespace, initFuture)) {
+ // Superseded by a retry or an unload; that owner is responsible for its own reader/state.
+ return;
+ }
+ // Complete the dropped future (a no-op if the caller already completed it) outside any map remapping function,
+ // so awaiting topic loads fail fast and retry instead of hanging until the broker restarts (issue #25294).
+ failPendingPolicyCacheInit(namespace, initFuture);
+ if (!closeReader) {
+ return;
+ }
+ policiesCache.entrySet().removeIf(entry -> Objects.equals(entry.getKey().getNamespaceObject(), namespace));
+ globalPoliciesCache.entrySet()
+ .removeIf(entry -> Objects.equals(entry.getKey().getNamespaceObject(), namespace));
+ TopicPolicyMessageHandlerTracker tracker = topicPolicyMessageHandlerTrackers.remove(namespace);
+ if (tracker != null) {
+ tracker.close();
+ }
+ if (readerFuture != null && readerCaches.remove(namespace, readerFuture)
+ && !readerFuture.isCompletedExceptionally()) {
+ readerFuture.thenCompose(SystemTopicClient.Reader::closeAsync)
+ .exceptionally(ex -> {
+ log.warn("[{}] Close change_event reader fail.", namespace, ex);
+ return null;
+ });
+ }
+ }
+
private void initPolicesCache(SystemTopicClient.Reader<PulsarEvent> reader, CompletableFuture<Void> future) {
if (closed.get()) {
future.completeExceptionally(new BrokerServiceException(getClass().getName() + " is closed."));
@@ -742,7 +847,7 @@ public class SystemTopicBasedTopicPoliciesService implements TopicPoliciesServic
@VisibleForTesting
void cleanPoliciesCacheInitMap(@NonNull NamespaceName namespace, boolean closeReader) {
if (!closeReader) {
- policyCacheInitMap.remove(namespace);
+ failPendingPolicyCacheInit(namespace, policyCacheInitMap.remove(namespace));
return;
}
@@ -753,12 +858,18 @@ public class SystemTopicBasedTopicPoliciesService implements TopicPoliciesServic
}
CompletableFuture<SystemTopicClient.Reader<PulsarEvent>> readerFuture = readerCaches.remove(namespace);
+ MutableObject<CompletableFuture<Void>> removedInitFuture = new MutableObject<>();
policyCacheInitMap.compute(namespace, (k, v) -> {
+ removedInitFuture.setValue(v);
policiesCache.entrySet().removeIf(entry -> Objects.equals(entry.getKey().getNamespaceObject(), namespace));
globalPoliciesCache.entrySet()
.removeIf(entry -> Objects.equals(entry.getKey().getNamespaceObject(), namespace));
return null;
});
+ // Complete the removed init future outside the compute() remapping function: completing it can run the
+ // awaiting topic-load callbacks synchronously, and doing that while holding the ConcurrentHashMap bin lock
+ // risks a recursive map update / deadlock (see #24977).
+ failPendingPolicyCacheInit(namespace, removedInitFuture.getValue());
if (readerFuture != null && !readerFuture.isCompletedExceptionally()) {
readerFuture
.thenCompose(SystemTopicClient.Reader::closeAsync)
@@ -769,6 +880,20 @@ public class SystemTopicBasedTopicPoliciesService implements TopicPoliciesServic
}
}
+ /**
+ * Complete an init future that is being dropped from {@link #policyCacheInitMap} but never completed, so the
+ * topic loads awaiting it fail fast and retry instead of hanging until the broker restarts (issue #25294).
+ * No-op if the future was already completed by its own initialization chain.
+ */
+ private void failPendingPolicyCacheInit(@NonNull NamespaceName namespace,
+ @Nullable CompletableFuture<Void> initFuture) {
+ if (initFuture != null && !initFuture.isDone()) {
+ initFuture.completeExceptionally(new BrokerServiceException(
+ "Topic policies cache initialization for namespace " + namespace
+ + " was aborted because the cached state was cleared"));
+ }
+ }
+
private void cleanWriterCache(@NonNull NamespaceName namespace) {
writerCaches.synchronous().invalidate(namespace);
}
@@ -1033,6 +1158,10 @@ public class SystemTopicBasedTopicPoliciesService implements TopicPoliciesServic
}
});
readerCaches.clear();
+ // Release any topic loads still waiting on an in-progress policy cache initialization so they fail
+ // fast instead of hanging until the awaited future is completed indirectly (issue #25294).
+ policyCacheInitMap.forEach(this::failPendingPolicyCacheInit);
+ policyCacheInitMap.clear();
}
}
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/SystemTopicBasedTopicPoliciesServiceTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/SystemTopicBasedTopicPoliciesServiceTest.java
index 8d3b16723ed4..1b5182a45452 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/SystemTopicBasedTopicPoliciesServiceTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/SystemTopicBasedTopicPoliciesServiceTest.java
@@ -28,6 +28,7 @@ import static org.testng.AssertJUnit.assertEquals;
import static org.testng.AssertJUnit.assertFalse;
import static org.testng.AssertJUnit.assertNotNull;
import static org.testng.AssertJUnit.assertNull;
+import static org.testng.AssertJUnit.assertSame;
import static org.testng.AssertJUnit.assertTrue;
import java.time.Duration;
import java.util.HashSet;
@@ -41,12 +42,15 @@ import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
import lombok.Cleanup;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.reflect.FieldUtils;
import org.apache.pulsar.broker.auth.MockedPulsarServiceBaseTest;
import org.apache.pulsar.broker.systopic.SystemTopicClient;
import org.apache.pulsar.client.admin.PulsarAdminException;
+import org.apache.pulsar.client.api.Message;
import org.apache.pulsar.client.api.Schema;
import org.apache.pulsar.common.events.PulsarEvent;
import org.apache.pulsar.common.naming.NamespaceName;
@@ -554,17 +558,18 @@ public class SystemTopicBasedTopicPoliciesServiceTest extends MockedPulsarServic
Assert.assertNull(readerCompletableFuture1);
});
-
- // make sure not do cleanPoliciesCacheInitMap() twice
- // totally trigger prepareInitPoliciesCacheAsync() twice, so the time of cleanPoliciesCacheInitMap() is 2.
- // in previous code, the time would be 3
+ // Cleanup must run exactly once per trigger and not repeat recursively (in older code it ran 3 times).
+ // Two failures are triggered here: the reader.close() above drives readMorePoliciesAsync into
+ // cleanPoliciesCacheInitMap (1x), and the second prepareInitPoliciesCacheAsync fails in initPolicesCache and
+ // is torn down by the identity-guarded cleanupFailedPolicyCacheInit (1x).
boolean logFound = testLogAppender.getEvents().stream().anyMatch(logEvent ->
logEvent.getMessage().toString().contains("Failed to create reader on __change_events topic"));
assertFalse(logFound);
boolean logFound2 = testLogAppender.getEvents().stream().anyMatch(logEvent ->
logEvent.getMessage().toString().contains("Failed to check the move events for the system topic"));
assertTrue(logFound2);
- verify(spyService, times(2)).cleanPoliciesCacheInitMap(any(), anyBoolean());
+ verify(spyService, times(1)).cleanPoliciesCacheInitMap(any(), anyBoolean());
+ verify(spyService, times(1)).cleanupFailedPolicyCacheInit(any(), any(), anyBoolean());
// make sure not occur Recursive update
boolean logFound3 = testLogAppender.getEvents().stream().anyMatch(logEvent ->
@@ -623,9 +628,9 @@ public class SystemTopicBasedTopicPoliciesServiceTest extends MockedPulsarServic
Assert.assertNull(readerCompletableFuture1);
});
-
- // make sure not do cleanPoliciesCacheInitMap() twice
- // totally trigger prepareInitPoliciesCacheAsync() once, so the time of cleanPoliciesCacheInitMap() is 1.
+ // Reader creation fails, so the single cleanup runs once via the identity-guarded
+ // cleanupFailedPolicyCacheInit (the reader-creation-failure branch no longer goes through
+ // cleanPoliciesCacheInitMap), and must not run more than once.
boolean logFound = testLogAppender.getEvents().stream().anyMatch(logEvent ->
logEvent.getMessage().toString().contains("Failed to create reader on __change_events topic"));
assertTrue(logFound);
@@ -633,6 +638,104 @@ public class SystemTopicBasedTopicPoliciesServiceTest extends MockedPulsarServic
logEvent.getMessage().toString().contains("Failed to check the move events for the system topic")
|| logEvent.getMessage().toString().contains("Failed to read event from the system topic"));
assertFalse(logFound2);
- verify(spyService, times(1)).cleanPoliciesCacheInitMap(any(), anyBoolean());
+ verify(spyService, times(1)).cleanupFailedPolicyCacheInit(any(), any(), anyBoolean());
+ verify(spyService, times(0)).cleanPoliciesCacheInitMap(any(), anyBoolean());
+ }
+
+ @Test(timeOut = 60_000)
+ public void testPrepareInitPoliciesCacheAsyncTimesOutWhenReaderStuck() throws Exception {
+ // Bound the policy-cache initialization to a short timeout for the test.
+ pulsar.getConfiguration().setTopicPoliciesCacheInitTimeoutSeconds(3);
+
+ pulsar.getTopicPoliciesService().close();
+ SystemTopicBasedTopicPoliciesService spyService =
+ Mockito.spy(new SystemTopicBasedTopicPoliciesService(pulsar));
+ FieldUtils.writeField(pulsar, "topicPoliciesService", spyService, true);
+
+ admin.namespaces().createNamespace(NAMESPACE5);
+ final NamespaceName namespace = NamespaceName.get(NAMESPACE5);
+
+ // Create a real __change_events reader, then spy it so that it reports more events but never delivers one —
+ // i.e. a reader that reconnected but is stuck (issue #25294). initPolicesCache would otherwise never complete.
+ SystemTopicClient.Reader<PulsarEvent> stuckReader = Mockito.spy(spyService.createSystemTopicClient(namespace)
+ .get(30, TimeUnit.SECONDS));
+ Mockito.doReturn(CompletableFuture.completedFuture(true)).when(stuckReader).hasMoreEventsAsync();
+ Mockito.doReturn(new CompletableFuture<Message<PulsarEvent>>()).when(stuckReader).readNextAsync();
+ Mockito.doReturn(CompletableFuture.completedFuture(stuckReader))
+ .when(spyService).createSystemTopicClient(namespace);
+
+ // Without the timeout the returned future never completes and topic loading for the namespace hangs forever.
+ CompletableFuture<Boolean> prepareFuture = spyService.prepareInitPoliciesCacheAsync(namespace);
+ try {
+ prepareFuture.get(15, TimeUnit.SECONDS);
+ Assert.fail("Expected the topic policies cache initialization to time out");
+ } catch (ExecutionException e) {
+ assertTrue("Expected a TimeoutException cause but got " + e.getCause(),
+ e.getCause() instanceof TimeoutException);
+ }
+
+ // The poisoned cache entry must be cleared and the stuck reader closed so a subsequent load can retry with a
+ // fresh reader instead of being pinned until the broker restarts.
+ Awaitility.await().untilAsserted(() -> assertNull(spyService.getPoliciesCacheInit(namespace)));
+ Mockito.verify(stuckReader, Mockito.atLeastOnce()).closeAsync();
+ }
+
+ @Test
+ public void testCleanPoliciesCacheInitMapCompletesPendingInitFuture() {
+ SystemTopicBasedTopicPoliciesService service =
+ (SystemTopicBasedTopicPoliciesService) pulsar.getTopicPoliciesService();
+ final NamespaceName namespace = NamespaceName.get(NAMESPACE1);
+
+ // Dropping the cached init future (e.g. on a namespace-bundle unload) must complete it so the topic loads
+ // awaiting it fail fast and retry, instead of hanging until the broker restarts (issue #25294).
+ CompletableFuture<Void> pendingWithReaderClose = new CompletableFuture<>();
+ service.policyCacheInitMap.put(namespace, pendingWithReaderClose);
+ service.cleanPoliciesCacheInitMap(namespace, true);
+ assertTrue(pendingWithReaderClose.isCompletedExceptionally());
+ assertNull(service.getPoliciesCacheInit(namespace));
+
+ CompletableFuture<Void> pendingWithoutReaderClose = new CompletableFuture<>();
+ service.policyCacheInitMap.put(namespace, pendingWithoutReaderClose);
+ service.cleanPoliciesCacheInitMap(namespace, false);
+ assertTrue(pendingWithoutReaderClose.isCompletedExceptionally());
+ assertNull(service.getPoliciesCacheInit(namespace));
+
+ // An already-completed init future must not be overwritten/disturbed.
+ CompletableFuture<Void> alreadyDone = CompletableFuture.completedFuture(null);
+ service.policyCacheInitMap.put(namespace, alreadyDone);
+ service.cleanPoliciesCacheInitMap(namespace, true);
+ assertFalse(alreadyDone.isCompletedExceptionally());
+ }
+
+ @Test
+ @SuppressWarnings("unchecked")
+ public void testCleanupFailedPolicyCacheInitIsIdentityGuarded() {
+ SystemTopicBasedTopicPoliciesService service =
+ (SystemTopicBasedTopicPoliciesService) pulsar.getTopicPoliciesService();
+ final NamespaceName namespace = NamespaceName.get(NAMESPACE1);
+
+ // A newer init attempt (B) already owns the namespace with a fresh future and reader.
+ CompletableFuture<Void> newerInitFuture = new CompletableFuture<>();
+ SystemTopicClient.Reader<PulsarEvent> newerReader = Mockito.mock(SystemTopicClient.Reader.class);
+ Mockito.doReturn(CompletableFuture.completedFuture(null)).when(newerReader).closeAsync();
+ service.policyCacheInitMap.put(namespace, newerInitFuture);
+ service.getReaderCaches().put(namespace, CompletableFuture.completedFuture(newerReader));
+
+ // A stale init attempt (A) whose reader was already torn down (e.g. by the timeout cleanup) fires its failure
+ // callback late. Cleaning it up by identity must be a no-op: it must not clobber B's future or close B's reader
+ // (issue #25294 follow-up).
+ CompletableFuture<Void> staleInitFuture = new CompletableFuture<>();
+ service.cleanupFailedPolicyCacheInit(namespace, staleInitFuture, true);
+ assertSame(newerInitFuture, service.getPoliciesCacheInit(namespace));
+ assertFalse(newerInitFuture.isDone());
+ assertNotNull(service.getReaderCaches().get(namespace));
+ Mockito.verify(newerReader, Mockito.never()).closeAsync();
+
+ // Cleaning up the owning attempt does drop its future and close its reader.
+ service.cleanupFailedPolicyCacheInit(namespace, newerInitFuture, true);
+ assertNull(service.getPoliciesCacheInit(namespace));
+ assertTrue(newerInitFuture.isCompletedExceptionally());
+ assertNull(service.getReaderCaches().get(namespace));
+ Mockito.verify(newerReader, Mockito.times(1)).closeAsync();
}
}