NPE in PersistentStickyKeyDispatcherMultipleConsumers

[shustsud]

If use HashRangeExclusiveStickyKeyConsumerSelector(Key_Shared), NPE may occurs.

Log:

13:43:42.150 [bookkeeper-ml-workers-OrderedExecutor-0-0] ERROR org.apache.bookkeeper.common.util.SafeRunnable - Unexpected throwable caught 
java.lang.NullPointerException: null
    at org.apache.pulsar.broker.service.persistent.PersistentStickyKeyDispatcherMultipleConsumers.sendMessagesToConsumers(PersistentStickyKeyDispatcherMultipleConsumers.java:178) ~[classes/:?]
    at org.apache.pulsar.broker.service.persistent.PersistentDispatcherMultipleConsumers.readEntriesComplete(PersistentDispatcherMultipleConsumers.java:476) ~[classes/:?]
    at org.apache.bookkeeper.mledger.impl.OpReadEntry.lambda$checkReadCompletion$2(OpReadEntry.java:152) ~[classes/:?]
    at org.apache.bookkeeper.mledger.util.SafeRun$1.safeRun(SafeRun.java:32) ~[classes/:?]
    at org.apache.bookkeeper.common.util.SafeRunnable.run(SafeRunnable.java:36) [bookkeeper-common-4.12.0.jar:4.12.0]
    at org.apache.bookkeeper.common.util.OrderedExecutor$TimedRunnable.run(OrderedExecutor.java:203) [bookkeeper-common-4.12.0.jar:4.12.0]
    at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149) [?:1.8.0_201]
    at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624) [?:1.8.0_201]
    at io.netty.util.concurrent.FastThreadLocalRunnable.run(FastThreadLocalRunnable.java:30) [netty-common-4.1.51.Final.jar:4.1.51.Final]
    at java.lang.Thread.run(Thread.java:748) [?:1.8.0_201]

If consumer is set for only part of hash range, consumer may be set to null.

pulsar/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentStickyKeyDispatcherMultipleConsumers.java

Line 168 in cc64889

Consumer c = selector.select(peekStickyKey(entry.getDataBuffer()));

If null is set in consumser, NPE will occur below.

pulsar/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentStickyKeyDispatcherMultipleConsumers.java

Line 178 in cc64889

final int availablePermits = Math.max(consumer.getAvailablePermits(), 0);

To Reproduce

• step1
  Create consumer as shown below and receive a message from broker.

PulsarClient pulsarClient = PulsarClient.builder()
        .serviceUrl(<url>)
        .build();

Consumer<byte[]> consumer = pulsarClient.newConsumer()
        .topic(<topic>)
        .subscriptionName(<subscriptionName>)
        .ackTimeout(5, TimeUnit.SECONDS)
        .subscriptionType(SubscriptionType.Key_Shared)
        .keySharedPolicy(KeySharedPolicy.stickyHashRange().ranges(Range.of(0, 9)))
        .subscribe();

• step2
  Create producer and send a message to broker.

• step3
  If a consumer with a hash value between 10 and 65535 is chosen with HashRangeExclusiveStickyKeyConsumerSelector, NPE occurs.