diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentStickyKeyDispatcherMultipleConsumers.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentStickyKeyDispatcherMultipleConsumers.java index ee2ebd7ca867e..2df9f38531f5d 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentStickyKeyDispatcherMultipleConsumers.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentStickyKeyDispatcherMultipleConsumers.java @@ -457,6 +457,11 @@ private int getAvailablePermits(Consumer c) { @Override protected synchronized NavigableSet filterOutEntriesWillBeDiscarded(NavigableSet 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; } @@ -501,6 +506,11 @@ protected synchronized NavigableSet 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; diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/KeySharedSubscriptionTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/KeySharedSubscriptionTest.java index 7219555050839..27aa98597ec12 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/KeySharedSubscriptionTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/KeySharedSubscriptionTest.java @@ -1741,6 +1741,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: @@ -1755,8 +1763,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 totalReceivedMessages = new TreeSet<>(); final String topic = BrokerTestUtil.newUniqueName("persistent://public/default/tp"); @@ -1775,6 +1783,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 @@ -1785,18 +1795,21 @@ public void testRecentJoinedPosWillNotStuckOtherConsumer() throws Exception { .subscriptionName(subName) .receiverQueueSize(10) .subscriptionType(SubscriptionType.Key_Shared) + .keySharedPolicy(keySharedPolicy) .subscribe(); Consumer consumer2 = pulsarClient.newConsumer(Schema.INT32) .topic(topic) .subscriptionName(subName) .receiverQueueSize(10) .subscriptionType(SubscriptionType.Key_Shared) + .keySharedPolicy(keySharedPolicy) .subscribe(); Consumer consumer3 = pulsarClient.newConsumer(Schema.INT32) .topic(topic) .subscriptionName(subName) .receiverQueueSize(10) .subscriptionType(SubscriptionType.Key_Shared) + .keySharedPolicy(keySharedPolicy) .subscribe(); List msgList1 = new ArrayList<>(); List msgList2 = new ArrayList<>(); @@ -1845,6 +1858,7 @@ public void testRecentJoinedPosWillNotStuckOtherConsumer() throws Exception { .subscriptionName(subName) .receiverQueueSize(1000) .subscriptionType(SubscriptionType.Key_Shared) + .keySharedPolicy(keySharedPolicy) .subscribe(); consumerWillBeClose.close();