Hi Yubiao, Thanks for the detailed proposal and for the reproduction. The problem is real, and having it documented this well helps a lot.
However, I'm requesting changes to PIP-499 and PR #27110, and I don't support moving the current solution forward as it is. The proposal adds accounting and throttling around a problem that is created in the broker's frame decoding, but it doesn't reduce the wasted memory. In the reproduction, more than half of the memory held by the buffers wasn't payload (938 of 1,587 MiB), and with the proposed solution it would still be wasted; the broker would only start throttling producers once it grows large enough. I think we should fix that root cause directly. Once the excess retention is gone, the existing maxMessagePublishBufferSizeInMB limiter bounds the memory, and no new accounting, metrics, throttling modes or PulsarFlowControlHandler changes are needed. What the heap dump shows I analyzed a heap dump of the broker in the PIP-499 reproduction (512 producers, 48 KiB payloads, 2 GiB direct memory, maxMessagePublishBufferSizeInMB=1024). The pooled allocator had allocated 1,988 MiB, and live buffers held 1,587 MiB. The 13,830 in-flight publish frames referenced only 649 MiB of that, so 938 MiB, 2.44x the payload, was overhead. The existing limiter counted the 649 MiB correctly and never engaged, because it was below its 1,024 MiB limit. Of the overhead, 701 MiB had never been written, and the rest were completed frames kept alive by later frames in the same buffer. Root cause The broker's receive buffers can grow to 1 MiB. When a partial frame remains and the cumulation buffer is shared by retained frames, Netty's MERGE_CUMULATOR replaces it in expandCumulation with a buffer rounded up to the next power of two. A partial frame plus a 1 MiB read becomes a 2 MiB buffer with about 1.05 MiB written; the dump has 493 of them (986 MiB). LengthFieldBasedFrameDecoder.extractFrame returns buffer.retainedSlice(index, length), so each 48 KiB publish keeps its whole buffer alive until BookKeeper acknowledges it. Netty allocator tuning doesn't fix this. The chunk size, arenas and caches of PooledByteBufAllocator don't change what is pinned, and Netty 4.2's AdaptiveByteBufAllocator also rounds large buffers to powers of two and doesn't pool allocations above 1 MiB. The buffer sizes are decided by the receive buffer allocator and the cumulator before the allocator is called. Three ways to address the root cause The numbers for 1 and 2 are calculated from the buffers in the heap dump. The dump can't show what other receive buffer settings would do, so the numbers for 3 come from a small simulation of Netty's cumulation logic with the same 48 KiB frames. To check that the simulation is realistic, I ran it with the current 1 MiB maximum: it gives a total buffer size of 2.28x the payload, close to the 2.44x measured in the dump. 1. Copy the frame in extractFrame. Extend LengthFieldBasedFrameDecoder and override extractFrame so that it copies the frame into its own buffer instead of returning a retained slice: https://github.com/netty/netty/blob/3feaf416ea8ca77d14c4b6fd5c2c7a5b043ca399/codec-base/src/main/java/io/netty/handler/codec/LengthFieldBasedFrameDecoder.java#L502-L507 The cumulation buffer is released immediately, so no memory is retained beyond the frames themselves. In the dump, the 13,830 frames would take 756 MiB instead of 1,587 MiB, 1.16x the payload. The remaining 107 MiB is the allocator's size-class rounding (49,228 bytes rounds to 57,344), which applies to any allocation. The existing limiter would then bound the real direct memory usage. The cost is one memory copy per frame. The broker already reads every publish payload once for the CRC32C check, so the copy is of the same order of cost. 2. A custom cumulator. LengthFieldBasedFrameDecoder inherits setCumulator from ByteToMessageDecoder, so the broker's frame decoder can replace the default MERGE_CUMULATOR with its own implementation. A custom cumulator can allocate exactly the needed size instead of the next power of two, and copy a mostly empty receive buffer into a right-sized buffer instead of adopting it as the cumulation. This removes most of the never-written memory while keeping frames zero-copy. Sizing each buffer in the dump to the bytes written into it gives 1,016 MiB instead of 1,587 MiB, reducing the overhead from 938 MiB to 367 MiB (61% less). The tradeoff is that frames are still slices of shared buffers, so completed frames can still be kept alive by later ones, and the existing limiter still undercounts the memory by about 1.6x. 3. Tunable receive buffer settings. The broker hardcodes the AdaptiveRecvByteBufAllocator parameters (minimum 1 KiB, initial 16 KiB, maximum 1 MiB), both for the broker's own listeners and for protocol handlers: https://github.com/apache/pulsar/blob/a996f4192d05c333944f64c1564e34dfb2187a41/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java#L657-L658 https://github.com/apache/pulsar/blob/a996f4192d05c333944f64c1564e34dfb2187a41/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java#L629-L630 PIP-430 already proposed making these configurable, but that part hasn't been implemented yet: https://github.com/apache/pulsar/blob/73c75fe5ae0fd1a663e38151efce2268c06f29cd/pip/pip-430.md?plain=1#L196-L224 In the simulation, lowering the maximum from 1 MiB to 64 KiB reduces the overhead by about 23%, and to 16 KiB by about 74%, at the cost of more read calls. Combined with the custom cumulator, a 16 KiB maximum brings the total down to about 1.2-1.3x the payload. The copy in extractFrame alone brings the overhead to the minimum, so it's the most important of the three. Having control over 2 and 3 is still valuable. They reduce the overhead for frames that stay zero-copy, and the receive buffer settings also control the number of read calls and the allocation sizes. Together they also make it possible to skip the copy in many cases. extractFrame receives the cumulation buffer and the frame's position, so it can return the retained slice without copying when the frame fills the whole buffer: it starts right after the length field, ends at the writer index, and the buffer's capacity isn't much larger than the frame. The slice then pins no more memory than a copy would. A custom cumulator that sizes the buffer for the pending frame, whose length is known from the length field, makes this the common case for frames that span several reads, so large frames stay zero-copy without overhead. Each of these changes is local to the frame decoder or the channel configuration and easy to put behind a configuration option, so they can be benchmarked against each other with the PIP-499 reproduction. Why not the accounting As mentioned above, the accounting doesn't reduce the wasted memory. In the default AUTO mode, the broker carries the full overhead on every allocation, plus the cost of tracking it, until direct memory usage reaches 85% and the tracked memory exceeds 1.5x the limit. Only then does it pause socket reads. That caps the total, but the waste remains, so producers would be throttled with less than half as much data in flight as the same memory could hold with the copy in extractFrame (2.44x vs 1.16x the payload). The PIP dismisses the copy because of its cost on the inbound path, but the accounting has costs of its own, and I'd expect them to be higher. In PR #27110 the tracking decoder runs in every mode except DISABLED, and the default is AUTO. (The PIP document still says it's always installed; it should be updated to match.) Every view of an inbound buffer is wrapped in an extra TrackedByteBuf object, which adds an allocation and a delegation layer for each frame and each slice derived from it on the whole publish path. Every retain and release performs a second atomic update on a per-buffer counter; in the dump one buffer was shared by up to 46 references, released from BookKeeper and managed ledger threads. In AUTO mode, every read and every final release also updates and sums a broker-wide LongAdder and checks a StampedLock. Synchronization, atomic updates and volatile reads and writes aren't free. An atomic read-modify-write acts as a full memory barrier and needs exclusive ownership of the cache line. When several cores update the same counter, the cache line moves between them on every update, which costs far more than an uncontended operation, and the barriers limit what the JIT compiler and the CPU can reorder. These costs grow with contention and core count, so they tend to show up on larger machines under high load rather than in small benchmarks. With the default AUTO mode, every broker pays them continuously, without any reduction in memory usage. The copy, by contrast, is a sequential copy of data that was just read and is most likely still in the CPU cache, done on the connection's own event loop thread without shared state. Copying a 48 KiB frame takes a few microseconds. This should be measured rather than assumed, but the copy is the simpler and more predictable cost. To summarize, I'm requesting changes: please revise PIP-499 and PR #27110 to address the root cause with these three approaches, so that the memory isn't wasted in the first place, instead of adding the request-frame memory accounting and throttling. I'm happy to help with the design and the benchmarks. -Lari
