[fix][broker] Fix NPE causing dispatching to stop when using Key_Shared mode and allowOutOfOrderDelivery=true

[poorbarcode]

Motivation

In Key_Shared mode, the variable recentlyJoinedConsumers of PersistentStickyKeyDispatcherMultipleConsumers will always be null when enabling the config below. #22245 introduced a NPE if set allowOutOfOrderDelivery to true, which will lead dispatching stuck.

// The default value is false.
pulsarClient.newConsumer().keySharedPolicy(KeySharedPolicy.setAllowOutOfOrderDelivery(true))

error logs

--- An unexpected error occurred in the server ---

Message: org.apache.bookkeeper.mledger.ManagedLedgerException: Other exception

Stacktrace:

org.apache.pulsar.broker.service.BrokerServiceException: org.apache.bookkeeper.mledger.ManagedLedgerException: Other exception
	at org.apache.pulsar.broker.service.persistent.PersistentSubscription$6.findEntryFailed(PersistentSubscription.java:761)
	at org.apache.pulsar.broker.service.persistent.PersistentMessageFinder.findEntryFailed(PersistentMessageFinder.java:113)
	at org.apache.bookkeeper.mledger.impl.OpFindNewest.readEntryFailed(OpFindNewest.java:141)
	at org.apache.bookkeeper.mledger.impl.cache.RangeEntryCacheImpl.lambda$asyncReadEntry0$1(RangeEntryCacheImpl.java:274)
	at java.base/java.util.concurrent.CompletableFuture.uniExceptionally(CompletableFuture.java:990)
	at java.base/java.util.concurrent.CompletableFuture$UniExceptionally.tryFire(CompletableFuture.java:974)
	at java.base/java.util.concurrent.CompletableFuture.postComplete(CompletableFuture.java:510)
	at java.base/java.util.concurrent.CompletableFuture.postFire(CompletableFuture.java:614)
	at java.base/java.util.concurrent.CompletableFuture$UniAccept.tryFire(CompletableFuture.java:726)
	at java.base/java.util.concurrent.CompletableFuture$Completion.run(CompletableFuture.java:482)
	at org.apache.bookkeeper.common.util.SingleThreadExecutor.safeRunTask(SingleThreadExecutor.java:128)
	at org.apache.bookkeeper.common.util.SingleThreadExecutor.run(SingleThreadExecutor.java:99)
	at io.netty.util.concurrent.FastThreadLocalRunnable.run(FastThreadLocalRunnable.java:30)
	at java.base/java.lang.Thread.run(Thread.java:840)
Caused by: org.apache.bookkeeper.mledger.ManagedLedgerException: Other exception
Caused by: java.lang.NullPointerException: Cannot invoke "java.util.LinkedHashMap.containsKey(Object)" because "this.recentlyJoinedConsumers" is null
	at org.apache.pulsar.broker.service.persistent.PersistentStickyKeyDispatcherMultipleConsumers.hasConsumersNeededNormalRead(PersistentStickyKeyDispatcherMultipleConsumers.java:508)
	at org.apache.pulsar.broker.service.persistent.PersistentDispatcherMultipleConsumers.readMoreEntries(PersistentDispatcherMultipleConsumers.java:344)
	at org.apache.pulsar.broker.service.persistent.PersistentDispatcherMultipleConsumers.removeConsumer(PersistentDispatcherMultipleConsumers.java:228)
	at org.apache.pulsar.broker.service.persistent.PersistentStickyKeyDispatcherMultipleConsumers.removeConsumer(PersistentStickyKeyDispatcherMultipleConsumers.java:151)
	at org.apache.pulsar.broker.service.persistent.PersistentSubscription.removeConsumer(PersistentSubscription.java:301)
	at org.apache.pulsar.broker.service.Consumer.close(Consumer.java:402)
	at org.apache.pulsar.broker.service.Consumer.disconnect(Consumer.java:418)
	at org.apache.pulsar.broker.service.persistent.PersistentDispatcherMultipleConsumers.lambda$disconnectAllConsumers$7(PersistentDispatcherMultipleConsumers.java:548)
	at java.base/java.util.concurrent.CopyOnWriteArrayList.forEach(CopyOnWriteArrayList.java:807)
	at org.apache.pulsar.broker.service.persistent.PersistentDispatcherMultipleConsumers.disconnectAllConsumers(PersistentDispatcherMultipleConsumers.java:548)
	at org.apache.pulsar.broker.service.Dispatcher.disconnectAllConsumers(Dispatcher.java:73)
	at org.apache.pulsar.broker.service.persistent.PersistentDispatcherMultipleConsumers.disconnectActiveConsumers(PersistentDispatcherMultipleConsumers.java:563)
	at org.apache.pulsar.broker.service.persistent.PersistentSubscription.resetCursor(PersistentSubscription.java:787)
	at org.apache.pulsar.broker.service.persistent.PersistentSubscription$6.findEntryComplete(PersistentSubscription.java:751)
	at org.apache.pulsar.broker.service.persistent.PersistentMessageFinder.findEntryComplete(PersistentMessageFinder.java:101)
	at org.apache.bookkeeper.mledger.impl.OpFindNewest.readEntryComplete(OpFindNewest.java:131)
	at org.apache.bookkeeper.mledger.impl.cache.RangeEntryCacheImpl.lambda$asyncReadEntry0$0(RangeEntryCacheImpl.java:262)
	at java.base/java.util.concurrent.CompletableFuture$UniAccept.tryFire(CompletableFuture.java:718)
	at java.base/java.util.concurrent.CompletableFuture$Completion.run(CompletableFuture.java:482)
	at org.apache.bookkeeper.common.util.SingleThreadExecutor.safeRunTask(SingleThreadExecutor.java:128)
	at org.apache.bookkeeper.common.util.SingleThreadExecutor.run(SingleThreadExecutor.java:99)
	at io.netty.util.concurrent.FastThreadLocalRunnable.run(FastThreadLocalRunnable.java:30)
	at java.base/java.lang.Thread.run(Thread.java:840)

Modifications

Fix the NPE.

Documentation

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

Matching PR in forked repository

PR in forked repository: x

[lhotari]

LGTM