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

Open
  • Is something wrong?
#11,403 1 comment 0 reactions 0 assignees View on GitHub

Maintainers usually reply within 1 day

@messere1 is already working on this.

Since Oct 13, 2026.

  • #11405 by @messere1 — open

Assessment

Difficulty
3/5
Estimated time
1-2 days
Newbie friendliness
62/100
Issue type
Bug
Clarity
Clearly specified
Activity status
Active
Tech stack
java
Domain
backend

Research direction

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.

Written by the indexing model from the issue text.

Description

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.

Dominant language
Java
Stars
22.7k
Forks
12k
Avg merge
3d 18h
Merged PRs (30d)
17

Getting set up

First steps

  1. Read the whole issue, then the project's contributing guide.
  2. Comment on the issue to say you are picking it up — it saves two people doing the same work.
  3. Fork the repository and make your change on a branch.
  4. Open a pull request that references the issue number.

More from apache/rocketmq

All issues in apache/rocketmq

Similar issues

More Java issues

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.