[fix][client] Fix NPE of MultiTopicsConsumerImpl due to race condition
@@ -256,6 +256,10 @@ public class MultiTopicsConsumerImpl<T> extends ConsumerBase<T> {
log.debug("[{}] [{}] Receive message from sub consumer:{}",
topic, subscription, consumer.getTopic());
}
+ // Stop to process the remaining message after the consumer is closed.
+ if (getState() == State.Closed) {
+ return;
+ }
// Process the message, add to the queue and trigger listener or async callback
messages.forEach(msg -> messageReceived(consumer, msg));
@@ -562,14 +566,14 @@ public class MultiTopicsConsumerImpl<T> extends ConsumerBase<T> {
.map(ConsumerImpl::unsubscribeAsync).collect(Collectors.toList());
FutureUtil.waitForAll(futureList)
- .thenCompose((r) -> {
+ .thenComposeAsync((r) -> {
setState(State.Closed);
cleanupMultiConsumer();
log.info("[{}] [{}] [{}] Unsubscribed Topics Consumer",
topic, subscription, consumerName);
// fail all pending-receive futures to notify application
return failPendingReceive();
- })
+ }, internalPinnedExecutor)
.whenComplete((r, ex) -> {
if (ex == null) {
unsubscribeFuture.complete(null);
@@ -604,13 +608,13 @@ public class MultiTopicsConsumerImpl<T> extends ConsumerBase<T> {
.map(ConsumerImpl::closeAsync).collect(Collectors.toList());
FutureUtil.waitForAll(futureList)
- .thenCompose((r) -> {
+ .thenComposeAsync((r) -> {
setState(State.Closed);
cleanupMultiConsumer();
log.info("[{}] [{}] Closed Topics Consumer", topic, subscription);
// fail all pending-receive futures to notify application
return failPendingReceive();
- })
+ }, internalPinnedExecutor)
.whenComplete((r, ex) -> {
if (ex == null) {
closeFuture.complete(null);
