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

オープン
  • 何か問題がありますか？
#11,379 コメント 1 件 リアクション 0 件 担当者 0 名 GitHub で見る

メンテナーはふだん 1 日以内に返信

@messere1 がすでに取り組んでいます。

2026年10月12日 から。

  • #11380 @messere1 による — オープン

評価

難易度
4/5
見積もり時間
1〜2日
初心者へのやさしさ
50/100
issue の種類
バグ
明瞭さ
明確に書かれている
活発さ
活発
技術スタック
java
領域
backend, distributed-systems

調査の方向性

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.

索引モデルが issue の本文から書いたものです。

説明

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.

主要言語
Java
スター
22.7k
フォーク
12k
平均マージ
4日 20時間
マージ済み PR（30日）
11

環境構築

はじめの一歩

  1. issue を最後まで読み、次にプロジェクトのコントリビューションガイドを読みます。
  2. 着手することを issue にコメントします — 二人が同じ作業をするのを防げます。
  3. リポジトリをフォークし、ブランチを切って変更します。
  4. issue 番号を参照したプルリクエストを送ります。

apache/rocketmq のほかの issue

apache/rocketmq の issue をすべて見る

似ている issue

Java の issue をもっと見る

新しい issue をメールで受け取る

初心者向けの GitHub issue を短くまとめたダイジェスト。