You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by lh...@apache.org on 2026/10/02 14:53:17 UTC
(pulsar) 02/03: [fix][broker] Fix Exclusive/Failover dispatch rate limiting using counts from another dispatch (#26810)
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a commit to branch branch-4.0
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 3ecf2cee9e3643c6a750408fd833a5e9348df26e
Author: Matteo Merli <mm...@apache.org>
AuthorDate: Thu Oct 1 22:33:57 2026 -0700
[fix][broker] Fix Exclusive/Failover dispatch rate limiting using counts from another dispatch (#26810)
(cherry picked from commit 33dca0ed994718091659b254f4e2b4b0a652fd32)
---
.../PersistentDispatcherSingleActiveConsumer.java | 12 +++--
...rsistentDispatcherSingleActiveConsumerTest.java | 56 ++++++++++++++++++++++
2 files changed, 63 insertions(+), 5 deletions(-)
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherSingleActiveConsumer.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherSingleActiveConsumer.java
index e48db73db4f3..c961431d4b36 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherSingleActiveConsumer.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherSingleActiveConsumer.java
@@ -239,14 +239,16 @@ public class PersistentDispatcherSingleActiveConsumer extends AbstractDispatcher
protected void dispatchEntriesToConsumer(Consumer currentConsumer, List<Entry> entries,
EntryBatchSizes batchSizes, EntryBatchIndexesAcks batchIndexesAcks,
SendMessageInfo sendMessageInfo, long epoch) {
+ // sendMessageInfo is a thread-local instance that the next dispatch on this thread can overwrite before
+ // the write listener below runs on the connection's event loop, so capture the counts here.
+ final int totalMessages = sendMessageInfo.getTotalMessages();
+ final long totalBytes = sendMessageInfo.getTotalBytes();
currentConsumer
- .sendMessages(entries, batchSizes, batchIndexesAcks, sendMessageInfo.getTotalMessages(),
- sendMessageInfo.getTotalBytes(), sendMessageInfo.getTotalChunkedMessages(),
- redeliveryTracker, epoch)
+ .sendMessages(entries, batchSizes, batchIndexesAcks, totalMessages, totalBytes,
+ sendMessageInfo.getTotalChunkedMessages(), redeliveryTracker, epoch)
.addListener(future -> {
if (future.isSuccess()) {
- acquirePermitsForDeliveredMessages(topic, cursor, entries.size(),
- sendMessageInfo.getTotalMessages(), sendMessageInfo.getTotalBytes());
+ acquirePermitsForDeliveredMessages(topic, cursor, entries.size(), totalMessages, totalBytes);
// Schedule a new read batch operation only after the previous batch has been written to the socket.
executor.execute(() -> readMoreEntries(getActiveConsumer()));
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherSingleActiveConsumerTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherSingleActiveConsumerTest.java
index 10295d06dac8..d9ee0de339ad 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherSingleActiveConsumerTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherSingleActiveConsumerTest.java
@@ -18,6 +18,16 @@
*/
package org.apache.pulsar.broker.service.persistent;
+import static org.apache.pulsar.common.protocol.Commands.DEFAULT_CONSUMER_EPOCH;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyInt;
+import static org.mockito.ArgumentMatchers.anyLong;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+import io.netty.util.concurrent.ImmediateEventExecutor;
+import io.netty.util.concurrent.Promise;
+import java.util.List;
import java.util.Optional;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutionException;
@@ -28,6 +38,7 @@ import java.util.concurrent.atomic.AtomicReference;
import lombok.Cleanup;
import lombok.extern.slf4j.Slf4j;
import org.apache.bookkeeper.mledger.AsyncCallbacks;
+import org.apache.bookkeeper.mledger.Entry;
import org.apache.bookkeeper.mledger.ManagedCursor;
import org.apache.bookkeeper.mledger.ManagedLedgerException;
import org.apache.bookkeeper.mledger.impl.ManagedCursorImpl;
@@ -35,8 +46,12 @@ import org.apache.pulsar.broker.BrokerTestUtil;
import org.apache.pulsar.broker.intercept.MockBrokerInterceptor;
import org.apache.pulsar.broker.service.BrokerServiceException;
import org.apache.pulsar.broker.service.Consumer;
+import org.apache.pulsar.broker.service.EntryBatchIndexesAcks;
+import org.apache.pulsar.broker.service.EntryBatchSizes;
+import org.apache.pulsar.broker.service.SendMessageInfo;
import org.apache.pulsar.broker.service.ServerCnx;
import org.apache.pulsar.broker.service.Subscription;
+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.PulsarClient;
@@ -143,6 +158,47 @@ public class PersistentDispatcherSingleActiveConsumerTest extends ProducerConsum
admin.topics().delete(topicName, false);
}
+ @Test
+ public void testDispatchQuotaWhenSendMessageInfoReusedBeforeWriteCompletes() throws Exception {
+ String topicName = BrokerTestUtil.newUniqueName(
+ "persistent://public/default/testDispatchQuotaWhenSendMessageInfoReused");
+ String subscription = "s1";
+ admin.topics().createNonPartitionedTopic(topicName);
+ admin.topics().createSubscription(topicName, subscription, MessageId.earliest);
+ PersistentTopic topic =
+ (PersistentTopic) pulsar.getBrokerService().getTopic(topicName, false).join().orElseThrow();
+ PersistentSubscription sub = topic.getSubscription(subscription);
+ DispatchRateLimiter rateLimiter = mock(DispatchRateLimiter.class);
+ PersistentDispatcherSingleActiveConsumer dispatcher =
+ new PersistentDispatcherSingleActiveConsumer(sub.getCursor(),
+ CommandSubscribe.SubType.Exclusive, 0, topic, sub) {
+ @Override
+ public Optional<DispatchRateLimiter> getRateLimiter() {
+ return Optional.of(rateLimiter);
+ }
+ };
+ // The write to the consumer connection stays pending until the test completes it.
+ Promise<Void> writePromise = ImmediateEventExecutor.INSTANCE.newPromise();
+ Consumer consumer = mock(Consumer.class);
+ when(consumer.sendMessages(any(), any(), any(), anyInt(), anyLong(), anyLong(), any(), anyLong()))
+ .thenReturn(writePromise);
+
+ SendMessageInfo sendMessageInfo = SendMessageInfo.getThreadLocal();
+ sendMessageInfo.setTotalMessages(3);
+ sendMessageInfo.setTotalBytes(300);
+ dispatcher.dispatchEntriesToConsumer(consumer, List.of(mock(Entry.class)), EntryBatchSizes.get(1),
+ EntryBatchIndexesAcks.get(1), sendMessageInfo, DEFAULT_CONSUMER_EPOCH);
+
+ // Another dispatch on this thread reuses the thread-local SendMessageInfo before the write completes.
+ SendMessageInfo nextSendMessageInfo = SendMessageInfo.getThreadLocal();
+ nextSendMessageInfo.setTotalMessages(7);
+ nextSendMessageInfo.setTotalBytes(700);
+ writePromise.setSuccess(null);
+
+ // The dispatch quota must be charged with the counts of the messages that were written.
+ verify(rateLimiter).consumeDispatchQuota(3, 300);
+ }
+
@DataProvider
public static Object[][] closeDelayMs() {
return new Object[][] { { 500 }, { 2000 } };