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

Aperta
  • Qualcosa non va?
#11,379 1 commento 0 reazioni 0 assegnatari Vedi su GitHub

I maintainer di solito rispondono entro 1 giorno

@messere1 ci sta già lavorando.

Dal 12/10/2026.

  • #11380 di @messere1 — aperta

Valutazione

Difficoltà
4/5
Tempo stimato
1-2 giorni
Idoneità per principianti
50/100
Tipo di issue
Bug
Chiarezza
Specificata chiaramente
Stato di attività
Attiva
Stack tecnologico
java
Ambito
backend, distributed-systems

Direzione di ricerca

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.

Scritto dal modello di indicizzazione a partire dal testo della issue.

Descrizione

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.

Lingua principale
Java
Stelle
22.7k
Fork
12k
Merge medio
4g 20h
PR unite (30g)
11

Preparare l'ambiente

Come iniziare

  1. Leggi tutta la issue e poi la guida ai contributi del progetto.
  2. Commenta sulla issue per dire che te ne occupi tu — evita che due persone facciano lo stesso lavoro.
  3. Fai un fork del repository e lavora su un branch.
  4. Apri una pull request che faccia riferimento al numero della issue.

Altre issue di apache/rocketmq

Tutte le issue di apache/rocketmq

Issue simili

Altre issue su Java

Ricevi le nuove issue nella tua casella

Un breve riepilogo di issue GitHub adatte ai principianti.