[fix][broker] Fix NPE causing dispatching to stop when using Key_Shar…

…ed mode and allowOutOfOrderDelivery=true (apache#22533)

(cherry picked from commit 2badcf6)

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

@@ -480,6 +480,11 @@ private int getAvailablePermits(Consumer c) {
 
     @Override
     protected synchronized NavigableSet<PositionImpl> filterOutEntriesWillBeDiscarded(NavigableSet<PositionImpl> src) {
+        // The variable "hashesToBeBlocked" and "recentlyJoinedConsumers" will be null if "isAllowOutOfOrderDelivery()",
+        // So skip this filter out.
+        if (isAllowOutOfOrderDelivery()) {
+            return src;
+        }
         if (src.isEmpty()) {
             return src;
         }
@@ -524,6 +529,11 @@ protected synchronized NavigableSet<PositionImpl> filterOutEntriesWillBeDiscarde
      */
     @Override
     protected boolean hasConsumersNeededNormalRead() {
+        // The variable "hashesToBeBlocked" and "recentlyJoinedConsumers" will be null if "isAllowOutOfOrderDelivery()",
+        // So the method "filterOutEntriesWillBeDiscarded" will filter out nothing, just return "true" here.
+        if (isAllowOutOfOrderDelivery()) {
+            return true;
+        }
         for (Consumer consumer : consumerList) {
             if (consumer == null || consumer.isBlocked()) {
                 continue;

pulsar-broker/src/test/java/org/apache/pulsar/client/api/KeySharedSubscriptionTest.java

@@ -1725,6 +1725,14 @@ public void testNoRepeatedReadAndDiscard() throws Exception {
         admin.topics().delete(topic, false);
     }
 
+    @DataProvider(name = "allowKeySharedOutOfOrder")
+    public Object[][] allowKeySharedOutOfOrder() {
+        return new Object[][]{
+                {true},
+                {false}
+        };
+    }
+
     /**
      * This test is in order to guarantee the feature added by https://github.com/apache/pulsar/pull/7105.
      * 1. Start 3 consumers:
@@ -1739,8 +1747,8 @@ public void testNoRepeatedReadAndDiscard() throws Exception {
      *   - no repeated Read-and-discard.
      *   - at last, all messages will be received.
      */
-    @Test(timeOut = 180 * 1000) // the test will be finished in 60s.
-    public void testRecentJoinedPosWillNotStuckOtherConsumer() throws Exception {
+    @Test(timeOut = 180 * 1000, dataProvider = "allowKeySharedOutOfOrder") // the test will be finished in 60s.
+    public void testRecentJoinedPosWillNotStuckOtherConsumer(boolean allowKeySharedOutOfOrder) throws Exception {
         final int messagesSentPerTime = 100;
         final Set<Integer> totalReceivedMessages = new TreeSet<>();
         final String topic = BrokerTestUtil.newUniqueName("persistent://public/default/tp");
@@ -1759,6 +1767,8 @@ public void testRecentJoinedPosWillNotStuckOtherConsumer() throws Exception {
             log.info("Published message :{}", messageId);
         }
 
+        KeySharedPolicy keySharedPolicy = KeySharedPolicy.autoSplitHashRange()
+                .setAllowOutOfOrderDelivery(allowKeySharedOutOfOrder);
         // 1. Start 3 consumers and make ack holes.
         //   - one consumer will be closed and trigger a messages redeliver.
         //   - one consumer will not ack any messages to make the new consumer joined late will be stuck due to the
@@ -1769,18 +1779,21 @@ public void testRecentJoinedPosWillNotStuckOtherConsumer() throws Exception {
                 .subscriptionName(subName)
                 .receiverQueueSize(10)
                 .subscriptionType(SubscriptionType.Key_Shared)
+                .keySharedPolicy(keySharedPolicy)
                 .subscribe();
         Consumer<Integer> consumer2 = pulsarClient.newConsumer(Schema.INT32)
                 .topic(topic)
                 .subscriptionName(subName)
                 .receiverQueueSize(10)
                 .subscriptionType(SubscriptionType.Key_Shared)
+                .keySharedPolicy(keySharedPolicy)
                 .subscribe();
         Consumer<Integer> consumer3 = pulsarClient.newConsumer(Schema.INT32)
                 .topic(topic)
                 .subscriptionName(subName)
                 .receiverQueueSize(10)
                 .subscriptionType(SubscriptionType.Key_Shared)
+                .keySharedPolicy(keySharedPolicy)
                 .subscribe();
         List<Message> msgList1 = new ArrayList<>();
         List<Message> msgList2 = new ArrayList<>();
@@ -1829,6 +1842,7 @@ public void testRecentJoinedPosWillNotStuckOtherConsumer() throws Exception {
                 .subscriptionName(subName)
                 .receiverQueueSize(1000)
                 .subscriptionType(SubscriptionType.Key_Shared)
+                .keySharedPolicy(keySharedPolicy)
                 .subscribe();
         consumerWillBeClose.close();