messere1 opened a new pull request, #11405: URL: https://github.com/apache/rocketmq/pull/11405
### What `PopConsumerService.transferToFsStore()` migrated POP in-flight records from the RocksDB KV store to the FS revive-topic store by ignoring the result of `asyncPutMessage(ckMsg).join()`: when the checkpoint put failed (`SERVICE_NOT_AVAILABLE`, `OS_PAGECACHE_BUSY`, ...), the whole scanned batch was still removed from the KV store, so the popped-but-unacked message had no checkpoint in either store and could never be revived - a silent message loss during the KV-to-FS migration (the operation operators run exactly when a broker is degraded). Additionally, when `asyncPutMessage` completed exceptionally, the `catch` block only logged and the `while (true)` loop rescanned the same undeleted records forever inside a `synchronized` method, so the admin request never returned and later invocations blocked on the monitor. See issue #11403 for the deterministic reproduction. ### How - Check the `PutMessageResult` of every checkpoint put. Only the statuses the other checkpoint writers already accept (`PUT_OK`, `FLUSH_DISK_TIMEOUT`, `FLUSH_SLAVE_TIMEOUT`, `SLAVE_NOT_AVAILABLE`, same set as `PopBufferMergeService#handleCkMessagePutResult` and `ChangeInvisibleTimeProcessor`) count as transferred. - Delete only the records whose checkpoint was actually persisted; records with a failed/exceptional put stay in the KV store, so they remain revivable and the migration can simply be re-run. - Stop the transfer loop when a scan batch makes no progress instead of spinning on the same failing batch forever (this also turns the exceptional-put path from a livelock into a logged, bounded stop). Transient failures still self-heal because the loop keeps retrying as long as each batch transfers at least one record. ### Verification Red (before the fix, on current develop): - `transferToFsStoreKeepsRecordWhenCheckpointPutFailsTest` failed with `expected:<1> but was:<0>` - the record was deleted although the checkpoint put returned `SERVICE_NOT_AVAILABLE`. - `transferToFsStoreDeletesOnlyTransferredRecordsTest` failed the same way - all 3 records were deleted although the third put failed. - `transferToFsStoreStopsWhenCheckpointPutThrowsTest` failed with `test timed out after 30000 milliseconds` - the exceptional put sent the method into an unbounded retry loop (the run produced ~694MB of repeated error logs in 30s and the surefire fork could not exit because the runaway thread never finished). Green (with the fix): - `mvn -pl broker test -Dtest='PopConsumerServiceTest'`: `Tests run: 21, Failures: 0, Errors: 0, Skipped: 0` (18 pre-existing tests plus the 3 new ones, including the original `transferToFsStoreTest` happy path). Environment note: the local build stops after the tests at `jacoco:0.8.5:report` with `Unsupported class file major version 61` (JaCoCo 0.8.5 vs Java 17 bytecode). This reproduces identically on a clean develop checkout and is unrelated to this change. ### Notes for reviewers - The per-batch info log now reports the number of records actually persisted instead of the number scanned. - This change touches `transferToFsStore()` only; it is orthogonal to the suspend-flag handling proposed for the same method in #10466 and should merge cleanly with it. -- 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]
