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/07/25 15:30:56 UTC
(pulsar) 07/08: [fix][broker] Prevent stale topic unload cleanup from removing active cache entries (#26145)
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 8d3a94bc044bfec8b8c8369bc651ad22fae1a823
Author: void-ptr974 <vo...@protonmail.com>
AuthorDate: Sat Jul 25 22:50:08 2026 +0800
[fix][broker] Prevent stale topic unload cleanup from removing active cache entries (#26145)
(cherry picked from commit 73549d246541dd3d12bd6e17e246cddda5e436d8)
---
.../pulsar/broker/service/BrokerService.java | 112 +++++++----
.../pulsar/broker/service/BrokerServiceTest.java | 213 ++++++++++++++++++++-
2 files changed, 287 insertions(+), 38 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 e13c11e3dde7..c99def4e42d4 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
@@ -229,6 +229,11 @@ public class BrokerService implements Closeable {
private final Map<String, CompletableFuture<Optional<Topic>>> topics = new ConcurrentHashMap<>();
+ // A topic can reach the cache-cleanup path through both its close callback and bundle cleanup.
+ // Keep only one cleanup active for a topic future so re-entrant unload listeners cannot duplicate side effects.
+ private final Set<CompletableFuture<Optional<Topic>>> topicCacheRemovalsInProgress =
+ ConcurrentHashMap.newKeySet();
+
private final Map<String, PulsarClient> replicationClients = new ConcurrentHashMap<>();
private final Map<String, PulsarAdmin> clusterAdmins = new ConcurrentHashMap<>();
@@ -2506,13 +2511,19 @@ public class BrokerService implements Closeable {
}
public void cleanUnloadedTopicFromCache(NamespaceBundle serviceUnit) {
- for (String topic : topics.keySet()) {
+ topics.forEach((topic, topicFuture) -> {
TopicName topicName = TopicName.get(topic);
- if (serviceUnit.includes(topicName) && getTopicReference(topic).isPresent()) {
+ Optional<Topic> topicRef = extractTopic(topicFuture);
+ // The bundle cleanup should only remove topics already fenced by close/unload.
+ if (serviceUnit.includes(topicName) && topicRef.isPresent() && isTopicBeingUnloaded(topicRef.get())) {
log.info("[{}][{}] Clean unloaded topic from cache.", serviceUnit.toString(), topic);
- pulsar.getBrokerService().removeTopicFromCache(topicName.toString(), serviceUnit, null);
+ removeTopicFromCache(topicName.toString(), serviceUnit, topicFuture);
}
- }
+ });
+ }
+
+ private boolean isTopicBeingUnloaded(Topic topic) {
+ return topic instanceof AbstractTopic abstractTopic && abstractTopic.isFenced();
}
public AuthorizationService getAuthorizationService() {
@@ -2521,7 +2532,8 @@ public class BrokerService implements Closeable {
/**
* Removes the topic from the cache only if the topicName and associated createFuture match exactly.
- * The TopicEvent.UNLOAD event will be triggered before and after removal.
+ * The TopicEvent.UNLOAD BEFORE event is triggered while the matching cache entry is still present;
+ * the SUCCESS event is triggered after it has been removed.
*
* @param topic The topic to be removed.
* @return A CompletableFuture that completes when the operation is done.
@@ -2536,45 +2548,71 @@ public class BrokerService implements Closeable {
private void removeTopicFromCache(String topic, NamespaceBundle namespaceBundle,
CompletableFuture<Optional<Topic>> createTopicFuture) {
- String bundleName = namespaceBundle.toString();
- String namespaceName = TopicName.get(topic).getNamespaceObject().toString();
-
- topicEventsDispatcher.notify(topic, TopicEvent.UNLOAD, EventStage.BEFORE);
-
- synchronized (multiLayerTopicsMap) {
- final var namespaceMap = multiLayerTopicsMap.get(namespaceName);
- if (namespaceMap != null) {
- final var bundleMap = namespaceMap.get(bundleName);
- if (bundleMap != null) {
- bundleMap.remove(topic);
- if (bundleMap.isEmpty()) {
- namespaceMap.remove(bundleName);
- }
+ if (createTopicFuture == null) {
+ if (log.isDebugEnabled()) {
+ log.debug("[{}] Skip removing topic from cache without its expected future.", topic);
+ }
+ return;
+ }
+ if (!topicCacheRemovalsInProgress.add(createTopicFuture)) {
+ if (log.isDebugEnabled()) {
+ log.debug("[{}] Skip removing topic from cache since cleanup is already in progress.", topic);
+ }
+ return;
+ }
+ try {
+ if (topics.get(topic) != createTopicFuture) {
+ // A stale close/unload callback must not emit unload side effects for a superseded topic future.
+ if (log.isDebugEnabled()) {
+ log.debug("[{}] Skip removing topic from cache since it was already removed or superseded.",
+ topic);
}
+ return;
+ }
+ String bundleName = namespaceBundle.toString();
+ String namespaceName = TopicName.get(topic).getNamespaceObject().toString();
+
+ topicEventsDispatcher.notify(topic, TopicEvent.UNLOAD, EventStage.BEFORE);
+
+ synchronized (multiLayerTopicsMap) {
+ final var namespaceMap = multiLayerTopicsMap.get(namespaceName);
+ if (namespaceMap != null) {
+ final var bundleMap = namespaceMap.get(bundleName);
+ if (bundleMap != null) {
+ bundleMap.remove(topic);
+ if (bundleMap.isEmpty()) {
+ namespaceMap.remove(bundleName);
+ }
+ }
- if (namespaceMap.isEmpty()) {
- multiLayerTopicsMap.remove(namespaceName);
- final ClusterReplicationMetrics clusterReplicationMetrics = pulsarStats
- .getClusterReplicationMetrics();
- replicationClients.forEach((cluster, client) -> {
- clusterReplicationMetrics.remove(clusterReplicationMetrics.getKeyName(namespaceName,
- cluster));
- });
+ if (namespaceMap.isEmpty()) {
+ multiLayerTopicsMap.remove(namespaceName);
+ final ClusterReplicationMetrics clusterReplicationMetrics = pulsarStats
+ .getClusterReplicationMetrics();
+ replicationClients.forEach((cluster, client) -> {
+ clusterReplicationMetrics.remove(clusterReplicationMetrics.getKeyName(namespaceName,
+ cluster));
+ });
+ }
}
}
- }
- if (createTopicFuture == null) {
- topics.remove(topic);
- } else {
- topics.remove(topic, createTopicFuture);
- }
+ Compactor compactor = pulsar.getNullableCompactor();
+ if (compactor != null) {
+ compactor.getStats().removeTopic(topic);
+ }
- Compactor compactor = pulsar.getNullableCompactor();
- if (compactor != null) {
- compactor.getStats().removeTopic(topic);
+ if (!topics.remove(topic, createTopicFuture)) {
+ if (log.isDebugEnabled()) {
+ log.debug("[{}] Skip unload success because the cache entry was already removed or superseded.",
+ topic);
+ }
+ return;
+ }
+ topicEventsDispatcher.notify(topic, TopicEvent.UNLOAD, EventStage.SUCCESS);
+ } finally {
+ topicCacheRemovalsInProgress.remove(createTopicFuture);
}
- topicEventsDispatcher.notify(topic, TopicEvent.UNLOAD, EventStage.SUCCESS);
}
public long getNumberOfNamespaceBundles() {
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 0a8f6601ad5f..b8fb257664e0 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
@@ -1325,6 +1325,187 @@ public class BrokerServiceTest extends BrokerTestBase {
}
+ @Test
+ public void testCleanUnloadedTopicFromCacheDoesNotRemoveNewTopicFuture() throws Exception {
+ final String namespace = "prop/ns-abc";
+ final String topicName = "persistent://" + namespace + "/cleanUnloadedTopicFromCache-"
+ + UUID.randomUUID();
+ final BrokerService brokerService = pulsar.getBrokerService();
+ Producer<byte[]> producer = pulsarClient.newProducer().topic(topicName).create();
+ producer.close();
+ Topic topic = brokerService.getTopicReference(topicName).orElseThrow();
+ CompletableFuture<Optional<Topic>> originalTopicFuture = brokerService.getTopics().get(topicName);
+ assertNotNull(originalTopicFuture);
+ NamespaceBundle bundle = pulsar.getNamespaceService().getBundle(TopicName.get(topicName));
+ BlockingCompletedFuture<Optional<Topic>> oldTopicFuture =
+ new BlockingCompletedFuture<>(Optional.of(topic));
+ brokerService.getTopics().put(topicName, oldTopicFuture);
+ ((AbstractTopic) topic).isFenced = true;
+
+ CompletableFuture<Optional<Topic>> newTopicFuture =
+ CompletableFuture.completedFuture(Optional.of(mock(Topic.class)));
+ List<TopicEventsListener.EventStage> unloadEvents = new ArrayList<>();
+ TopicEventsListener listener = (name, event, stage, t) -> {
+ if (topicName.equals(name) && event == TopicEventsListener.TopicEvent.UNLOAD) {
+ unloadEvents.add(stage);
+ }
+ };
+ brokerService.addTopicEventListener(listener);
+ try {
+ CompletableFuture<Void> cleanupFuture = CompletableFuture.runAsync(
+ () -> brokerService.cleanUnloadedTopicFromCache(bundle));
+ assertTrue(oldTopicFuture.awaitIsDone(5, TimeUnit.SECONDS),
+ "cleanup should capture the old future before it is replaced");
+
+ // Simulate a reload racing after the bundle cleanup has captured the old future.
+ brokerService.getTopics().put(topicName, newTopicFuture);
+ oldTopicFuture.allowIsDone();
+ cleanupFuture.get(5, TimeUnit.SECONDS);
+
+ assertTrue(brokerService.getTopics().get(topicName) == newTopicFuture,
+ "A stale bundle unload cleanup must not remove a newer topic future");
+ assertTrue(unloadEvents.isEmpty(),
+ "A stale bundle unload cleanup must not dispatch unload events");
+ } finally {
+ oldTopicFuture.allowIsDone();
+ brokerService.removeTopicEventListener(listener);
+ brokerService.getTopics().put(topicName, originalTopicFuture);
+ topic.close(false, true).get(5, TimeUnit.SECONDS);
+ }
+ }
+
+ @Test
+ public void testUnloadBeforeDoesNotStartTopicReload() throws Exception {
+ final String namespace = "prop/ns-abc";
+ final String topicName = "persistent://" + namespace + "/unloadBeforeDoesNotStartTopicReload-"
+ + UUID.randomUUID();
+ final BrokerService brokerService = pulsar.getBrokerService();
+ Producer<byte[]> producer = pulsarClient.newProducer().topic(topicName).create();
+ producer.close();
+ Topic topic = brokerService.getTopicReference(topicName).orElseThrow();
+ CompletableFuture<Optional<Topic>> topicFuture = brokerService.getTopics().get(topicName);
+ assertNotNull(topicFuture);
+
+ AtomicReference<CompletableFuture<Optional<Topic>>> futureAtUnloadBefore = new AtomicReference<>();
+ TopicEventsListener listener = (name, event, stage, t) -> {
+ if (topicName.equals(name) && event == TopicEventsListener.TopicEvent.UNLOAD) {
+ if (stage == TopicEventsListener.EventStage.BEFORE) {
+ futureAtUnloadBefore.set(brokerService.getTopic(TopicName.get(topicName), true, null));
+ }
+ }
+ };
+ brokerService.addTopicEventListener(listener);
+ try {
+ topic.close(true, true).get(5, TimeUnit.SECONDS);
+
+ assertTrue(futureAtUnloadBefore.get() == topicFuture,
+ "UNLOAD BEFORE must return the existing topic future instead of starting a reload");
+ assertNull(brokerService.getTopics().get(topicName),
+ "UNLOAD SUCCESS must observe the topic cache entry removed");
+ } finally {
+ brokerService.removeTopicEventListener(listener);
+ }
+ }
+
+ @Test
+ public void testRemoveTopicFromCacheIgnoresReentrantUnloadCallback() throws Exception {
+ final String namespace = "prop/ns-abc";
+ final String topicName = "persistent://" + namespace + "/removeTopicFromCacheReentrant-"
+ + UUID.randomUUID();
+ final BrokerService brokerService = pulsar.getBrokerService();
+ Producer<byte[]> producer = pulsarClient.newProducer().topic(topicName).create();
+ producer.close();
+ Topic topic = brokerService.getTopicReference(topicName).orElseThrow();
+
+ List<TopicEventsListener.EventStage> unloadEvents = new ArrayList<>();
+ AtomicReference<Throwable> reentrantCallbackError = new AtomicReference<>();
+ TopicEventsListener listener = (name, event, stage, t) -> {
+ if (topicName.equals(name) && event == TopicEventsListener.TopicEvent.UNLOAD) {
+ unloadEvents.add(stage);
+ if (stage == TopicEventsListener.EventStage.BEFORE) {
+ try {
+ brokerService.removeTopicFromCache((AbstractTopic) topic).get(5, TimeUnit.SECONDS);
+ } catch (Throwable e) {
+ reentrantCallbackError.set(e);
+ }
+ }
+ }
+ };
+ brokerService.addTopicEventListener(listener);
+ try {
+ brokerService.removeTopicFromCache((AbstractTopic) topic).get(5, TimeUnit.SECONDS);
+
+ assertNull(reentrantCallbackError.get(), "A reentrant cache cleanup should complete without error");
+ assertEquals(unloadEvents, List.of(TopicEventsListener.EventStage.BEFORE,
+ TopicEventsListener.EventStage.SUCCESS));
+ } finally {
+ brokerService.removeTopicEventListener(listener);
+ topic.close(false, true).get(5, TimeUnit.SECONDS);
+ }
+ }
+
+ @Test
+ public void testRemoveTopicFromCacheDoesNotRemoveSupersededTopicFuture() throws Exception {
+ final String namespace = "prop/ns-abc";
+ final String topicName = "persistent://" + namespace + "/removeSupersededTopicFromCache-"
+ + UUID.randomUUID();
+ final BrokerService brokerService = pulsar.getBrokerService();
+ Producer<byte[]> producer = pulsarClient.newProducer().topic(topicName).create();
+ producer.close();
+ Topic topic = brokerService.getTopicReference(topicName).orElseThrow();
+ CompletableFuture<Optional<Topic>> oldTopicFuture = brokerService.getTopics().get(topicName);
+ assertNotNull(oldTopicFuture);
+
+ List<TopicEventsListener.EventStage> unloadEvents = new ArrayList<>();
+ TopicEventsListener listener = (name, event, stage, t) -> {
+ if (topicName.equals(name) && event == TopicEventsListener.TopicEvent.UNLOAD) {
+ unloadEvents.add(stage);
+ }
+ };
+ brokerService.addTopicEventListener(listener);
+
+ CompletableFuture<Optional<Topic>> newTopicFuture =
+ CompletableFuture.completedFuture(Optional.of(topic));
+ // Simulate the same topic being reloaded before the old topic close callback removes it.
+ brokerService.getTopics().put(topicName, newTopicFuture);
+ try {
+ brokerService.removeTopicFromCache((AbstractTopic) topic).get(5, TimeUnit.SECONDS);
+
+ assertTrue(brokerService.getTopics().get(topicName) == newTopicFuture,
+ "A stale topic close callback must not remove a newer topic future");
+ assertTrue(unloadEvents.isEmpty(),
+ "A stale topic close callback must not dispatch unload events");
+ } finally {
+ brokerService.removeTopicEventListener(listener);
+ brokerService.getTopics().put(topicName, oldTopicFuture);
+ brokerService.removeTopicFromCache((AbstractTopic) topic).get(5, TimeUnit.SECONDS);
+ }
+ }
+
+ @Test
+ public void testCleanUnloadedTopicFromCacheRemovesInactiveBundleTopicFuture() throws Exception {
+ final String namespace = "prop/ns-abc";
+ final String topicName = "persistent://" + namespace + "/cleanInactiveUnloadedTopicFromCache-"
+ + UUID.randomUUID();
+ final BrokerService brokerService = pulsar.getBrokerService();
+ Producer<byte[]> producer = pulsarClient.newProducer().topic(topicName).create();
+ producer.close();
+ Topic topic = brokerService.getTopicReference(topicName).orElseThrow();
+ NamespaceBundle bundle = pulsar.getNamespaceService().getBundle(TopicName.get(topicName));
+
+ pulsar.getNamespaceService().getOwnershipCache().updateBundleState(bundle, false).get(5, TimeUnit.SECONDS);
+ ((AbstractTopic) topic).isFenced = true;
+ try {
+ brokerService.cleanUnloadedTopicFromCache(bundle);
+
+ assertFalse(brokerService.getTopics().containsKey(topicName),
+ "Inactive bundle cleanup should remove the matching topic future");
+ } finally {
+ pulsar.getNamespaceService().getOwnershipCache().updateBundleState(bundle, true).get(5, TimeUnit.SECONDS);
+ topic.close(false, true).get(5, TimeUnit.SECONDS);
+ }
+ }
+
@Test
public void testCheckInactiveSubscriptionWhenNoMessageToAck() throws Exception {
String namespace = "prop/testInactiveSubscriptionWhenNoMessageToAck";
@@ -2047,6 +2228,37 @@ public class BrokerServiceTest extends BrokerTestBase {
assertFalse(MockTopicPoliciesService.FAILED_TOPICS.contains(topicName));
}
+ private static class BlockingCompletedFuture<T> extends CompletableFuture<T> {
+ private final CountDownLatch isDoneEntered = new CountDownLatch(1);
+ private final CountDownLatch allowIsDone = new CountDownLatch(1);
+
+ BlockingCompletedFuture(T value) {
+ complete(value);
+ }
+
+ @Override
+ public boolean isDone() {
+ isDoneEntered.countDown();
+ try {
+ if (!allowIsDone.await(10, TimeUnit.SECONDS)) {
+ throw new AssertionError("Timed out waiting to resume topic-cache cleanup");
+ }
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw new AssertionError("Interrupted while waiting to resume topic-cache cleanup", e);
+ }
+ return super.isDone();
+ }
+
+ boolean awaitIsDone(long timeout, TimeUnit unit) throws InterruptedException {
+ return isDoneEntered.await(timeout, unit);
+ }
+
+ void allowIsDone() {
+ allowIsDone.countDown();
+ }
+ }
+
static class MockTopicPoliciesService extends TopicPoliciesService.TopicPoliciesServiceDisabled {
static final Set<TopicName> FAILED_TOPICS = ConcurrentHashMap.newKeySet();
@@ -2062,4 +2274,3 @@ public class BrokerServiceTest extends BrokerTestBase {
}
}
}
-