This is an automated email from the ASF dual-hosted git repository. lhotari pushed a commit to branch branch-4.2 in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 9cf8a613f76a8e97678b7961ee63fc5d6e96e447 Author: void-ptr974 <[email protected]> AuthorDate: Thu Jun 25 03:34:33 2026 +0800 [improve][broker] Improve dispatch performance by summing entry bytes with a loop (#26055) (cherry picked from commit b2f37560b356a011bcd9bdb7d37d8ce01d95ea98) --- ...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 79d365b9fee..3a63dc65947 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 c855550069e..71e822e6dd6 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 @@ -707,7 +707,7 @@ public class PersistentDispatcherMultipleConsumers extends AbstractPersistentDis log.debug("[{}] Distributing {} messages to {} consumers", name, entries.size(), consumerList.size()); } - 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 59f2ce973da..7c35e626dc4 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 @@ -621,7 +621,7 @@ public class PersistentDispatcherMultipleConsumersClassic extends AbstractPersis log.debug("[{}] Distributing {} messages to {} consumers", name, entries.size(), consumerList.size()); } - 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 00000000000..239a698ab01 --- /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); + } + } +}
