fix npe when unloading persistent partitioned topic (apache#11310)

Co-authored-by: wuzhanpeng <wuzhanpeng@bigo.sg> (cherry picked from commit a73dc61)
datastax · Feb 9, 2022 · 820b069 · 820b069
1 parent 89e76f2
commit 820b069
Show file tree

Hide file tree

Showing 2 changed files with 41 additions and 5 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
@@ -1739,7 +1739,8 @@ private CompletableFuture<Integer> unloadServiceUnit(NamespaceBundle serviceUnit
     public void cleanUnloadedTopicFromCache(NamespaceBundle serviceUnit) {
         for (String topic : topics.keys()) {
             TopicName topicName = TopicName.get(topic);
-            if (serviceUnit.includes(topicName)) {
+            if (getTopicReference(topic).isPresent() && serviceUnit.includes(topicName)) {
+                log.info("[{}][{}] Clean unloaded topic from cache.", serviceUnit.toString(), topic);
                 pulsar.getBrokerService().removeTopicFromCache(topicName.toString(), serviceUnit);
             }
         }
@@ -1766,9 +1767,11 @@ public void removeTopicFromCache(String topic, NamespaceBundle namespaceBundle)
                     .get(namespaceName);
             if (namespaceMap != null) {
                 ConcurrentOpenHashMap<String, Topic> bundleMap = namespaceMap.get(bundleName);
-                bundleMap.remove(topic);
-                if (bundleMap.isEmpty()) {
-                    namespaceMap.remove(bundleName);
+                if (bundleMap != null) {
+                    bundleMap.remove(topic);
+                    if (bundleMap.isEmpty()) {
+                        namespaceMap.remove(bundleName);
+                    }
                 }
 
                 if (namespaceMap.isEmpty()) {

diff --git a/...broker/src/test/java/org/apache/pulsar/broker/service/persistent/PersistentTopicTest.java b/...broker/src/test/java/org/apache/pulsar/broker/service/persistent/PersistentTopicTest.java
@@ -32,7 +32,8 @@
 import static org.testng.Assert.assertNull;
 
 import java.lang.reflect.Field;
-import java.util.Map;
+import java.util.ArrayList;
+import java.util.List;
 import java.util.UUID;
 import java.util.concurrent.TimeUnit;
 
@@ -240,4 +241,36 @@ public void testAccumulativeStats() throws Exception {
         assertEquals(statsAfterUnsubscribe.getBytesOutCounter(), statsBeforeUnsubscribe.getBytesOutCounter());
         assertEquals(statsAfterUnsubscribe.getMsgOutCounter(), statsBeforeUnsubscribe.getMsgOutCounter());
     }
+
+    @Test
+    public void testPersistentPartitionedTopicUnload() throws Exception {
+        final String topicName = "persistent://prop/ns/failedUnload";
+        final String ns = "prop/ns";
+        final int partitions = 5;
+        final int producers = 1;
+        // ensure that the number of bundle is greater than 1
+        final int bundles = 2;
+
+        admin.namespaces().createNamespace(ns, bundles);
+        admin.topics().createPartitionedTopic(topicName, partitions);
+
+        List<Producer> producerSet = new ArrayList<>();
+        for (int i = 0; i < producers; i++) {
+            producerSet.add(pulsarClient.newProducer(Schema.STRING).topic(topicName).create());
+        }
+
+        assertFalse(pulsar.getBrokerService().getTopics().containsKey(topicName));
+        pulsar.getBrokerService().getTopicIfExists(topicName).get();
+        assertTrue(pulsar.getBrokerService().getTopics().containsKey(topicName));
+
+        // ref of partitioned-topic name should be empty
+        assertFalse(pulsar.getBrokerService().getTopicReference(topicName).isPresent());
+
+        NamespaceBundle bundle = pulsar.getNamespaceService().getBundle(TopicName.get(topicName));
+        pulsar.getNamespaceService().unloadNamespaceBundle(bundle, 5, TimeUnit.SECONDS).get();
+
+        for (Producer producer : producerSet) {
+            producer.close();
+        }
+    }
 }