[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 77.77778% with 2 lines in your changes missing coverage. Please review.
✅ Project coverage is 48.24%. Comparing base (b37e2bb) to head (b84fcc0).
⚠️ Report is 3 commits behind head on develop.

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

Additional details and impacted files

@@              Coverage Diff              @@
##             develop   #10659      +/-   ##
=============================================
- Coverage      48.35%   48.24%   -0.12%     
+ Complexity     13510    13487      -23     
=============================================
  Files           1380     1380              
  Lines         101044   101097      +53     
  Branches       13094    13101       +7     
=============================================
- Hits           48863    48772      -91     
- Misses         46225    46346     +121     
- Partials        5956     5979      +23     

☔ 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

[fuyou001]

[P1] Preserve failed records in the PopConsumerCache call path

revive(PopConsumerRecord) is not only called by revive(AtomicLong, int). When enablePopBufferMerge is enabled, PopConsumerService passes this::revive to PopConsumerCache as a Consumer<PopConsumerRecord>. In cleanupRecords(), consumer.accept(record) discards the returned CompletableFuture<Boolean>, and clearStagedRecords() then unconditionally removes the staged checkpoints.

The new exception-to-false conversion works in the RocksDB batch path because false is consumed and a backoff record is added to failureList. In the cache path, however, that false result is never observed. An asynchronously failed revive can therefore be cleared from the cache without being persisted for retry, preventing the message from being revived. The outer catch in PopConsumerCache.run() only sees synchronous exceptions thrown before a future is returned; it cannot observe an exceptional future completion.

Please either attach the exception-to-false handling only at the batch call site if PopConsumerCache is intentionally out of scope, or change the cache callback to return and handle the future, clearing a staged record only after success and retaining or persisting it with backoff on false or exception. Please also add a regression test with enablePopBufferMerge=true and an asynchronously failed revive, verifying that the checkpoint remains retryable.

[wang-jiahua]

[P1] Preserve failed records in the PopConsumerCache call path

@fuyou001 Thanks for the careful review — you are right. Addressed by taking the first option in commit b84fcc0: the exception-to-false conversion is now attached only at the batch call site.

• revive(PopConsumerRecord) no longer carries an .exceptionally handler, so its exception semantics are exactly what they were before this PR; the PopConsumerCache callback path is untouched.
• In revive(AtomicLong, int) the handler sits right where the false result is consumed into the failureList backoff retry, with a comment noting that other callers do not consume the result.
• This also removes a side problem: the "will retry" log would have been misleading on the cache path, where a cleared record is never retried.

On the pre-existing cache-path behavior, I verified: an asynchronously failed revive was already lost before this PR — the exceptionally-completed future is unobserved, the revive-attempted records are not in writeConsumerRecords, and clearStagedRecords() clears the staging map unconditionally; the run() catch only sees synchronous throws, which abort cleanupRecords before the clear and get retried on the next sweep. Since the cache path is intentionally out of scope here, I've opened #10667 to track clearing a staged record only after a successful revive, including the enablePopBufferMerge=true regression test you suggested.

Both regression tests still pass (PopConsumerServiceTest 19/19).

[RockteMQ-AI]

Review by github-manager-bot (re-review after new commit)

Summary

Re-reviewed after new commit (Jul 28). The previous suggestion to add explanatory comments on the fallback future path has been addressed — both the .exceptionally() call site and the synchronous catch block now have clear comments explaining why the conversion happens here and not inside revive(record) itself.

Changes Since Last Review

• Added inline comments explaining the .exceptionally() placement rationale (call-site-specific semantics vs. revive(record) internal behavior)
• Added inline comment on the synchronous fallback future explaining the semaphore release path
• InterruptedException handling correctly restores the interrupt flag

Verdict

Previous findings remain addressed. The batch processing error isolation is solid. LGTM.

---

Automated review by github-manager-bot

[fuyou001]

[P1] Avoid leaving the revive service permanently interrupted

The current InterruptedException handling restores the interrupt flag and throws. PopConsumerService.run() catches the exception and calls waitForRunning(500), but ServiceThread.waitForRunning() preserves the interrupt status. As a result, every subsequent Semaphore.acquire() immediately throws again, so a single unexpected interrupt can stop all future revive progress and cause a tight error/log loop. The current batch is also abandoned before writeRecords and deleteRecords run.

The normal broker shutdown path calls PopConsumerService.shutdown() without interrupting the worker; it stops the service through the stopped flag and wakeup(). Therefore, an incidental interrupt while waiting for the per-batch semaphore can be consumed and retried:

while (true) {
    try {
        semaphore.acquire();
        break;
    } catch (InterruptedException e) {
        // PopConsumerService shutdown uses the stopped flag and wakeup(), rather
        // than thread interruption. Do not restore the interrupt flag here;
        // otherwise subsequent acquire() and waitForRunning() calls may return
        // immediately and leave the revive service in a permanent busy loop.
        log.warn("Interrupted while acquiring semaphore, retry");
    }
}

Please also add a regression test that blocks on the semaphore, interrupts the revive thread once, releases the outstanding operation, and verifies that the remaining records are processed without repeated exceptions or CPU spinning.

If the inherited shutdown(true) path is expected to be supported in the future, it should be handled separately by checking the stopped state and terminating cleanly instead of swallowing that shutdown interrupt.

[fuyou001]

[P1] Preserve the suspend flag in the backoff record

When a synchronous or asynchronous revive failure is converted to false, the backoff record is created with the eight-argument PopConsumerRecord constructor, which defaults suspend to false.

This changes the semantics of records created by changeInvisibilityDuration(..., suspend=true). After the original record is replaced by this backoff record, a later successful reviveRetry sees suspend=false and increments reconsumeTimes unexpectedly. For example, a suspended message with reconsumeTimes=2 can become 3 merely because its first revive attempt encountered a transient read or write failure.

Please preserve the original value when constructing the retry record:

PopConsumerRecord retryRecord = new PopConsumerRecord(
    System.currentTimeMillis(),
    record.getGroupId(),
    record.getTopicId(),
    record.getQueueId(),
    record.getRetryFlag(),
    nextInvisibleTime,
    record.getOffset(),
    record.getAttemptId(),
    record.isSuspend());

Please cover both synchronous and asynchronous failure paths with suspend=true. Reload the generated retry record from the store and assert that suspend remains true, attemptTimes is incremented exactly once, the original record is deleted only after the retry record is persisted, and a later successful reviveRetry does not increment reconsumeTimes.