Denovo1998 commented on code in PR #26145:
URL: https://github.com/apache/pulsar/pull/26145#discussion_r3565567564
##########
pulsar-broker/src/test/java/org/apache/pulsar/broker/service/BrokerServiceTest.java:
##########
@@ -1330,6 +1330,96 @@ public void
testCheckInactiveSubscriptionsShouldNotDeleteCompactionCursor() thro
}
+ @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>> oldTopicFuture =
brokerService.getTopics().get(topicName);
+ assertNotNull(oldTopicFuture);
+ NamespaceBundle bundle =
pulsar.getNamespaceService().getBundle(TopicName.get(topicName));
+
+ CompletableFuture<Optional<Topic>> newTopicFuture =
+ CompletableFuture.completedFuture(Optional.of(topic));
+ // Simulate the same topic being reloaded before a stale bundle
cleanup callback runs.
+ brokerService.getTopics().put(topicName, newTopicFuture);
Review Comment:
This test reuses the same unfenced Topic instance in newTopicFuture. As a
result, cleanUnloadedTopicFromCache() exits at isTopicBeingUnloaded() and never
calls removeTopicFromCache(topic, bundle, capturedFuture).
I confirmed this by replacing the conditional removal with an unconditional
topics.remove(topic); the test still passed. To resolve this, coordinate the
replacement only after the cleanup iteration has captured the old future—for
example, using latches around the fenced-state check—and use a distinct active
topic and future. Additionally, it would be useful to verify that the new topic
remains in multiLayerTopicsMap and that no stale unload events are emitted.
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java:
##########
@@ -2748,6 +2756,16 @@ public CompletableFuture<Void>
removeTopicFromCache(AbstractTopic topic) {
private void removeTopicFromCache(String topic, NamespaceBundle
namespaceBundle,
CompletableFuture<Optional<Topic>>
createTopicFuture) {
+ boolean removed = createTopicFuture == null
Review Comment:
The `topics.remove(topic, createTopicFuture)` method validates the topic
generation only at the moment the primary cache entry is removed. Once removal
succeeds, the key becomes absent, allowing `getTopic()` to immediately install
a newer future while the old callback is still executing. This callback then
removes the topic from `multiLayerTopicsMap`, clears compactor/segment state,
and emits `UNLOAD SUCCESS`—all of which can impact the newly loaded topic.
This issue can be reproduced deterministically by installing or reloading
the new future from an `UNLOAD BEFORE` listener. The listener runs after the
primary entry has already been removed but before the auxiliary cleanup below.
To resolve this, keep the expected cache entry present or reserved until its
side effects are complete, or ensure all auxiliary cleanup and event generation
is aware of the current state. The existing ordering still allows stale unload
side effects, which this PR aims to prevent.
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java:
##########
Review Comment:
After this change, UNLOAD BEFORE is dispatched after the matching entry has
already been removed from BrokerService.topics. This changes the observable
event contract: EventStage.BEFORE is documented as “before starting the event”,
and before this PR the notification also occurred before cache removal.
A listener handling BEFORE can now observe the topic as absent or trigger a
reload before the rest of the unload cleanup has run. Please preserve the
BEFORE-before-removal ordering, or explicitly redefine and test the event
semantics if this behavior change is intentional.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
