You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by yu...@apache.org on 2026/09/22 11:13:40 UTC
(pulsar) branch branch-4.0 updated: [fix][broker] Bound classic Key_Shared dispatcher replay queue look-ahead (#26677)
This is an automated email from the ASF dual-hosted git repository.
poorbarcode pushed a commit to branch branch-4.0
in repository https://gitbox.apache.org/repos/asf/pulsar.git
The following commit(s) were added to refs/heads/branch-4.0 by this push:
new bb7aea7a3939 [fix][broker] Bound classic Key_Shared dispatcher replay queue look-ahead (#26677)
bb7aea7a3939 is described below
commit bb7aea7a3939d6c17c30a149e1033401ead4c8c0
Author: fengyubiao <yu...@streamnative.io>
AuthorDate: Tue Sep 22 19:00:02 2026 +0800
[fix][broker] Bound classic Key_Shared dispatcher replay queue look-ahead (#26677)
(cherry picked from commit 3d5fbc323371a3293b3d43339fd238641d20d2cc)
---
...tickyKeyDispatcherMultipleConsumersClassic.java | 11 +++
...tentDispatcherMultipleConsumersClassicTest.java | 95 ++++++++++++++++++++++
...yKeyDispatcherMultipleConsumersClassicTest.java | 59 ++++++++++++++
3 files changed, 165 insertions(+)
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentStickyKeyDispatcherMultipleConsumersClassic.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentStickyKeyDispatcherMultipleConsumersClassic.java
index 16e1f3d64370..596868204b17 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentStickyKeyDispatcherMultipleConsumersClassic.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentStickyKeyDispatcherMultipleConsumersClassic.java
@@ -594,6 +594,9 @@ public class PersistentStickyKeyDispatcherMultipleConsumersClassic
*/
@Override
protected boolean hasConsumersNeededNormalRead() {
+ if (!isReplayQueueSizeBelowLimit()) {
+ return false;
+ }
// Classic out-of-order replay filtering is bypassed, so normal reads do not need the ordered-mode escape check.
if (isAllowOutOfOrderDelivery()) {
return true;
@@ -612,6 +615,14 @@ public class PersistentStickyKeyDispatcherMultipleConsumersClassic
return false;
}
+ private boolean isReplayQueueSizeBelowLimit() {
+ int effectiveLookAheadLimit = PersistentStickyKeyDispatcherMultipleConsumers
+ .getEffectiveLookAheadLimit(serviceConfig, consumerList.size());
+ // The limit is checked before a normal read. A read that starts below the limit can add up to one read batch
+ // to the replay queue, so this provides a bounded look-ahead limit with an acceptable one-batch overflow.
+ return redeliveryMessages.size() < effectiveLookAheadLimit;
+ }
+
@Override
public SubType getType() {
return SubType.Key_Shared;
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherMultipleConsumersClassicTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherMultipleConsumersClassicTest.java
index bc0278870829..586e66c82822 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherMultipleConsumersClassicTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherMultipleConsumersClassicTest.java
@@ -18,6 +18,7 @@
*/
package org.apache.pulsar.broker.service.persistent;
+import static java.nio.charset.StandardCharsets.UTF_8;
import com.carrotsearch.hppc.ObjectSet;
import java.util.List;
import java.util.Optional;
@@ -34,16 +35,21 @@ import org.apache.bookkeeper.mledger.ManagedCursor;
import org.apache.bookkeeper.mledger.ManagedLedgerException;
import org.apache.bookkeeper.mledger.impl.ManagedCursorImpl;
import org.apache.pulsar.broker.BrokerTestUtil;
+import org.apache.pulsar.broker.ServiceConfiguration;
import org.apache.pulsar.broker.service.Dispatcher;
+import org.apache.pulsar.broker.service.StickyKeyConsumerSelector;
import org.apache.pulsar.broker.service.Subscription;
import org.apache.pulsar.broker.service.Topic;
import org.apache.pulsar.client.api.Consumer;
+import org.apache.pulsar.client.api.KeySharedPolicy;
+import org.apache.pulsar.client.api.Message;
import org.apache.pulsar.client.api.MessageId;
import org.apache.pulsar.client.api.Producer;
import org.apache.pulsar.client.api.ProducerConsumerBase;
import org.apache.pulsar.client.api.Schema;
import org.apache.pulsar.client.api.SubscriptionType;
import org.apache.pulsar.common.api.proto.MessageMetadata;
+import org.awaitility.Awaitility;
import org.awaitility.reflect.WhiteboxImpl;
import org.mockito.Mockito;
import org.testng.Assert;
@@ -68,6 +74,80 @@ public class PersistentDispatcherMultipleConsumersClassicTest extends ProducerCo
super.internalCleanup();
}
+ @Test(timeOut = 30 * 1000)
+ public void testKeySharedReplayQueueResumesAfterConsumerFlow() throws Exception {
+ boolean useClassicDispatcher = getConfig().isSubscriptionKeySharedUseClassicPersistentImplementation();
+ int perConsumerLimit = getConfig().getKeySharedLookAheadMsgInReplayThresholdPerConsumer();
+ int perSubscriptionLimit = getConfig().getKeySharedLookAheadMsgInReplayThresholdPerSubscription();
+ getConfig().setSubscriptionKeySharedUseClassicPersistentImplementation(true);
+ getConfig().setKeySharedLookAheadMsgInReplayThresholdPerConsumer(2);
+ getConfig().setKeySharedLookAheadMsgInReplayThresholdPerSubscription(2);
+ try {
+ String topicName = newTopicName();
+ String subscriptionName = "key-shared";
+ KeySharedPolicy keySharedPolicy = KeySharedPolicy.autoSplitHashRange();
+ @Cleanup
+ Consumer<String> fastConsumer = pulsarClient.newConsumer(Schema.STRING)
+ .topic(topicName)
+ .subscriptionName(subscriptionName)
+ .consumerName("fast")
+ .subscriptionType(SubscriptionType.Key_Shared)
+ .keySharedPolicy(keySharedPolicy)
+ .subscribe();
+ @Cleanup
+ Consumer<String> slowConsumer = pulsarClient.newConsumer(Schema.STRING)
+ .topic(topicName)
+ .subscriptionName(subscriptionName)
+ .consumerName("slow")
+ .subscriptionType(SubscriptionType.Key_Shared)
+ .keySharedPolicy(keySharedPolicy)
+ .receiverQueueSize(1)
+ .subscribe();
+ @Cleanup
+ Producer<String> producer = pulsarClient.newProducer(Schema.STRING).topic(topicName).create();
+
+ PersistentTopic topic = (PersistentTopic) getTopic(topicName, false).join().orElseThrow();
+ PersistentStickyKeyDispatcherMultipleConsumersClassic dispatcher =
+ (PersistentStickyKeyDispatcherMultipleConsumersClassic) topic.getSubscription(subscriptionName)
+ .getDispatcher();
+ org.apache.pulsar.broker.service.Consumer slowServiceConsumer = dispatcher.getConsumers().stream()
+ .filter(consumer -> consumer.consumerName().equals("slow"))
+ .findFirst()
+ .orElseThrow();
+ org.apache.pulsar.broker.service.Consumer fastServiceConsumer = dispatcher.getConsumers().stream()
+ .filter(consumer -> consumer.consumerName().equals("fast"))
+ .findFirst()
+ .orElseThrow();
+ StickyKeyConsumerSelector selector = dispatcher.getSelector();
+ String slowKey = keyForConsumer(selector, slowServiceConsumer);
+ String fastKey = keyForConsumer(selector, fastServiceConsumer);
+
+ producer.newMessage().key(slowKey).value("slow-1").send();
+ producer.newMessage().key(slowKey).value("slow-2").send();
+ producer.newMessage().key(slowKey).value("slow-3").send();
+
+ Awaitility.await().untilAsserted(() ->
+ Assert.assertEquals(dispatcher.getNumberOfMessagesInReplay(), 2L));
+
+ producer.newMessage().key(fastKey).value("fast").send();
+ Assert.assertNull(fastConsumer.receive(5, TimeUnit.SECONDS));
+
+ Message<String> firstSlowMessage = slowConsumer.receive(5, TimeUnit.SECONDS);
+ Assert.assertNotNull(firstSlowMessage);
+ slowConsumer.acknowledge(firstSlowMessage);
+
+ Message<String> fastMessage = fastConsumer.receive(5, TimeUnit.SECONDS);
+ Assert.assertNotNull(fastMessage);
+ Assert.assertEquals(fastMessage.getValue(), "fast");
+ Awaitility.await().untilAsserted(() ->
+ Assert.assertTrue(dispatcher.getNumberOfMessagesInReplay() < 2));
+ } finally {
+ getConfig().setSubscriptionKeySharedUseClassicPersistentImplementation(useClassicDispatcher);
+ getConfig().setKeySharedLookAheadMsgInReplayThresholdPerConsumer(perConsumerLimit);
+ getConfig().setKeySharedLookAheadMsgInReplayThresholdPerSubscription(perSubscriptionLimit);
+ }
+ }
+
@Test(timeOut = 30 * 1000)
public void testTopicDeleteIfConsumerSetMismatchConsumerList() throws Exception {
final String topicName = BrokerTestUtil.newUniqueName("persistent://public/default/tp");
@@ -281,4 +361,19 @@ public class PersistentDispatcherMultipleConsumersClassicTest extends ProducerCo
protected CompletableFuture<Optional<Topic>> getTopic(String topic, boolean createIfMissing) {
return pulsar.getBrokerService().getTopic(topic, createIfMissing);
}
+
+ private String keyForConsumer(StickyKeyConsumerSelector selector,
+ org.apache.pulsar.broker.service.Consumer consumer) {
+ for (int i = 0; i < 100_000; i++) {
+ String key = "key-" + i;
+ if (selector.select(key.getBytes(UTF_8)) == consumer) {
+ return key;
+ }
+ }
+ throw new IllegalStateException("No key found for consumer " + consumer.consumerName());
+ }
+
+ private ServiceConfiguration getConfig() {
+ return pulsar.getConfig();
+ }
}
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/PersistentStickyKeyDispatcherMultipleConsumersClassicTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/PersistentStickyKeyDispatcherMultipleConsumersClassicTest.java
index edece4342e40..0bdd648c1a73 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/PersistentStickyKeyDispatcherMultipleConsumersClassicTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/PersistentStickyKeyDispatcherMultipleConsumersClassicTest.java
@@ -37,6 +37,7 @@ import static org.mockito.Mockito.when;
import static org.testng.Assert.assertEquals;
import static org.testng.Assert.assertTrue;
import static org.testng.Assert.fail;
+import static org.testng.AssertJUnit.assertFalse;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.EventLoopGroup;
@@ -212,6 +213,64 @@ public class PersistentStickyKeyDispatcherMultipleConsumersClassicTest {
}
}
+ @Test
+ public void testNormalReadIsBoundedByKeySharedLookAheadLimit() throws Exception {
+ doReturn(5).when(configMock).getKeySharedLookAheadMsgInReplayThresholdPerConsumer();
+ doReturn(10).when(configMock).getKeySharedLookAheadMsgInReplayThresholdPerSubscription();
+ persistentDispatcher.addConsumer(consumerMock).get();
+
+ persistentDispatcher.redeliveryMessages.add(1, 1, 1);
+ persistentDispatcher.redeliveryMessages.add(1, 2, 1);
+ persistentDispatcher.redeliveryMessages.add(1, 3, 1);
+
+ assertTrue(persistentDispatcher.hasConsumersNeededNormalRead());
+
+ persistentDispatcher.redeliveryMessages.add(1, 4, 1);
+ persistentDispatcher.redeliveryMessages.add(1, 5, 1);
+
+ assertFalse(persistentDispatcher.hasConsumersNeededNormalRead());
+ }
+
+ @Test(timeOut = 10000)
+ public void testConsumerFlowResumesDispatchWhenReplayQueueIsFull() throws Exception {
+ doReturn(1).when(configMock).getKeySharedLookAheadMsgInReplayThresholdPerConsumer();
+ doReturn(1).when(configMock).getKeySharedLookAheadMsgInReplayThresholdPerSubscription();
+
+ AtomicInteger availablePermits = new AtomicInteger();
+ Consumer slowConsumer = createMockConsumer();
+ doReturn("slow-consumer").when(slowConsumer).consumerName();
+ doAnswer(invocation -> availablePermits.get()).when(slowConsumer).getAvailablePermits();
+ doReturn(true).when(slowConsumer).isWritable();
+ CountDownLatch messageSent = new CountDownLatch(1);
+ mockSendMessages(slowConsumer, entries -> messageSent.countDown());
+ persistentDispatcher.addConsumer(slowConsumer).get();
+
+ StickyKeyConsumerSelector selector = persistentDispatcher.getSelector();
+ String key = generateKeyForConsumer(selector, slowConsumer);
+ Entry entry = createEntry(1, 1, "message", 1, key);
+ int stickyKeyHash = selector.makeStickyKeyHash(key.getBytes(UTF_8));
+ persistentDispatcher.redeliveryMessages.add(1, 1, stickyKeyHash);
+ assertFalse(persistentDispatcher.hasConsumersNeededNormalRead());
+
+ doAnswer(invocation -> {
+ AsyncCallbacks.ReadEntriesCallback callback = invocation.getArgument(1);
+ Object ctx = invocation.getArgument(2);
+ callback.readEntriesComplete(copyEntries(List.of(entry)), ctx);
+ return Collections.emptySet();
+ }).when(cursorMock).asyncReplayEntries(anySet(), any(), any(), anyBoolean());
+
+ availablePermits.set(1);
+ persistentDispatcher.consumerFlow(slowConsumer, 1);
+
+ assertTrue(messageSent.await(5, TimeUnit.SECONDS));
+ Awaitility.await().untilAsserted(() -> {
+ assertTrue(persistentDispatcher.redeliveryMessages.isEmpty());
+ assertTrue(persistentDispatcher.hasConsumersNeededNormalRead());
+ });
+
+ entry.release();
+ }
+
@Test(timeOut = 10000)
public void testAddConsumerWhenClosed() throws Exception {
persistentDispatcher.close().get();