wang-jiahua opened a new pull request, #10659: URL: https://github.com/apache/rocketmq/pull/10659
### Which Issue(s) This PR Fixes Fixes #10658 ### Brief Description In the popkv pop consumer (`PopConsumerService`), the batch `revive(AtomicLong, int)` awaits all per-record futures with `CompletableFuture.allOf(...).join()` and has no per-record error isolation, so a single failing record aborts the whole batch (`writeRecords(failureList)`, `deleteRecords(consumerRecords)` and the `currentTime` advance are skipped, and `revive()` throws). A persistently failing record therefore blocks the whole batch forever, and the healthy records' retries in that batch are never persisted (head-of-line blocking). A record can fail two ways, both handled here: 1. **Async** — `revive(PopConsumerRecord)` gets an `.exceptionally(...)` handler that logs and downgrades the failure to a "failed revive" (`false`), so it goes through the existing `failureList` backoff-retry path instead of propagating through `allOf(...).join()`. 2. **Sync** — in the batch loop, a synchronous throw from `revive(record)` (e.g. `getMessageAsync` throwing before it returns a future) is caught and turned into `completedFuture(false)` instead of being rethrown and aborting the batch. The semaphore permit is released by the `whenComplete` stage; `acquire()`'s `InterruptedException` is handled separately. Either way a single record's failure only affects that record, and `writeRecords` / `deleteRecords` / `currentTime` still run. Note: this is the newer popkv pop path (`PopConsumerService`, gated by `popConsumerKVServiceEnable`), which is disabled by default; it does not touch the legacy `PopBufferMergeService`. ### How Did You Test This Change? Added two tests in `PopConsumerServiceTest`, both failing before the fix and passing after: - `reviveShouldNotAbortBatchWhenGetMessageFailsExceptionally` — a single record whose read completes exceptionally; asserts `revive` does not throw and the record is consumed (batch bookkeeping still runs). - `reviveShouldIsolateSynchronousReadFailureAndNotBlockHealthyRecords` — two records, one whose read throws synchronously and one healthy; asserts `revive` returns 2 and both original records are consumed (`scanExpiredRecords` returns 0), i.e. the failed record is isolated and the healthy record is not head-of-line blocked. Full `PopConsumerServiceTest` passes (19/19), and `mvn -pl broker validate` reports 0 checkstyle violations. -- 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]
