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

Abierto
  • ¿Algo no está bien?
#11,379 1 comentario 0 reacciones 0 asignados Ver en GitHub

Los mantenedores suelen responder en 1 día

@messere1 ya está trabajando en esto.

Desde el 12/10/2026.

  • #11380 de @messere1 — abierto

Evaluación

Dificultad
4/5
Tiempo estimado
1-2 días
Aptitud para principiantes
50/100
Tipo de issue
Error
Claridad
Bien especificado
Estado de actividad
Activo
Stack tecnológico
java
Área
backend, distributed-systems

Línea de trabajo

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.

Escrito por el modelo de indexación a partir del texto del issue.

Descripción

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.

Lenguaje dominante
Java
Estrellas
22.7k
Forks
12k
Merge medio
4 d 20 h
PR fusionados (30 d)
11

Preparar el entorno

Primeros pasos

  1. Lee el issue completo y luego la guía de contribución del proyecto.
  2. Comenta en el issue que vas a ocuparte — evita que dos personas hagan lo mismo.
  3. Haz un fork del repositorio y trabaja en una rama.
  4. Abre un pull request que haga referencia al número del issue.

Más de apache/rocketmq

Todos los issues de apache/rocketmq

Issues similares

Más issues de Java

Recibe los nuevos issues en tu correo

Un resumen breve de issues de GitHub para principiantes.