apache / apache/rocketmq

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

Open
#10,658 3 comments 0 reactions 0 assignees View on GitHub
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

Open the contributing 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

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.