wang-jiahua opened a new issue, #10658:
URL: https://github.com/apache/rocketmq/issues/10658
### 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
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
