[ISSUE #10658] Fix PopConsumerService.revive aborting the whole batch on a single failed record

[wang-jiahua]

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.

[Copilot]

Copilot was unable to review this pull request because the user who requested the review has reached their quota limit.

[RockteMQ-AI]

Review by github-manager-bot

Summary

Fixes head-of-line blocking in PopConsumerService.revive(AtomicLong, int) where a single failing record would abort the entire batch via CompletableFuture.allOf(...).join(), preventing healthy records from being persisted and retried.

Findings

• [Info] PopConsumerService.java:564-569 — The .exceptionally() handler correctly downgrades async read failures to false, routing the record through the existing failureList backoff-retry path. Clean approach that reuses existing retry infrastructure.

• [Info] PopConsumerService.java:597-614 — Good separation of semaphore.acquire() InterruptedException handling from the revive(record) sync-failure catch. Restoring the interrupt flag with Thread.currentThread().interrupt() before wrapping in RuntimeException is the correct pattern.

• [Info] PopConsumerService.java:610-613 — The sync-failure catch creates CompletableFuture.completedFuture(false) so the record enters the failureList path. The semaphore permit is correctly released by the downstream whenComplete stage since completedFuture triggers it immediately.

• [Info] PopConsumerServiceTest.java — Two well-targeted tests covering both failure modes (async exceptionally and sync throw). Both verify that batch bookkeeping (writeRecords/deleteRecords/scanExpiredRecords) still runs after a failure, which is the key invariant this fix protects.

Assessment

This is a solid, minimal fix for a classic batch-processing head-of-line blocking problem. The error isolation is correct for both async and sync failure paths, semaphore management is sound, and the tests directly validate the fix. Scope is appropriately limited to the newer popkv path (popConsumerKVServiceEnable), not touching the legacy PopBufferMergeService.

---

Automated review by github-manager-bot

[codecov-commenter]

Codecov Report

❌ Patch coverage is 66.66667% with 3 lines in your changes missing coverage. Please review.
✅ Project coverage is 48.22%. Comparing base (b37e2bb) to head (95e36e2).

Files with missing lines	Patch %	Lines
...apache/rocketmq/broker/pop/PopConsumerService.java	66.66%	3 Missing ⚠️

Additional details and impacted files

@@              Coverage Diff              @@
##             develop   #10659      +/-   ##
=============================================
- Coverage      48.35%   48.22%   -0.14%     
+ Complexity     13510    13468      -42     
=============================================
  Files           1380     1380              
  Lines         101044   101051       +7     
  Branches       13094    13094              
=============================================
- Hits           48863    48728     -135     
- Misses         46225    46340     +115     
- Partials        5956     5983      +27     

☔ View full report in Codecov by Harness.
📢 Have feedback on the report? Share it here.

🚀 New features to boost your workflow:

• ❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.

[RockteMQ-AI]

Review by github-manager-bot

Summary

Fixes PopConsumerService.revive so that a single failed record (async or sync) does not abort the entire revive batch. Adds an .exceptionally() handler to the per-record future chain and catches synchronous failures from revive(record) in the batch loop.

Findings

• [Critical] PopConsumerService.java:564-568 — The new .exceptionally() handler correctly catches async failures (e.g., remote read via escape bridge failing) and returns false instead of propagating the exception. This prevents allOf().join() from throwing and skipping the writeRecords/deleteRecords cleanup.
• [Warning] PopConsumerService.java:597-608 — The synchronous failure handling creates a CompletableFuture.completedFuture(false) as a fallback. This is correct, but note that the semaphore.acquire() is now in a separate try-catch before the future creation. If revive(record) throws synchronously, the semaphore permit was already acquired but the .whenComplete stage on the fallback future will release it. Verify that the thenAccept callback attached later (at futureList.add(future.thenAccept(...))) also properly releases the semaphore in all paths.
• [Info] PopConsumerService.java:594-596 — Good fix for the InterruptedException handling: properly restores the interrupt flag with Thread.currentThread().interrupt() before throwing.
• [Info] Tests are comprehensive: reviveShouldNotAbortBatchWhenGetMessageFailsExceptionally and reviveShouldIsolateSynchronousReadFailureAndNotBlockHealthyRecords directly validate both failure modes.

Suggestions

Consider adding a brief comment on the fallback future explaining why it's needed (the synchronous throw path that bypasses the .exceptionally() chain).

Verdict

Solid fix for a critical reliability issue. The batch processing pattern is now resilient to both async and sync failures.

---

Automated review by github-manager-bot