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

Đang mở
  • Có gì không ổn?
#11,403 1 bình luận 0 reaction 0 người được giao Xem trên GitHub

Maintainer thường phản hồi trong vòng 1 ngày

@messere1 đang làm issue này rồi.

Từ ngày 13/10/2026.

  • #11405 của @messere1 — đang mở

Đánh giá

Độ khó
3/5
Thời gian dự kiến
1-2 ngày
Mức phù hợp với người mới
62/100
Loại issue
Lỗi
Độ rõ ràng
Đặc tả rõ ràng
Mức độ hoạt động
Sôi nổi
Công nghệ
java
Lĩnh vực
backend

Hướng nghiên cứu

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.

Do mô hình lập chỉ mục viết ra từ nội dung của issue.

Mô tả

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.

Ngôn ngữ chính
Java
Star
22.7k
Fork
12k
Merge trung bình
3 ngày 18 giờ
Pull request đã merge (30 ngày)
17

Chuẩn bị môi trường

Bắt đầu từ đâu

  1. Đọc hết issue, rồi đọc hướng dẫn đóng góp của dự án.
  2. Bình luận trên issue rằng bạn sẽ nhận — tránh hai người làm cùng một việc.
  3. Fork repository và làm thay đổi trên một nhánh.
  4. Mở pull request có tham chiếu số hiệu của issue.

Issue khác của apache/rocketmq

Tất cả issue của apache/rocketmq

Issue tương tự

Thêm issue về Java

Nhận issue mới trong hộp thư của bạn

Bản tóm tắt ngắn những issue GitHub phù hợp với người mới.