messere1 opened a new issue, #11403: URL: https://github.com/apache/rocketmq/issues/11403
### Before Creating the Bug Report - [x] This is a reproducible bug rather than a usage question. - [x] I searched open and closed issues, pull requests, and Discussions and found no duplicate. - [x] 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: ```java brokerController.getMessageStore().asyncPutMessage(ckMsg).join(); ``` The returned `PutMessageResult` is discarded, and afterwards the whole scanned batch is removed from the KV store unconditionally: ```java 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. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
