PopConsumerService.revive aborts the whole batch when a single record fails (head-of-line blocking in the popkv pop path)
- Dominant language
- Java
- Stars
- 22.6k
- Forks
- 12k
- Avg merge
- 3d 1h
- Merged PRs (30d)
- 27
Description
### 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:
```java
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
Contributor guide
Research direction
Start in PopConsumerService, focusing on revive(AtomicLong currentTime, int maxCount) and revive(PopConsumerRecord), then run the named PopConsumerServiceTest scenarios with one synchronous or asynchronous record failure. Done means the failing record is scheduled for retry while healthy records are consumed, batch bookkeeping runs, and progress advances.
Written by the indexing model from the issue text.
Assessment
- Tech stack
- java
- Domain
- backend, distributed-systems
- Issue type
- Bug
- Difficulty
- 3/5
- Estimated time
- 1-2 days
- Activity status
- Quiet
- Clarity
- Clearly specified
- Newbie friendliness
- 72/100