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

未关闭
  • 哪里不对？
#11,403 1 条评论 0 个 reaction 已指派 0 人 在 GitHub 查看

维护者通常 1 天内回复

@messere1 已经在做这个了。

开始于 2026年10月13日。

  • #11405 来自 @messere1 —— 未关闭

评估

难度
3/5
预计耗时
1-2 天
新手友好度
62/100
Issue 类型
缺陷
描述清晰度
描述清楚
活跃度
活跃
技术栈
java
领域
backend

调研方向

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.

由索引模型根据 Issue 内容生成。

描述

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.

主要语言
Java
星标
22.7k
派生
12k
平均合并
3 天 18 小时
30 天内合并 PR
17

环境准备

从这里开始

  1. 先读完整个 Issue，再读项目的贡献指南。
  2. 在 Issue 下留言说明你要接手 —— 这能避免两个人做同样的事。
  3. Fork 仓库，在一个分支上完成修改。
  4. 提交 Pull Request，并在描述里引用这个 Issue 编号。

apache/rocketmq 的其他 Issue

查看 apache/rocketmq 的全部 Issue

相似的 Issue

更多 Java Issue

把新 issue 发到你的邮箱

精选适合新手参与的 GitHub issue 摘要。