I'm good to only revert from branch-4.0 too. Thanks, Wenzhi Feng.
On 2025/06/19 00:36:18 PengHui Li wrote: > > the main problem is that this kind of optimization should never have been > cherry-picked into a stable branch. > > > We should be way more strict in only backporting bug-fixes / > security-fixes > into stable branches. Every other improvement or feature will roll into the > next feature release. > > +1, I'm good to only revert from branch-4.0 and keep working on the fix > on the master branch > > Thanks, > penghui > > On Wed, Jun 18, 2025 at 5:06 PM Matteo Merli <matteo.me...@gmail.com> wrote: > > > Hi WenZhi, > > > > the main problem is that this kind of optimization should never have been > > cherry-picked into a stable branch. > > > > We should be way more strict in only backporting bug-fixes / security-fixes > > into stable branches. Every other improvement or feature will roll into the > > next feature release. > > > > > > -- > > Matteo Merli > > <matteo.me...@gmail.com> > > > > > > On Wed, Jun 18, 2025 at 4:42 PM WenZhi Feng <thetumb...@apache.org> wrote: > > > > > Hi Penghui, > > > Thanks for your reply and the evidence provided. We can fix this problem > > > by improving the implementation of method `getNumberOfDelayedMessages()`, > > > which is a simple patch. I can help to do this. > > > > > > thanks, > > > Wenzhi Feng. > > > > > > On 2025/06/18 23:26:08 PengHui Li wrote: > > > > Hi Wenzhi, > > > > > > > > After upgrading our broker environment from version 3.0.x to 4.0.x, > > > > we observed a critical performance regression. > > > > The broker's CPU usage has consistently been more than three times > > > > higher than pre-upgrade levels, causing it to hit its allocated CPU > > > limit. > > > > > > > > The stacktrace from the process during this high-utilization period is > > > > attached below for diagnostics. > > > > > > > > ``` > > > > "broker-topic-workers-OrderedExecutor-1-0" #51 [95] prio=5 os_prio=0 > > > > cpu=1678706.57ms elapsed=15712.43s tid=0x00007fa0da32b820 nid=95 > > runnable > > > > [0x00007fa0d6c00000] > > > > java.lang.Thread.State: RUNNABLE > > > > at java.util.stream.AbstractPipeline.wrapSink(java.base@21.0.7 > > > /Unknown > > > > Source) > > > > at > > > > java.util.stream.AbstractPipeline.wrapAndCopyInto(java.base@21.0.7 > > > /Unknown > > > > Source) > > > > at > > > > java.util.stream.ReduceOps$ReduceOp.evaluateSequential(java.base@21.0.7 > > > /Unknown > > > > Source) > > > > at java.util.stream.AbstractPipeline.evaluate(java.base@21.0.7 > > > /Unknown > > > > Source) > > > > at java.util.stream.LongPipeline.reduce(java.base@21.0.7 > > /Unknown > > > > Source) > > > > at java.util.stream.LongPipeline.sum(java.base@21.0.7/Unknown > > > > Source) > > > > at > > > > > > > > > org.apache.pulsar.broker.delayed.InMemoryDelayedDeliveryTracker.lambda$getNumberOfDelayedMessages$3(InMemoryDelayedDeliveryTracker.java:231) > > > > at > > > > > > > > > org.apache.pulsar.broker.delayed.InMemoryDelayedDeliveryTracker$$Lambda/0x0000000100e73078.applyAsLong(Unknown > > > > Source) > > > > at > > java.util.stream.ReferencePipeline$5$1.accept(java.base@21.0.7 > > > /Unknown > > > > Source) > > > > at java.util.Iterator.forEachRemaining(java.base@21.0.7 > > /Unknown > > > > Source) > > > > at > > > > > > > > > it.unimi.dsi.fastutil.objects.ObjectSpliterators$SpliteratorFromIterator.forEachRemaining(ObjectSpliterators.java:1158) > > > > at java.util.stream.AbstractPipeline.copyInto(java.base@21.0.7 > > > /Unknown > > > > Source) > > > > at > > > > java.util.stream.AbstractPipeline.wrapAndCopyInto(java.base@21.0.7 > > > /Unknown > > > > Source) > > > > at > > > > java.util.stream.ReduceOps$ReduceOp.evaluateSequential(java.base@21.0.7 > > > /Unknown > > > > Source) > > > > at java.util.stream.AbstractPipeline.evaluate(java.base@21.0.7 > > > /Unknown > > > > Source) > > > > at java.util.stream.LongPipeline.reduce(java.base@21.0.7 > > /Unknown > > > > Source) > > > > at java.util.stream.LongPipeline.sum(java.base@21.0.7/Unknown > > > > Source) > > > > at > > > > > > > > > org.apache.pulsar.broker.delayed.InMemoryDelayedDeliveryTracker.getNumberOfDelayedMessages(InMemoryDelayedDeliveryTracker.java:231) > > > > at > > > > > > > > > org.apache.pulsar.broker.delayed.AbstractDelayedDeliveryTracker.updateTimer(AbstractDelayedDeliveryTracker.java:91) > > > > at > > > > > > > > > org.apache.pulsar.broker.delayed.InMemoryDelayedDeliveryTracker.addMessage(InMemoryDelayedDeliveryTracker.java:128) > > > > at > > > > > > > > > org.apache.pulsar.broker.service.persistent.PersistentDispatcherMultipleConsumers.trackDelayedDelivery(PersistentDispatcherMultipleConsumers.java:1314) > > > > - locked <0x00000000c21d88b8> (a > > > > > > > > > org.apache.pulsar.broker.service.persistent.PersistentDispatcherMultipleConsumers) > > > > at > > > > > > > > > org.apache.pulsar.broker.service.AbstractBaseDispatcher.filterEntriesForConsumer(AbstractBaseDispatcher.java:228) > > > > at > > > > > > > > > org.apache.pulsar.broker.service.persistent.PersistentDispatcherMultipleConsumers.trySendMessagesToConsumers(PersistentDispatcherMultipleConsumers.java:864) > > > > - locked <0x00000000c21d88b8> (a > > > > > > > > > org.apache.pulsar.broker.service.persistent.PersistentDispatcherMultipleConsumers) > > > > at > > > > > > > > > org.apache.pulsar.broker.service.persistent.PersistentDispatcherMultipleConsumers.sendMessagesToConsumers(PersistentDispatcherMultipleConsumers.java:769) > > > > - eliminated <0x00000000c21d88b8> (a > > > > > > > > > org.apache.pulsar.broker.service.persistent.PersistentDispatcherMultipleConsumers) > > > > at > > > > > > > > > org.apache.pulsar.broker.service.persistent.PersistentDispatcherMultipleConsumers.handleSendingMessagesAndReadingMore(PersistentDispatcherMultipleConsumers.java:729) > > > > - locked <0x00000000c21d88b8> (a > > > > > > > > > org.apache.pulsar.broker.service.persistent.PersistentDispatcherMultipleConsumers) > > > > at > > > > > > > > > org.apache.pulsar.broker.service.persistent.PersistentDispatcherMultipleConsumers.lambda$readEntriesComplete$9(PersistentDispatcherMultipleConsumers.java:719) > > > > at > > > > > > > > > org.apache.pulsar.broker.service.persistent.PersistentDispatcherMultipleConsumers$$Lambda/0x0000000100c0f440.run(Unknown > > > > Source) > > > > at > > > > > > > > > org.apache.bookkeeper.common.util.SingleThreadExecutor.safeRunTask(SingleThreadExecutor.java:128) > > > > at > > > > > > > > > org.apache.bookkeeper.common.util.SingleThreadExecutor.run(SingleThreadExecutor.java:99) > > > > at > > > > > > > > > io.netty.util.concurrent.FastThreadLocalRunnable.run(FastThreadLocalRunnable.java:30) > > > > at java.lang.Thread.runWith(java.base@21.0.7/Unknown Source) > > > > at java.lang.Thread.run(java.base@21.0.7/Unknown Source) > > > > > > > > "broker-topic-workers-OrderedExecutor-2-0" #52 [96] prio=5 os_prio=0 > > > > cpu=1797445.47ms elapsed=15712.43s tid=0x00007fa0d9419fc0 nid=96 > > runnable > > > > [0x00007fa0d6aff000] > > > > java.lang.Thread.State: RUNNABLE > > > > at java.util.stream.AbstractPipeline.copyInto(java.base@21.0.7 > > > /Unknown > > > > Source) > > > > at > > > > java.util.stream.AbstractPipeline.wrapAndCopyInto(java.base@21.0.7 > > > /Unknown > > > > Source) > > > > at > > > > java.util.stream.ReduceOps$ReduceOp.evaluateSequential(java.base@21.0.7 > > > /Unknown > > > > Source) > > > > at java.util.stream.AbstractPipeline.evaluate(java.base@21.0.7 > > > /Unknown > > > > Source) > > > > at java.util.stream.LongPipeline.reduce(java.base@21.0.7 > > /Unknown > > > > Source) > > > > at java.util.stream.LongPipeline.sum(java.base@21.0.7/Unknown > > > > Source) > > > > at > > > > > > > > > org.apache.pulsar.broker.delayed.InMemoryDelayedDeliveryTracker.lambda$getNumberOfDelayedMessages$3(InMemoryDelayedDeliveryTracker.java:231) > > > > at > > > > > > > > > org.apache.pulsar.broker.delayed.InMemoryDelayedDeliveryTracker$$Lambda/0x0000000100e73078.applyAsLong(Unknown > > > > Source) > > > > at > > java.util.stream.ReferencePipeline$5$1.accept(java.base@21.0.7 > > > /Unknown > > > > Source) > > > > at java.util.Iterator.forEachRemaining(java.base@21.0.7 > > /Unknown > > > > Source) > > > > at > > > > > > > > > it.unimi.dsi.fastutil.objects.ObjectSpliterators$SpliteratorFromIterator.forEachRemaining(ObjectSpliterators.java:1158) > > > > at java.util.stream.AbstractPipeline.copyInto(java.base@21.0.7 > > > /Unknown > > > > Source) > > > > at > > > > java.util.stream.AbstractPipeline.wrapAndCopyInto(java.base@21.0.7 > > > /Unknown > > > > Source) > > > > at > > > > java.util.stream.ReduceOps$ReduceOp.evaluateSequential(java.base@21.0.7 > > > /Unknown > > > > Source) > > > > at java.util.stream.AbstractPipeline.evaluate(java.base@21.0.7 > > > /Unknown > > > > Source) > > > > at java.util.stream.LongPipeline.reduce(java.base@21.0.7 > > /Unknown > > > > Source) > > > > at java.util.stream.LongPipeline.sum(java.base@21.0.7/Unknown > > > > Source) > > > > at > > > > > > > > > org.apache.pulsar.broker.delayed.InMemoryDelayedDeliveryTracker.getNumberOfDelayedMessages(InMemoryDelayedDeliveryTracker.java:231) > > > > at > > > > > > > > > org.apache.pulsar.broker.delayed.AbstractDelayedDeliveryTracker.updateTimer(AbstractDelayedDeliveryTracker.java:91) > > > > at > > > > > > > > > org.apache.pulsar.broker.delayed.InMemoryDelayedDeliveryTracker.getScheduledMessages(InMemoryDelayedDeliveryTracker.java:217) > > > > at > > > > > > > > > org.apache.pulsar.broker.service.persistent.PersistentDispatcherMultipleConsumers.getMessagesToReplayNow(PersistentDispatcherMultipleConsumers.java:1327) > > > > - locked <0x00000000b8cd7968> (a > > > > > > > > > org.apache.pulsar.broker.service.persistent.PersistentDispatcherMultipleConsumers) > > > > at > > > > > > > > > org.apache.pulsar.broker.service.persistent.PersistentDispatcherMultipleConsumers.readMoreEntries(PersistentDispatcherMultipleConsumers.java:386) > > > > - locked <0x00000000b8cd7968> (a > > > > > > > > > org.apache.pulsar.broker.service.persistent.PersistentDispatcherMultipleConsumers) > > > > at > > > > > > > > > org.apache.pulsar.broker.service.persistent.PersistentDispatcherMultipleConsumers.handleSendingMessagesAndReadingMore(PersistentDispatcherMultipleConsumers.java:743) > > > > - locked <0x00000000b8cd7968> (a > > > > > > > > > org.apache.pulsar.broker.service.persistent.PersistentDispatcherMultipleConsumers) > > > > at > > > > > > > > > org.apache.pulsar.broker.service.persistent.PersistentDispatcherMultipleConsumers.lambda$readEntriesComplete$9(PersistentDispatcherMultipleConsumers.java:719) > > > > at > > > > > > > > > org.apache.pulsar.broker.service.persistent.PersistentDispatcherMultipleConsumers$$Lambda/0x0000000100c0f440.run(Unknown > > > > Source) > > > > at > > > > > > > > > org.apache.bookkeeper.common.util.SingleThreadExecutor.safeRunTask(SingleThreadExecutor.java:128) > > > > at > > > > > > > > > org.apache.bookkeeper.common.util.SingleThreadExecutor.run(SingleThreadExecutor.java:99) > > > > at > > > > > > > > > io.netty.util.concurrent.FastThreadLocalRunnable.run(FastThreadLocalRunnable.java:30) > > > > at java.lang.Thread.runWith(java.base@21.0.7/Unknown Source) > > > > at java.lang.Thread.run(java.base@21.0.7/Unknown Source) > > > > ``` > > > > The flamegraph and stacktrace I provided are the primary evidence > > > > we've captured for the performance regression. > > > > > > > > > > > > > https://github.com/apache/pulsar/pull/23611/files#diff-f159b4e262ff6213bba19d20e6dc01d07ea8c19f4675524e4ceb1470f456e8fcR229-R231 > > > > Here is the change that caused this issue. Each message added to the > > > > DelayedMessageTracker will call `getNumberOfDelayedMessages()` method > > > > which will iterate the map defined in InMemoryDelayedDeliveryTracker > > > > > > > > ``` > > > > > > > > protected final > > > Long2ObjectSortedMap<Long2ObjectSortedMap<Roaring64Bitmap>> > > > > delayedMessageMap = new Long2ObjectAVLTreeMap<>(); > > > > > > > > ``` > > > > > > > > Regards > > > > > > > > Penghui > > > > > > > > > > > > On Wed, Jun 18, 2025 at 4:01 PM WenZhi Feng <thetumb...@apache.org> > > > wrote: > > > > > > > > > -1 until much clearer proof is shown. > > > > > > > > > > On 2025/06/18 22:26:26 PengHui Li wrote: > > > > > > Hi team, > > > > > > > > > > > > I am writing to propose an urgent revert of PR #23611 > > > > > > > > > > > > - https://github.com/apache/pulsar/pull/23611 > > > > > > > > > > > > *Reason for Revert:* > > > > > > > > > > > > This PR has introduced a significant performance regression in the > > > Pulsar > > > > > > broker. > > > > > > The attached flame graph visually demonstrates increased CPU > > > utilization > > > > > > and time > > > > > > spent in the code paths related to DelayedDeliveryTracker and > > stream > > > > > > operations. > > > > > > > > > > > > While the intention was to optimize memory usage, the current > > > > > > implementation > > > > > > appears to have an adverse effect on CPU performance, leading to > > > overall > > > > > > degraded > > > > > > broker throughput and increased latency. > > > > > > > > > > > > *Impact:* > > > > > > > > > > > > This regression is impacting the stability and performance of our > > > Pulsar > > > > > > clusters, > > > > > > especially when you have large-scale delayed messages. Reverting > > the > > > > > change > > > > > > will > > > > > > allow us to restore the previous performance characteristics while > > we > > > > > > investigate > > > > > > a more robust and performant solution for DelayedDeliveryTracker > > > memory > > > > > > optimization. > > > > > > > > > > > > *Proposed Action:* > > > > > > > > > > > > I propose to revert PR #23611 as soon as possible to reduce the > > risk > > > to > > > > > > other users, > > > > > > and we can then collectively work on a more thoroughly tested and > > > > > > performant approach > > > > > > to optimize the DelayedDeliveryTracker memory. > > > > > > > > > > > > Please let me know your thoughts, and if there are any immediate > > > concerns > > > > > > with this proposed revert. > > > > > > > > > > > > Regards, > > > > > > Penghui > > > > > > > > > > > > > > > > > > > > >
