From 1ca7376b9982d38aa6f569a32df33ee9a0a0281f Mon Sep 17 00:00:00 2001
From: Lari Hotari <lhotari@users.noreply.github.com>
Date: Wed, 7 Jun 2023 16:39:45 +0300
Subject: [PATCH] [fix][broker] Restore solution for certain topic unloading
 race conditions (#20527)

(cherry picked from commit 03f916702ec2a887833543fdea5a2d5305a87302)
---
 .../java/org/apache/pulsar/broker/service/BrokerService.java  | 4 ----
 .../pulsar/broker/service/persistent/PersistentTopic.java     | 2 +-
 .../apache/pulsar/broker/service/BrokerBkEnsemblesTests.java  | 4 ++--
 3 files changed, 3 insertions(+), 7 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 2cee82a97ae21..5dba45576a667 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
@@ -2059,10 +2059,6 @@ public AuthorizationService getAuthorizationService() {
         return authorizationService;
     }
 
-    public CompletableFuture<Void> removeTopicFromCache(String topicName) {
-        return removeTopicFutureFromCache(topicName, null);
-    }
-
     public CompletableFuture<Void> removeTopicFromCache(Topic topic) {
         Optional<CompletableFuture<Optional<Topic>>> createTopicFuture = findTopicFutureInCache(topic);
         if (createTopicFuture.isEmpty()){
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java
index 22cd0b7d123f4..364d32b55919a 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java
@@ -1327,7 +1327,7 @@ public void closeFailed(ManagedLedgerException exception, Object ctx) {
     }
 
     private void disposeTopic(CompletableFuture<?> closeFuture) {
-        brokerService.removeTopicFromCache(topic)
+        brokerService.removeTopicFromCache(PersistentTopic.this)
                 .thenRun(() -> {
                     replicatedSubscriptionsController.ifPresent(ReplicatedSubscriptionsController::close);
 
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/BrokerBkEnsemblesTests.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/BrokerBkEnsemblesTests.java
index 8b0dbb76eecb7..4cc5b009d108b 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/BrokerBkEnsemblesTests.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/BrokerBkEnsemblesTests.java
@@ -127,7 +127,7 @@ public void testCrashBrokerWithoutCursorLedgerLeak() throws Exception {
         // (3) remove topic and managed-ledger from broker which means topic is not closed gracefully
         consumer.close();
         producer.close();
-        pulsar.getBrokerService().removeTopicFromCache(topic1);
+        pulsar.getBrokerService().removeTopicFromCache(topic);
         ManagedLedgerFactoryImpl factory = (ManagedLedgerFactoryImpl) pulsar.getManagedLedgerFactory();
         Field field = ManagedLedgerFactoryImpl.class.getDeclaredField("ledgers");
         field.setAccessible(true);
@@ -252,7 +252,7 @@ public void testSkipCorruptDataLedger() throws Exception {
 
         // clean managed-ledger and recreate topic to clean any data from the cache
         producer.close();
-        pulsar.getBrokerService().removeTopicFromCache(topic1);
+        pulsar.getBrokerService().removeTopicFromCache(topic);
         ManagedLedgerFactoryImpl factory = (ManagedLedgerFactoryImpl) pulsar.getManagedLedgerFactory();
         Field field = ManagedLedgerFactoryImpl.class.getDeclaredField("ledgers");
         field.setAccessible(true);