fix PatternTopicsChangedListener blocked when topic removed
@@ -1208,7 +1208,9 @@ public class MultiTopicsConsumerImpl<T> extends ConsumerBase<T> {
});
removeTopic(topicName);
+ if (unAckedMessageTracker instanceof UnAckedTopicMessageTracker) {
((UnAckedTopicMessageTracker) unAckedMessageTracker).removeTopicMessages(topicName);
+ }
unsubscribeFuture.complete(null);
log.info("[{}] [{}] [{}] Unsubscribed Topics Consumer, allTopicPartitionsNumber: {}",
@@ -1256,7 +1258,9 @@ public class MultiTopicsConsumerImpl<T> extends ConsumerBase<T> {
});
removeTopic(topicName);
+ if (unAckedMessageTracker instanceof UnAckedTopicMessageTracker) {
((UnAckedTopicMessageTracker) unAckedMessageTracker).removeTopicMessages(topicName);
+ }
unsubscribeFuture.complete(null);
log.info("[{}] [{}] [{}] Removed Topics Consumer, allTopicPartitionsNumber: {}",
