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/06/24 19:34:40 UTC
(pulsar) branch master updated: [improve][broker] Improve dispatch performance by summing entry bytes with a loop (#26055)
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/pulsar.git
The following commit(s) were added to refs/heads/master by this push:
new b2f37560b356 [improve][broker] Improve dispatch performance by summing entry bytes with a loop (#26055)
b2f37560b356 is described below
commit b2f37560b356a011bcd9bdb7d37d8ce01d95ea98
Author: void-ptr974 <vo...@protonmail.com>
AuthorDate: Thu Jun 25 03:34:33 2026 +0800
[improve][broker] Improve dispatch performance by summing entry bytes with a loop (#26055)
---
...tractPersistentDispatcherMultipleConsumers.java | 10 +++
.../PersistentDispatcherMultipleConsumers.java | 2 +-
...rsistentDispatcherMultipleConsumersClassic.java | 2 +-
.../PersistentDispatcherTotalBytesTest.java | 78 ++++++++++++++++++++++
4 files changed, 90 insertions(+), 2 deletions(-)
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/AbstractPersistentDispatcherMultipleConsumers.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/AbstractPersistentDispatcherMultipleConsumers.java
index 79d365b9fee2..3a63dc659470 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/AbstractPersistentDispatcherMultipleConsumers.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/AbstractPersistentDispatcherMultipleConsumers.java
@@ -18,8 +18,10 @@
*/
package org.apache.pulsar.broker.service.persistent;
+import java.util.List;
import java.util.Map;
import org.apache.bookkeeper.mledger.AsyncCallbacks;
+import org.apache.bookkeeper.mledger.Entry;
import org.apache.bookkeeper.mledger.ManagedCursor;
import org.apache.pulsar.broker.ServiceConfiguration;
import org.apache.pulsar.broker.service.AbstractDispatcherMultipleConsumers;
@@ -64,4 +66,12 @@ public abstract class AbstractPersistentDispatcherMultipleConsumers extends Abst
public abstract Map<String, TopicMetricBean> getBucketDelayedIndexStats();
public abstract boolean isClassic();
+
+ static long getTotalBytesSize(List<Entry> entries) {
+ long totalBytesSize = 0;
+ for (int i = 0, entriesSize = entries.size(); i < entriesSize; i++) {
+ totalBytesSize += entries.get(i).getLength();
+ }
+ return totalBytesSize;
+ }
}
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherMultipleConsumers.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherMultipleConsumers.java
index 15a6071c23c4..5682b87f41bd 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherMultipleConsumers.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherMultipleConsumers.java
@@ -704,7 +704,7 @@ public class PersistentDispatcherMultipleConsumers extends AbstractPersistentDis
.attr("consumerCount", consumerList.size())
.log("Distributing messages to consumers");
- long totalBytesSize = entries.stream().mapToLong(Entry::getLength).sum();
+ long totalBytesSize = getTotalBytesSize(entries);
updatePendingBytesToDispatch(totalBytesSize);
// dispatch messages to a separate thread, but still in order for this subscription
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherMultipleConsumersClassic.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherMultipleConsumersClassic.java
index be1a29ad6bf1..1ae1683a486d 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherMultipleConsumersClassic.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherMultipleConsumersClassic.java
@@ -616,7 +616,7 @@ public class PersistentDispatcherMultipleConsumersClassic extends AbstractPersis
.attr("consumerCount", consumerList.size())
.log("Distributing messages to consumers");
- long size = entries.stream().mapToLong(Entry::getLength).sum();
+ long size = getTotalBytesSize(entries);
updatePendingBytesToDispatch(size);
// dispatch messages to a separate thread, but still in order for this subscription
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherTotalBytesTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherTotalBytesTest.java
new file mode 100644
index 000000000000..239a698ab011
--- /dev/null
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherTotalBytesTest.java
@@ -0,0 +1,78 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.broker.service.persistent;
+
+import static org.testng.Assert.assertEquals;
+import java.util.ArrayList;
+import org.apache.bookkeeper.mledger.Entry;
+import org.apache.bookkeeper.mledger.impl.EntryImpl;
+import org.testng.annotations.DataProvider;
+import org.testng.annotations.Test;
+
+@Test(groups = "broker")
+public class PersistentDispatcherTotalBytesTest {
+
+ @DataProvider(name = "entryCounts")
+ public Object[][] entryCounts() {
+ return new Object[][] {
+ {0},
+ {1},
+ {32},
+ {1024}
+ };
+ }
+
+ @Test(dataProvider = "entryCounts")
+ public void testGetTotalBytesSize(int entryCount) {
+ EntriesAndExpectedSize entries = entriesWithVaryingPayloadSizes(entryCount);
+ try {
+ assertEquals(AbstractPersistentDispatcherMultipleConsumers.getTotalBytesSize(entries.entries),
+ entries.expectedSize);
+ } finally {
+ entries.release();
+ }
+ }
+
+ private static EntriesAndExpectedSize entriesWithVaryingPayloadSizes(int entryCount) {
+ EntriesAndExpectedSize entries = new EntriesAndExpectedSize(entryCount);
+ for (int i = 0; i < entryCount; i++) {
+ int payloadSize = payloadSize(i);
+ entries.entries.add(EntryImpl.create(1, i, new byte[payloadSize]));
+ entries.expectedSize += payloadSize;
+ }
+ return entries;
+ }
+
+ private static int payloadSize(int index) {
+ return index % 97 == 0 ? 4096 : 1 + ((index * 31) & 1023);
+ }
+
+ private static final class EntriesAndExpectedSize {
+ private final ArrayList<Entry> entries;
+ private long expectedSize;
+
+ private EntriesAndExpectedSize(int entryCount) {
+ this.entries = new ArrayList<>(entryCount);
+ }
+
+ private void release() {
+ entries.forEach(Entry::release);
+ }
+ }
+}