void-ptr974 commented on code in PR #26145:
URL: https://github.com/apache/pulsar/pull/26145#discussion_r3566281677
##########
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:
Fixed in d8443489685. The expected future now remains in topics through
UNLOAD BEFORE and all auxiliary cleanup, so a normal getTopic() call cannot
install a replacement in the middle. We conditionally remove the expected
future only after cleanup, then emit UNLOAD SUCCESS. Added a real UNLOAD BEFORE
getTopic() regression test.
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java:
##########
Review Comment:
Fixed in d8443489685. UNLOAD BEFORE is now emitted while the matching cache
entry is still present; UNLOAD SUCCESS is emitted only after its conditional
removal. The new regression test invokes getTopic() from an UNLOAD BEFORE
listener and verifies it receives the existing future rather than starting a
reload.
##########
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:
Fixed in d8443489685. The test now blocks immediately after the cleanup scan
captures the old future, replaces the cache entry with a distinct new future,
and then resumes cleanup. It verifies the new future remains and that the stale
callback emits no UNLOAD events. A separate real UNLOAD BEFORE getTopic() test
covers the normal reload path and ensures a replacement cannot be created
during old-topic cleanup.
--
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]
