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/05 18:42:18 UTC
(pulsar) 01/07: [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-5.0
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 3a6def9a110a35b455186ab005b11b7f59864c11
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 | 53 ++++++++++++++++++++++
2 files changed, 60 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 360c8dfad15a..e0fd3e4418ad 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
@@ -246,14 +246,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 de95404425c6..fe0d1aa60b8e 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,7 +18,17 @@
*/
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.Collections;
+import java.util.List;
import java.util.Optional;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CountDownLatch;
@@ -29,12 +39,16 @@ import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicReference;
import lombok.Cleanup;
import lombok.CustomLog;
+import org.apache.bookkeeper.mledger.Entry;
import org.apache.bookkeeper.mledger.ManagedLedgerException;
import org.apache.pulsar.broker.intercept.BrokerInterceptor;
import org.apache.pulsar.broker.intercept.MockBrokerInterceptor;
import org.apache.pulsar.broker.service.BrokerService;
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.SharedPulsarBaseTest;
import org.apache.pulsar.broker.service.SharedPulsarCluster;
@@ -76,6 +90,45 @@ public class PersistentDispatcherSingleActiveConsumerTest extends SharedPulsarBa
admin.topics().delete(topicName, false);
}
+ @Test
+ public void testDispatchQuotaWhenSendMessageInfoReusedBeforeWriteCompletes() throws Exception {
+ String topicName = newTopicName();
+ String subscription = "s1";
+ admin.topics().createNonPartitionedTopic(topicName);
+ admin.topics().createSubscription(topicName, subscription, MessageId.earliest);
+ PersistentTopic topic = (PersistentTopic) 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 } };