[ISSUE #10525] Reduce pull/dispatch path allocation via primitive arrays, ThreadLocal reuse, and lambda elimination

[wang-jiahua]

Which Issue(s) This PR Fixes

Fixes #10525

Brief Description

Reduce allocation in the pull/dispatch path through five independent optimizations:

1. GetMessageResult — Replace List<Long> with long[] + addQueueOffset(long) method. Eliminates Long boxing and ArrayList resize overhead. Right-sized initial capacity via constructor parameter (capped at 256 to prevent OOM).
2. DispatchRequest — Change final fields to mutable + add reset() method for ThreadLocal reuse. Eliminates per-message DispatchRequest allocation in the dispatch hot path.
3. DefaultMappedFile — Cache append slice and merge dual-slice operation in selectMappedBuffer. Eliminates redundant ByteBuffer.slice() allocation.
4. DefaultMessageStore — Replace thenAccept with whenComplete for stats recording. Preserves put-latency metrics, slow-log, and failure counting while eliminating lambda capture overhead.
5. ConsumeQueue — Make topicQueueKey a final field to avoid per-call string computation.

How Did You Test This Change?

1. Unit tests: All 62 tests pass (EscapeBridgeTest, PopConsumerServiceTest, PopLiteMessageProcessorTest, BatchConsumeMessageTest).
2. Compilation: store + broker modules compile cleanly on JDK 21.
3. Commercial compatibility: No commercial classes extend any modified class.
4. Microbenchmark: R1 encode/decode microbenchmarks confirmed that JIT-optimized String operations outperform manual caching. S3 changes avoid this pitfall by focusing on primitive arrays and ThreadLocal reuse.

[Copilot]

Pull request overview

Note

Copilot was unable to run its full agentic suite in this review.

This PR focuses on reducing allocations and improving I/O paths in the store layer by reusing buffers, adding a ByteBuffer-based FileChannel append API, and optimizing queue offset storage.

Changes:

• Add appendMessageUsingFileChannel(ByteBuffer) to MappedFile and implement it in DefaultMappedFile
• Optimize buffer usage (cached slice reuse; safer SelectMappedBufferResult slicing via duplicate() + explicit limits)
• Replace GetMessageResult queue offsets from List<Long> to a primitive long[] with a lightweight List view and add addQueueOffset(...)

Reviewed changes

Copilot reviewed 9 out of 9 changed files in this pull request and generated 6 comments.

Show a summary per file

File	Description
store/src/main/java/org/apache/rocketmq/store/logfile/MappedFile.java	Adds a ByteBuffer overload for FileChannel-based appends
store/src/main/java/org/apache/rocketmq/store/logfile/DefaultMappedFile.java	Implements ByteBuffer FileChannel append; reuses append slices; adjusts buffer selection logic
store/src/main/java/org/apache/rocketmq/store/GetMessageResult.java	Switches queue offsets to primitive array + view; adds addQueueOffset
store/src/main/java/org/apache/rocketmq/store/DispatchRequest.java	Makes fields resettable; adds reset(...) and a package-private ctor
store/src/main/java/org/apache/rocketmq/store/DefaultMessageStore.java	Simplifies async put methods; pre-sizes GetMessageResult with maxMsgNums
store/src/main/java/org/apache/rocketmq/store/ConsumeQueue.java	Reuses a precomputed topic-queue key; uses ByteBuffer FileChannel append
broker/src/test/java/org/apache/rocketmq/broker/processor/PopLiteMessageProcessorTest.java	Updates tests to use addQueueOffset
broker/src/test/java/org/apache/rocketmq/broker/pop/PopConsumerServiceTest.java	Updates tests to use addQueueOffset
broker/src/test/java/org/apache/rocketmq/broker/failover/EscapeBridgeTest.java	Updates tests to use addQueueOffset

---

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

[RockteMQ-AI]

Review by github-manager-bot

Summary

This PR reduces allocation in the pull/dispatch hot path through six optimizations: primitive long[] replacing List<Long> in GetMessageResult, DispatchRequest object reuse via reset(), cached topicQueueKey in ConsumeQueue, ByteBuffer-based FileChannel write in DefaultMappedFile, and removal of async callback in DefaultMessageStore. The direction is correct — eliminating autoboxing and buffer copies in high-throughput paths.

Findings

• [Critical] DefaultMessageStore.java:620-630 — Removal of put-message stats tracking. The thenAccept callback that tracked setPutMessageEntireTimeMax(), getPutMessageFailedTimes(), and the slow-put warning log has been removed entirely. This eliminates observability for:

  1. Put message latency percentiles (used by monitoring/alerting)
  2. Failed put message counts (used for error rate tracking)
  3. Slow put warnings (>500ms threshold)

  The PR title says "Reduce pull/dispatch path allocation" but this removes stats from the put path. This appears to be scope creep or an unintended side effect. If the goal is to reduce allocation, consider keeping the stats tracking but using a more efficient mechanism (e.g., direct field assignment instead of lambda capture).

• [Warning] DispatchRequest.java:45-70 — reset() method always sets success = true. The error constructor DispatchRequest(..., long tagsCode) sets success = false, but reset() unconditionally sets it to true. If the object pool reuses a failed DispatchRequest, the error state is silently lost. Consider either:

  1. Adding a success parameter to reset()
  2. Or ensuring failed requests are never reused (document this contract)

• [Info] GetMessageResult.java:45-55 — The long[] array with AbstractList<Long> wrapper for backward compatibility is a good pattern. The growth formula size + (size >> 1) avoids the +1 in addQueueOffset, but since the default constructor initializes with new long[100], this is safe in practice.

• [Info] DefaultMappedFile.java:200-210 — The cachedAppendSlice lazy initialization is not volatile/synchronized, but since the slice is deterministic (same underlying buffer), the worst case is one redundant allocation, which is benign.

Suggestions

1. Restore or relocate the put-message stats tracking in DefaultMessageStore — this is a regression in observability.
2. Clarify the DispatchRequest reuse contract regarding the success field.
3. Consider adding a JMH benchmark for the ConsumeQueue.appendMessage path to quantify the improvement.

---

Automated review by github-manager-bot

[codecov-commenter]

Codecov Report

❌ Patch coverage is 64.78873% with 25 lines in your changes missing coverage. Please review.
✅ Project coverage is 48.09%. Comparing base (59a70d9) to head (a313999).
⚠️ Report is 6 commits behind head on develop.

Files with missing lines	Patch %	Lines
...va/org/apache/rocketmq/store/GetMessageResult.java	59.25%	9 Missing and 2 partials ⚠️
...ache/rocketmq/store/logfile/DefaultMappedFile.java	61.53%	9 Missing and 1 partial ⚠️
...org/apache/rocketmq/store/DefaultMessageStore.java	71.42%	1 Missing and 1 partial ⚠️
...ava/org/apache/rocketmq/store/DispatchRequest.java	77.77%	2 Missing ⚠️

Additional details and impacted files

@@              Coverage Diff              @@
##             develop   #10526      +/-   ##
=============================================
- Coverage      48.19%   48.09%   -0.11%     
+ Complexity     13398    13380      -18     
=============================================
  Files           1377     1377              
  Lines         100730   100885     +155     
  Branches       13012    13039      +27     
=============================================
- Hits           48548    48517      -31     
- Misses         46252    46408     +156     
- Partials        5930     5960      +30     

☔ View full report in Codecov by Harness.
📢 Have feedback on the report? Share it here.

🚀 New features to boost your workflow:

• ❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.

[RockteMQ-AI]

Review by github-manager-bot (Re-review after new commits)

Summary

Force-pushed update to the pull/dispatch path allocation reduction PR. The core optimizations remain: long[] replacing List<Long> in GetMessageResult, DispatchRequest ThreadLocal reuse, cached topicQueueKey, DefaultMappedFile buffer optimizations, and async callback removal in DefaultMessageStore. New: ByteBuffer-based appendMessageUsingFileChannel in ConsumeQueue and duplicate() instead of slice() in selectMappedBuffer.

Findings

• [Critical] DefaultMessageStore.java — Put-message stats tracking still removed. The thenAccept callback containing setPutMessageEntireTimeMax(), getPutMessageFailedTimes(), and the slow-put warning log (>500ms) has been removed. This eliminates observability for put latency, failure counts, and slow-put alerts. This was flagged in the previous review and remains unaddressed. This is a regression in monitoring capability.

  Suggestion: Keep the stats tracking but avoid the lambda allocation. Options:

  1. Inline the stats tracking after the synchronous putMessage call (no lambda needed)
  2. Use a pre-allocated callback object stored in a field
  3. Move stats tracking to the caller of putMessage/putMessages

• [Warning] ConsumeQueue.java — byteBufferIndex.flip() on shared buffer. The flip() call before appendMessageUsingFileChannel(this.byteBufferIndex) is correct for the write path (sets limit=position, position=0). However, byteBufferIndex is a shared field. If any code path reads from this buffer after the flip() without first resetting it, it will see stale data. Since the dispatch path is single-threaded per ConsumeQueue, this is safe in practice, but consider adding a comment documenting the buffer lifecycle contract.

• [Info] DefaultMappedFile.java:selectMappedBuffer — Using duplicate() + limit() instead of slice() + slice() reduces one ByteBuffer allocation. The duplicate() shares the underlying data but has independent position/limit, which is correct here.

• [Info] DefaultMappedFile.java:appendMessageUsingFileChannel(ByteBuffer) — New overload accepting ByteBuffer directly. The data.remaining() call correctly computes the write length. The flip() in the caller ensures the buffer is in read mode. This avoids the .array() call which would throw UnsupportedOperationException for direct buffers.

Unresolved from Previous Review

• DispatchRequest.reset() always sets success = true — if a failed request is reused from the ThreadLocal pool, the error state is silently lost. Consider documenting that failed requests must not be reused, or adding a success parameter to reset().

---

Automated review by github-manager-bot

[RockteMQ-AI]

Review by github-manager-bot (Re-review after new commits)

Summary

Re-review after new commit a3139997 (2026-06-19). This update addresses the critical put-message stats tracking regression from the previous review by switching thenAccept → whenComplete with proper null-check. The DispatchRequest reset/pooling pattern and GetMessageResult long[] optimization are carried forward. Test files are updated to use the new addQueueOffset() API.

Resolved Issues ✅

• [Critical] RESOLVED DefaultMessageStore.java — Put-message stats tracking is now restored via whenComplete. The callback correctly handles both success and failure:
  • null == result covers the exceptional completion case (counts as failure)
  • Slow-put warning (>500ms) is preserved
  • setPutMessageEntireTimeMax and getPutMessageFailedTimes are both called
  • Note: The exception ex parameter is not logged. Consider adding if (ex != null) LOGGER.warn("...", ex) for debuggability, though the failure counter correctly captures it.

Remaining Findings

• [Warning] DispatchRequest.java — reset() always sets success = true. If a failed dispatch request is returned to the ThreadLocal pool and later reused, the error state is silently lost. The caller would see success = true even though the original dispatch failed.

  • Suggestion: Either document that failed requests must not be pooled, or add a reset(boolean success) parameter so callers can preserve the original state.

• [Warning] DefaultMappedFile.java — cachedAppendSlice is a shared instance field. The cached ByteBuffer.slice() is stored as this.cachedAppendSlice. If appendMessage and appendMessagesInner are ever called concurrently on the same MappedFile instance, the shared slice could cause data corruption. This is likely safe in practice (putMessage is serialized by lock), but consider adding a comment documenting the thread-safety assumption.

• [Info] GetMessageResult.java — long[] replacing List<Long> is a good allocation reduction. The AbstractList<Long> view returned by getMessageQueueOffset() boxes on each get() call, but this is acceptable for read paths.

• [Info] ConsumeQueue.java — Caching topicQueueKey as an instance field avoids repeated string concatenation on the dispatch hot path. Good optimization.

• [Info] DefaultMappedFile.java — duplicate() + limit() instead of double slice() reduces one ByteBuffer allocation. The appendMessageUsingFileChannel(ByteBuffer) overload avoids the .array() call which would fail for direct buffers.

Verdict

The critical regression is fixed. The remaining warnings are non-blocking. This PR looks good to merge after addressing the DispatchRequest.reset() concern.

---

Automated re-review by github-manager-bot