[improve][broker] Optimize SegmentedLongArray with heap-backed long[][]

[nodece]

Motivation

SegmentedLongArray previously used Netty ByteBuf as its backing store. Every readLong/writeLong operation went through ByteBuf.getLong()/setLong(), adding extra indirection on the hot path of TripleLongPriorityQueue heap operations(siftUp/siftDown).

PR #26010 identified SegmentedLongArray as the remaining bottleneck after optimizing the heap algorithms, and suggested replacing the backing storage with primitive arrays.

This PR replaces the ByteBuf backing store with segmented heap-backed long[][] arrays while preserving the segmented architecture required to support capacities beyond Integer.MAX_VALUE elements (PR #16490).

Modifications

• SegmentedLongArray
  • Replace the ByteBuf backing store with segmented heap-backed long[][] arrays.
  • Use bit-shift/bit-mask for segment lookup instead of division/modulo.
  • Grow the outer segment array using an amortized expansion strategy to avoid copying on every new segment allocation.

Benchmark Results

JMH benchmarks from TripleLongPriorityQueueBenchmark(2 forks, 3 warmup iterations, 5 measurement iterations, JDK 21.0.5).

Before (ByteBuf)

Benchmark	Size	Score (us/op)	Error
recoveryBulkAddThenPop	500,000	454,423	±142,390
recoveryBulkAddThenPop	2,000,000	2,406,656	±680,108
interleavedAddPop	500,000	335,764	±461,320
interleavedAddPop	2,000,000	1,560,565	±403,289
steadyState	500,000	261,449	±11,931
steadyState	2,000,000	1,148,007	±699,420

After (heap-backed long[][])

Benchmark	Size	Score (us/op)	Error
recoveryBulkAddThenPop	500,000	337,634	±6,460
recoveryBulkAddThenPop	2,000,000	1,805,059	±99,786
interleavedAddPop	500,000	231,696	±1,425
interleavedAddPop	2,000,000	1,102,539	±89,740
steadyState	500,000	261,390	±7,856
steadyState	2,000,000	1,054,573	±58,502

Comparison (Size = 2,000,000)

Benchmark	Before (ms)	After (ms)	Improvement
recoveryBulkAddThenPop	2,407	1,805	+25%
interleavedAddPop	1,561	1,103	+29%
steadyState	1,148	1,055	+8%

Note: The baseline measurements showed relatively large variance in some
scenarios, while the new implementation produced noticeably more stable
results. The reported improvements should therefore be interpreted together
with the benchmark error margins.

[void-ptr974]

This looks like a capacity growth contract issue.

SegmentedLongArray uses fixed-size segment addressing:

segmentIndex = offset >>> SEGMENT_SHIFT;
indexInSegment = offset & SEGMENT_MASK;

With this addressing scheme, every segment before the last one must be fully allocated. Otherwise a logical offset can map into a partial middle segment.

For example, with an initial capacity of 3 * 1024 * 1024 longs:

segment[0] = 2M longs
segment[1] = 1M longs
capacity   = 3M longs

After increaseCapacity() appends a new full segment:

segment[0] = 2M longs
segment[1] = 1M longs
segment[2] = 2M longs
capacity   = 5M longs

Offsets in [3M, 4M) are now within the reported capacity, but still map to segment[1], whose physical length is only 1M. A valid write can therefore fail:

SegmentedLongArray a = new SegmentedLongArray(3 * 1024 * 1024);
a.increaseCapacity();

a.writeLong(3 * 1024 * 1024, 1L); // within capacity, but fails

There is also a related caller-side issue: TripleLongPriorityQueue.add() calls increaseCapacity() once and then writes 3 longs for one tuple. If increaseCapacity() only expands a partial last segment to SEGMENT_SIZE, it may still not provide enough room for the pending 3-long write when the capacity is just below a segment boundary.

It would be safer for the growth API to express the required target capacity, for example:

array.ensureCapacity(arrayIdx + ITEMS_COUNT);

and have SegmentedLongArray guarantee:

capacity >= requiredCapacity
all segments before the last segment are full-sized

Alternatively, the caller would need to keep growing in a loop until the pending write is guaranteed to fit.

[void-ptr974]

This is a good CPU-side improvement. One concern I have is the GC impact from moving the backing storage from Netty direct memory to Java heap.

The logical storage size is roughly the same, but large queues will now use heap long[] segments. This may add sizable long-lived heap allocations and, under G1, could also involve humongous allocations.

It would be good to have an estimate of the GC impact in realistic broker-side scenarios. It may also be worth evaluating whether the existing segment size is still appropriate when the backing storage is on-heap.

[nodece]

@void-ptr974

Thanks for your GC concern. I analyzed production metric CSV files under realistic broker load (50k msg/s, 30M ledger entries), with full charts attached:

1. Zero G1 humongous allocations, Eden usage fluctuates normally.
2. Massive heap reclamation (over 2GiB freed) after every seal shows temporary objects get cleaned up efficiently by GC; Survivor region recycling is healthy.
3. CPU drop post-seal comes from reduced allocation pressure instead of GC STW pauses, no abnormal GC stalls observed.
4. Primitive long[] adds negligible GC marking overhead without reference tracing cost.

Two seal events captured during a 16-minute window:

Event	Time	Heap	CPU User
seal-1	09:46:40	3.52 GiB → 1.42 GiB (−2.1 GiB / 16s)	22.9% → 14.6%
seal-2	09:56:00	3.62 GiB → 1.33 GiB (−2.29 GiB / 13s)	22.3% → 19.3%

CPU:

GC:

[void-ptr974]

Thanks for addressing the capacity growth issue.

The latest version full-allocates every segment in allocateSegments():

segments[i] = new long[SEGMENT_SIZE];

This avoids partial middle segments, but it changes the default memory behavior significantly. A default TripleLongPriorityQueue starts with only 16 logical entries:

16 entries * 3 longs = 48 logical longs

but the underlying SegmentedLongArray would allocate a full segment:

2,097,152 longs = 16 MiB

This is expensive for the default path. In bucket delayed delivery there can be multiple TripleLongPriorityQueue instances per active delayed subscription, so the baseline heap cost can add up before the queues actually grow large. For example, two queues per delayed subscription would already be around 32 MiB of heap allocation.

There is also a memory accounting issue. bytesCapacity() still returns capacity * Long.BYTES, so the default queue would report only:

48 longs * 8 = 384 bytes

while physically allocating a 16 MiB long[]. That means delayedMessageIndexSizeInBytes may under-report the actual heap used by these queues.

A safer approach would be to keep the initial allocation proportional to the logical initial capacity, and fix the growth issue with an explicit ensureCapacity(requiredCapacity) API. For example, TripleLongPriorityQueue.add() can request enough capacity for the pending tuple:

array.ensureCapacity(arrayIdx + ITEMS_COUNT);

Then SegmentedLongArray can grow while preserving:

capacity >= requiredCapacity
all segments before the last segment are full-sized
the last segment may be partial

For example:

initial capacity = 3M longs
segments = [2M, 1M]

ensureCapacity(5M)
segments = [2M, 2M, 1M]

This fixes the segment mapping issue without forcing small/default queues to allocate a 16 MiB heap array upfront, and keeps memory accounting aligned with the actual backing allocation.

It may also be worth re-evaluating whether SEGMENT_SIZE = 2M longs is still the right segment size for on-heap storage. The 16 MiB segment size was reasonable for direct memory, but on heap each full segment becomes a large long[] object.

Separately, delayed index memory appears to be exposed mainly as a metric today, rather than being used as a hard limit or backpressure signal. This does not need to be solved in this PR, but if this storage moves on-heap, it may be worth discussing as a follow-up whether delayed tracker memory should have an explicit limit or backpressure policy.

[nodece]

@void-ptr974 Could you have a chance to review this PR?

[void-ptr974]

Thanks for the updates. The current ensureCapacity(required) approach looks much safer and fixes the default-footprint issue.

One remaining question is whether SEGMENT_SIZE should still be 2 * 1024 * 1024 longs. From PR #16490, segments were mainly introduced to avoid the single ByteBuf 2GB limit and avoid copying a huge buffer. I do not see a strict requirement that each segment must stay at 16MiB.

Now each full segment is a 16MiB long[] on heap. I think it is worth validating a smaller power-of-two size, such as 512K longs (4MiB) or 256K longs (2MiB). Small queues should not be affected much because allocation is now exact, while medium/large queues may get better allocation granularity and lower GC pressure. The tradeoff is more segment allocations, so it would be helpful to compare hot-path performance and GC metrics.

Also, some unit tests allocate many real segments, e.g. seg * 30 is about 480MiB. Could we reduce that in regular unit tests?

[nodece]

@void-ptr974 Glad the ensureCapacity(required) approach addresses the default-footprint concern.

Regarding SEGMENT_SIZE, I kept the current 2M-long segment size to stay consistent with the previous implementation. The original value was inherited during the migration from ByteBuf to heap arrays, and there is no fundamental correctness requirement that the segment size must remain 2M longs.

I agree this is a tunable parameter. Smaller segments can improve allocation granularity, while larger segments reduce the number of segments and resizing operations. The current segment size has been evaluated with our existing workloads and the GC behavior is acceptable. If real-world workloads show allocation pressure or segment management becoming a bottleneck, we can revisit this constant in a separate change with dedicated benchmarks.

For the test memory footprint, I added a package-private @VisibleForTesting constructor that allows tests to use a custom segment size while keeping the production default unchanged at 2M longs. Multi-segment grow/shrink tests now use 1024-long segments. The existing seg * 30 coverage is preserved, including the shrink-trim path, while reducing allocation from ~480 MiB to ~240 KiB.

[void-ptr974]

Thanks for sharing the GC data. The 50k msg/s / 30M ledger entries result is useful. I would be interested in the number of consumer groups and the peak delayed backlog per group/broker for that run.

Could you also add a few heavier cases — lots of consumer groups with a large backlog, consumer catch-up, and broker recovery? That would make it easier to judge the impact at the upper end.

[nodece]

Could you also add a few heavier cases — lots of consumer groups with a large backlog, consumer catch-up, and broker recovery? That would make it easier to judge the impact at the upper end.

@void-ptr974 Regarding the long[][] replacement of the previous buffer-based implementation, the main goal of this change was to reduce the memory overhead and GC pressure in the delayed message sorting path.

With this change, for the 30M delayed entries case, the full process of sorting the entries and writing them back to BookKeeper can complete within around 30 seconds. The previous implementation had higher overhead due to the buffer-based storage approach.

That said, the data structure optimization only addresses the local processing cost. There are still larger architectural optimization opportunities. For example, when multiple consumer groups share the same subscription state, snapshot operations can introduce lock contention. A snapshot may need to wait for another thread holding the lock to finish before proceeding. This is an area we should optimize in the long term by reducing blocking during snapshot and improving concurrency between different operations.

So the current change improves the sorting/storage efficiency significantly, while snapshot concurrency and large-scale consumer group scenarios are follow-up areas that need further design and optimization.

[nodece]

Additionally, I think this optimization direction is consistent with the existing approach in the broker codebase. We already use fastutil extensively to avoid unnecessary object allocation, boxing overhead, and reduce memory footprint in performance-sensitive paths.

Replacing the buffer-based structure with a primitive-based long[][] implementation follows the same principle. It keeps the data layout simple, reduces memory overhead, and avoids unnecessary GC pressure. Given that this path is a hot path for large delayed message workloads, I think this is the right direction for the optimization.

[dao-jun]

LGTM

[void-ptr974]

Thanks for the clarification. I wasn't questioning the optimization direction. For a pure delayed-delivery topic, every message is tracked independently by each subscription. With the 50k msg/s workload and the default 10-minute minimum ledger rollover interval, one subscription can accumulate roughly 30M delayed entries before the bucket is sealed, which is about 688MiB of queue backing with this implementation.

For a topic with 20 independent subscriptions, that would be roughly 13.4GiB of queue backing if they reach the same peak. This is why I was concerned about the per-subscription memory cost at larger scale.