[Bug] transferToFsStore deletes POP KV records even when the checkpoint put fails, losing in-flight messages

Abierto
  • ¿Algo no está bien?
#11,403 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 13/10/2026.

  • #11405 de @messere1 — abierto

Evaluación

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

Línea de trabajo

Start in PopConsumerService.transferToFsStore(), where the PutMessageResult from asyncPutMessage(ckMsg).join() is discarded before popConsumerStore.deleteRecords(consumerRecords) runs. Use handleCkMessagePutResult in PopBufferMergeService as the reference for which statuses count as success. The reproduction is PopConsumerServiceTest#transferToFsStoreTest with a stub returning SERVICE_NOT_AVAILABLE; done means that test shows the record still present after the put fails, and the retry loop no longer spins forever.

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

Descripción

Before Creating the Bug Report
  • This is a reproducible bug rather than a usage question.
  • I searched open and closed issues, pull requests, and Discussions and found no duplicate.
  • The affected code is in this RocketMQ repository.
Runtime platform environment

Linux x86_64. The reproduction is a deterministic broker unit test and does not require a running RocketMQ cluster.

RocketMQ version
  • Branch: develop
  • Commit: a7edeecfb
JDK Version

OpenJDK 17

Describe the Bug

PopConsumerService.transferToFsStore() migrates in-flight POP consumer records from the RocksDB KV store to the FS (revive topic) store when an operator switches the pop consumption implementation back to the filesystem store (AdminBrokerProcessor#transferPopToFsStore). For each record it builds a PopCheckPoint and appends it to the revive topic:

brokerController.getMessageStore().asyncPutMessage(ckMsg).join();

The returned PutMessageResult is discarded, and afterwards the whole scanned batch is removed from the KV store unconditionally:

this.popConsumerStore.deleteRecords(consumerRecords);

When the checkpoint put does not succeed (SERVICE_NOT_AVAILABLE, OS_PAGECACHE_BUSY, CREATE_MAPPED_FILE_FAILED, ...), no checkpoint exists in the FS store, yet the KV record is deleted anyway. The popped-but-not-yet-acked message then has no checkpoint in either store: if the consumer never acks it (client crash, processing failure), nothing will ever revive or re-deliver it - the message is silently lost. Note that store puts are most likely to fail exactly when a broker is degraded, which is precisely when an operator performs this migration, so an entire batch (up to popReviveMaxReturnSizePerRead records) can be dropped in one pass.

Two related consequences in the same method:

  1. If asyncPutMessage completes exceptionally, join() throws, the catch (Throwable) block only logs, and the while (true) loop rescans the same (still present) records forever inside a synchronized method. The admin request never returns, and any later invocation of the method blocks on the monitor.
  2. The per-batch log line always reports the scanned count as transferred, so even the logs claim records were migrated that were not persisted.
Steps to Reproduce

A deterministic JUnit + Mockito reproduction, mirroring the existing PopConsumerServiceTest#transferToFsStoreTest:

  1. Build a PopConsumerService with the usual mocked BrokerController and start its KV store.
  2. Write one PopConsumerRecord (e.g. offset 0) into the store.
  3. Stub messageStore.asyncPutMessage(any()) to complete with new PutMessageResult(PutMessageStatus.SERVICE_NOT_AVAILABLE, null).
  4. Call consumerService.transferToFsStore().
  5. Call popConsumerStore.scanExpiredRecords(0, Long.MAX_VALUE, 10).

Actual: the scan returns 0 records - the record was deleted from the KV store although no checkpoint was ever persisted.

Expected: the record must still be present (1 record), because its checkpoint was not put successfully; it should stay revivable/re-migratable.

Additional Context

PopBufferMergeService#handleCkMessagePutResult and ChangeInvisibleTimeProcessor#appendCheckPointThenAckOrigin already treat only PUT_OK, FLUSH_DISK_TIMEOUT, FLUSH_SLAVE_TIMEOUT and SLAVE_NOT_AVAILABLE as success when appending checkpoints to the revive topic; transferToFsStore should not use a weaker contract than the other checkpoint writers.

Lenguaje dominante
Java
Estrellas
22.7k
Forks
12k
Merge medio
3 d 18 h
PR fusionados (30 d)
17

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.