[pulsar] branch branch-2.7 updated: fix PatternTopicsChangedListener blocked when topic removed (#16842)

[jianghaiting]

This is an automated email from the ASF dual-hosted git repository.

jianghaiting pushed a commit to branch branch-2.7
in repository https://gitbox.apache.org/repos/asf/pulsar.git


The following commit(s) were added to refs/heads/branch-2.7 by this push:
     new f741a5686f0 fix PatternTopicsChangedListener blocked when topic 
removed (#16842)
f741a5686f0 is described below

commit f741a5686f0df7ad4f52c44b4558cbae3a7a00ae
Author: WangJialing <65590138+wangjialing...@users.noreply.github.com>
AuthorDate: Fri Jul 29 22:30:23 2022 +0800

    fix PatternTopicsChangedListener blocked when topic removed (#16842)
    
    (cherry picked from commit 2d6b534ef72824ffb5b2f76568ca8346bbc058c6)
---
 .../org/apache/pulsar/client/impl/MultiTopicsConsumerImpl.java    | 8 ++++++--
 1 file changed, 6 insertions(+), 2 deletions(-)

diff --git 
a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/MultiTopicsConsumerImpl.java
 
b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/MultiTopicsConsumerImpl.java
index b438c08fecb..998ebc7ba4e 100644
--- 
a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/MultiTopicsConsumerImpl.java
+++ 
b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/MultiTopicsConsumerImpl.java
@@ -1063,7 +1063,9 @@ public class MultiTopicsConsumerImpl<T> extends 
ConsumerBase<T> {
                     });
 
                     removeTopic(topicName);
-                    ((UnAckedTopicMessageTracker) 
unAckedMessageTracker).removeTopicMessages(topicName);
+                    if (unAckedMessageTracker instanceof 
UnAckedTopicMessageTracker) {
+                        ((UnAckedTopicMessageTracker) 
unAckedMessageTracker).removeTopicMessages(topicName);
+                    }
 
                     unsubscribeFuture.complete(null);
                     log.info("[{}] [{}] [{}] Unsubscribed Topics Consumer, 
allTopicPartitionsNumber: {}",
@@ -1111,7 +1113,9 @@ public class MultiTopicsConsumerImpl<T> extends 
ConsumerBase<T> {
                     });
 
                     removeTopic(topicName);
-                    ((UnAckedTopicMessageTracker) 
unAckedMessageTracker).removeTopicMessages(topicName);
+                    if (unAckedMessageTracker instanceof 
UnAckedTopicMessageTracker) {
+                        ((UnAckedTopicMessageTracker) 
unAckedMessageTracker).removeTopicMessages(topicName);
+                    }
 
                     unsubscribeFuture.complete(null);
                     log.info("[{}] [{}] [{}] Removed Topics Consumer, 
allTopicPartitionsNumber: {}",