Tiered storage dispatch loop loses message buffers and NPEs when CQ/commitlog entries expire mid-batch

Đang mở
  • Có gì không ổn?
#11,379 1 bình luận 0 reaction 0 người được giao Xem trên GitHub

Maintainer thường phản hồi trong vòng 1 ngày

@messere1 đang làm issue này rồi.

Từ ngày 12/10/2026.

  • #11380 của @messere1 — đang mở

Đánh giá

Độ khó
4/5
Thời gian dự kiến
1-2 ngày
Mức phù hợp với người mới
50/100
Loại issue
Lỗi
Độ rõ ràng
Đặc tả rõ ràng
Mức độ hoạt động
Sôi nổi
Công nghệ
java
Lĩnh vực
backend, distributed-systems

Hướng nghiên cứu

Start in MessageStoreDispatcherImpl#doScheduleDispatch and trace the group-commit append loop, GroupCommitContext#release(), and the paths that call commitAsync or constructIndexFile. Check how missing CQ units or message buffers and a failed consume-queue append affect the collected buffers. Done means dispatch handles these cases without an NPE or leaked mapped-buffer references; the issue does not name a test to run.

Do mô hình lập chỉ mục viết ra từ nội dung của issue.

Mô tả

Background

MessageStoreDispatcherImpl#doScheduleDispatch is the tiered-storage dispatch loop. Before entering the group-commit append loop it preflights the first offset and guards both lookups (added by #8965 / issue #8955, which fixed the dispatch thread exiting on NPE):

CqUnit cqUnit = consumeQueue.get(currentOffset);
if (cqUnit == null) { log.warn(...); return ...completedFuture(false); }
SelectMappedBufferResult message = defaultStore.selectOneMessageByOffset(cqUnit.getPos(), cqUnit.getSize());
if (message == null) { log.warn(...); return ...completedFuture(false); }

However, the append loop that follows re-fetches the CQ unit and the message for every subsequent offset without any of those guards:

for (; offset < targetOffset; offset++) {
    cqUnit = consumeQueue.get(offset);
    bufferSize += cqUnit.getSize();                       // NPE if cqUnit == null
    ...
    message = defaultStore.selectOneMessageByOffset(cqUnit.getPos(), cqUnit.getSize());
    appendingBufferList.add(message);
    AppendResult result = flatFile.appendCommitLog(message); // NPE/garbage if message == null
    ...
}
Problem

  1. Mid-loop NPE. ConsumeQueueInterface#get returns null once the consume queue has been truncated past the requested offset, and selectOneMessageByOffset returns null once the commitlog data has been deleted (expired files). Both events happen through the ordinary expiry sweeps while the dispatcher is catching up (e.g. after a downtime longer than the file retention). The preflight only proves the first offset of the batch is still there; the batch spans up to tieredStoreGroupCommitCount more offsets, so the lookups can fail inside the loop. The resulting NPE escapes doScheduleDispatch (only ConsumeQueueException is caught there).

  2. Message buffer leak. Every SelectMappedBufferResult fetched inside the loop is added to appendingBufferList and is only ever released through GroupCommitContext#release(). When the loop dies with the NPE above, the context is never constructed, so every buffer fetched before the failing offset leaks its MappedFile reference.

    The same leak exists on a non-exceptional path: if appendCommitLog returns SUCCESS but the following appendConsumeQueue fails on the first message, the loop breaks with appendingBufferList non-empty and dispatchRequestList empty. The context is built, but the if (!dispatchRequestList.isEmpty()) block — the only place that hands it to commitAsync/constructIndexFile, both of which release it — never runs. The context is silently dropped and its buffers leak.

Impact
  • Repeated mapped-buffer/reference leaks on queues whose data expires while dispatch lags, and on flat files that fail a consume-queue append after a successful commitlog append — eventually holding expired commitlog files from being cleaned.
  • The NPE aborts the current dispatch round for the queue (the dispatcher thread itself survives because dispatch() catches Throwable), so dispatch progress depends on the next cycle re-failing the preflight instead of degrading gracefully.
Suggestion

Mirror the preflight guards inside the loop (break on a missing CQ unit or message so the partial batch still commits and the context is still released), and release the context when nothing was appended to the consume queue so no code path will consume it.

Ngôn ngữ chính
Java
Star
22.7k
Fork
12k
Merge trung bình
4 ngày 20 giờ
Pull request đã merge (30 ngày)
11

Chuẩn bị môi trường

Bắt đầu từ đâu

  1. Đọc hết issue, rồi đọc hướng dẫn đóng góp của dự án.
  2. Bình luận trên issue rằng bạn sẽ nhận — tránh hai người làm cùng một việc.
  3. Fork repository và làm thay đổi trên một nhánh.
  4. Mở pull request có tham chiếu số hiệu của issue.

Issue khác của apache/rocketmq

Tất cả issue của apache/rocketmq

Issue tương tự

Thêm issue về Java

Nhận issue mới trong hộp thư của bạn

Bản tóm tắt ngắn những issue GitHub phù hợp với người mới.