add topic null check

streamnative · Dec 18, 2020 · b8f2b30 · b8f2b30
1 parent a50ace5
commit b8f2b30
Showing 1 changed file with 8 additions and 2 deletions.
diff --git a/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/KafkaTopicManager.java b/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/KafkaTopicManager.java
@@ -344,7 +344,10 @@ public synchronized void close() {
                         requestHandler.ctx.channel(), references.get(topicName), topicName);
                 }
                 if (references.get(topicName) != null) {
-                    topicFuture.get().removeProducer(references.get(topicName));
+                    PersistentTopic persistentTopic = topicFuture.get();
+                    if (persistentTopic != null) {
+                        persistentTopic.removeProducer(references.get(topicName));
+                    }
                     references.remove(topicName);
                 }
             }
@@ -373,7 +376,10 @@ public void deReference(String topicName) {
             if (!topics.containsKey(topicName)) {
                 return;
             }
-            topics.get(topicName).get().removeProducer(references.get(topicName));
+            PersistentTopic persistentTopic = topics.get(topicName).get();
+            if (persistentTopic != null) {
+                persistentTopic.removeProducer(references.get(topicName));
+            }
             topics.remove(topicName);
         } catch (Exception e) {
             log.error("[{}] Failed to close reference for individual topic {}. exception:",