[fix][client] Fix NPE of MultiTopicsConsumerImpl due to race condition

[codelipenghui]

Motivation

 "exception":"java.util.concurrent.CompletionException: java.lang.NullPointerException\n\tat java.base/java.util.concurrent.CompletableFuture.encodeThrowable(CompletableFuture.java:314)\n\tat 
java.base/java.util.concurrent.CompletableFuture.completeThrowable(CompletableFuture.java:319)\n\tat 
java.base/java.util.concurrent.CompletableFuture$UniAccept.tryFire(CompletableFuture.java:718)\n\tat 
java.base/java.util.concurrent.CompletableFuture$Completion.run(CompletableFuture.java:478)\n\tat 
java.base/java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:515)\n\tat 
java.base/java.util.concurrent.FutureTask.run(FutureTask.java:264)\n\tat 
java.base/java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.jav
a:304)\n\tat java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1128)\n\tat 
java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:628)\n\tat 
org.apache.pulsar.shade.io.netty.util.concurrent.FastThreadLocalRunnable.run(FastThreadLocalRunnable.java:30)\n\tat 
java.base/java.lang.Thread.run(Thread.java:829)\nCaused by: java.lang.NullPointerException: null\n\tat 
org.apache.pulsar.client.impl.MultiTopicsConsumerImpl.messageReceived(MultiTopicsConsumerImpl.java:298)\n\tat 
org.apache.pulsar.client.impl.MultiTopicsConsumerImpl.lambda$receiveMessageFromConsumer$7(MultiTopicsConsumerI
mpl.java:254)\n\tat java.base/java.util.concurrent.CompletableFuture$UniAccept.tryFire(CompletableFuture.java:714)\n\t... 
8 common frames omitted\n" 

The stack trace is based on 2.9.2

The close and unsubscribe operation will change the unAckedMessageTracker to null which will cause the above problem.

Modifications

We should stop processing messages after the consumer is closed.

Does this pull request potentially affect one of the following parts:

If the box was checked, please highlight the changes

• Dependencies (add or upgrade a dependency)
• The public API
• The schema
• The default values of configurations
• The threading model
• The binary protocol
• The REST endpoints
• The admin CLI options
• Anything that affects deployment

Documentation

• doc
• doc-required
• doc-not-needed
• doc-complete

[codelipenghui]

/pulsarbot run-failure-checks

[codecov-commenter]

Codecov Report

Merging #18287 (30dd5c7) into master (0866c3a) will decrease coverage by 0.80%.
The diff coverage is 64.28%.

@@             Coverage Diff              @@
##             master   #18287      +/-   ##
============================================
- Coverage     38.97%   38.16%   -0.81%     
+ Complexity     8311     8014     -297     
============================================
  Files           683      659      -24     
  Lines         67325    65091    -2234     
  Branches       7217     6972     -245     
============================================
- Hits          26239    24845    -1394     
+ Misses        38079    37410     -669     
+ Partials       3007     2836     -171     

Flag	Coverage Δ
unittests	38.16% <64.28%> (-0.81%)	⬇️

Flags with carried forward coverage won't be shown. Click here to find out more.

Impacted Files	Coverage Δ
.../main/java/org/apache/pulsar/PulsarStandalone.java	0.00% <ø> (ø)
...pulsar/broker/service/PulsarCommandSenderImpl.java	69.63% <ø> (-4.73%)	⬇️
...ransaction/buffer/impl/InMemTransactionBuffer.java	0.00% <ø> (ø)
...nsaction/buffer/impl/TransactionBufferDisable.java	52.63% <ø> (ø)
.../metadata/v2/TransactionBufferSnapshotIndexes.java	0.00% <0.00%> (ø)
...a/v2/TransactionBufferSnapshotIndexesMetadata.java	0.00% <0.00%> (ø)
...oker/transaction/buffer/metadata/v2/TxnIDData.java	0.00% <0.00%> (ø)
...he/pulsar/client/impl/MultiTopicsConsumerImpl.java	0.00% <0.00%> (ø)
...ulsar/client/impl/transaction/TransactionImpl.java	0.00% <0.00%> (ø)
...rg/apache/pulsar/client/util/ExecutorProvider.java	0.00% <0.00%> (ø)
... and 109 more

[poorbarcode]

Good Catch!!!