remove cache if can not get broker address when a broker node lost

streamnative · Jan 26, 2021 · 47b9f87 · 47b9f87
1 parent 241800b
commit 47b9f87
Showing 1 changed file with 3 additions and 0 deletions.
diff --git a/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/KafkaRequestHandler.java b/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/KafkaRequestHandler.java
@@ -1521,6 +1521,7 @@ public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
                 if (matchBrokers.isEmpty()) {
                     log.error("No node for broker {} under zk://loadbalance", pulsarAddress);
                     returnFuture.complete(Optional.empty());
+                    topicManager.removeTopicManagerCache(topic.toString());
                     return;
                 }
 
@@ -1538,6 +1539,7 @@ public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
                             if (th != null) {
                                 log.error("Error in getDataAsync() for {}", pulsarAddress, th);
                                 returnFuture.complete(Optional.empty());
+                                topicManager.removeTopicManagerCache(topic.toString());
                                 return;
                             }
 
@@ -1561,6 +1563,7 @@ public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
                             } catch (Exception e) {
                                 log.error("Error in {} lookupFuture get: ", pulsarAddress, e);
                                 returnFuture.complete(Optional.empty());
+                                topicManager.removeTopicManagerCache(topic.toString());
                                 return;
                             }