PopConsumerService.revive aborts the whole batch when a single record fails (head-of-line blocking in the popkv pop path)

[wang-jiahua]

Describe the bug

In the KV-based ("popkv") pop consumer org.apache.rocketmq.broker.pop.PopConsumerService (instantiated when popConsumerKVServiceInit is enabled, used when popConsumerKVServiceEnable is enabled), the batch revive method revive(AtomicLong currentTime, int maxCount) awaits all per-record futures and then runs the batch bookkeeping:

CompletableFuture.allOf(futureList.toArray(new CompletableFuture[0])).join();
this.popConsumerStore.writeRecords(new ArrayList<>(failureList));
this.popConsumerStore.deleteRecords(consumerRecords);
currentTime.set(...);

There is no per-record error isolation, so a single failing record aborts the whole batch:

• revive(PopConsumerRecord) chains getMessageAsync(record).thenCompose(...) with no .exceptionally(...) handler, so an exceptional completion (e.g. reviveRetry throwing inside thenCompose, or a decode failure in the escape bridge) makes the returned future complete exceptionally.
• revive(record) can also throw synchronously before it returns a future (e.g. DefaultMessageStore.getMessageAsync is completedFuture(getMessage(...)) and getMessage throws), which the batch loop currently rethrows as RuntimeException.

Either way, writeRecords(failureList), deleteRecords(consumerRecords) and the currentTime advance are all skipped and revive() throws out to the run loop. If one record keeps failing (e.g. a persistently unreachable remote), revive is stuck reprocessing the same batch forever, and the retries for the other, healthy records in that batch are never persisted — head-of-line blocking.

Scope

This is in the newer "popkv" pop path (PopConsumerService, gated by popConsumerKVServiceEnable), which is disabled by default; it does not affect the legacy PopBufferMergeService. It is a correctness fix that matters as the popkv path is rolled out.

Steps to reproduce

In PopConsumerServiceTest: write two expired records into the store, make one record's read fail (either an async exceptional completion, or a synchronous throw), and run revive(...). Before the fix the whole batch aborts (the healthy record is not consumed and progress does not advance); after the fix the failing record is isolated (scheduled for retry) and the batch still advances.

Version

develop

[RockteMQ-AI]

Issue Evaluation

Category: type/bug | Status: Confirmed | Severity: Medium

Verification Summary

Verified against PopConsumerService.java on develop branch. The described head-of-line blocking behavior is confirmed in the code.

Root Cause Analysis

Two failure paths in revive(AtomicLong, int) cause the entire batch to abort:

1. Synchronous throw (line 593-597): When this.revive(record) throws synchronously (e.g., getMessageAsync completes synchronously with a throw), the catch block re-throws as RuntimeException, aborting the for-loop. All subsequent records are not processed.

2. Async exceptional completion (line 617): revive(PopConsumerRecord) at line 552 chains getMessageAsync(record).thenCompose(...) with no .exceptionally() handler. If any future completes exceptionally, CompletableFuture.allOf(...).join() throws CompletionException, skipping:

   • writeRecords(failureList) — line 618
   • deleteRecords(consumerRecords) — line 619
   • currentTime.set(...) — line 620

This means healthy records in the same batch are never persisted as processed, and the batch gets stuck retrying the same failing record indefinitely.

Scope

Confirmed: popkv path only (popConsumerKVServiceEnable), disabled by default. Does not affect legacy PopBufferMergeService.

Suggested Fix Direction

• Add .exceptionally() to the per-record future in revive(PopConsumerRecord) to isolate failures
• Wrap the per-record call in the batch loop to catch both sync and async failures
• Ensure writeRecords, deleteRecords, and currentTime advance always execute even when individual records fail

---

Automated evaluation by github-manager-bot