PopConsumerCache.cleanupRecords clears staged checkpoints regardless of the revive result, losing messages when an asynchronous revive fails
- Dominant language
- Java
- Stars
- 22.6k
- Forks
- 12k
- Avg merge
- 3d 1h
- Merged PRs (30d)
- 27
Description
### Before Creating the Enhancement Request
- [x] I have confirmed that this should be classified as an enhancement rather than a bug/feature.
### Summary
Split out of the review discussion on #10659 (https://github.com/apache/rocketmq/pull/10659#issuecomment-5099396425). This behavior pre-dates that PR and is left out of its scope on purpose.
When `enablePopBufferMerge` is on (default off), `PopConsumerCache.run()` periodically calls `cleanupRecords(reviveConsumer)`, where `reviveConsumer` is `PopConsumerService::revive` used as a `Consumer`:
- expired records staged in `removeTreeMap` are handed to `consumer.accept(record)`, and the returned `CompletableFuture` is discarded;
- these records are not added to `writeConsumerRecords`, so they are not persisted;
- `clearStagedRecords()` then unconditionally clears the whole staging map.
If the revive future completes exceptionally (for example `reviveRetry` throwing inside the chain), nobody observes the failure: the record has already been removed from the cache and was never persisted, so the message is never revived. The outer catch in `run()` only sees synchronous throws, which abort `cleanupRecords` before the clear and are retried on the next sweep; it cannot see exceptional future completions.
### Motivation
A checkpoint of an un-acked POP message silently disappears on an asynchronously failed revive, which means the message is lost for the consumer. The batch revive path handles the same failure by writing a backoff-retry record; the cache path should not be weaker.
### Describe the Solution You'd Like
Change the cache callback so it can observe the future: clear a staged record only after a successful revive, and retain it or persist it with backoff on `false` or exceptional completion. Add a regression test with `enablePopBufferMerge=true` and an asynchronously failed revive, verifying the checkpoint remains retryable.
### Describe Alternatives You've Considered
Blocking on each future inside `cleanupRecords` before `clearStagedRecords()`. Simpler to reason about, but it serializes the sweep and changes the timing of the cleanup loop.
### Additional Context
Verified semantics behind the analysis: a `thenCompose` lambda throwing on a completed upstream produces an exceptionally-completed future (no synchronous throw), and an unobserved exceptionally-completed future is silent in the JVM.
Contributor guide
Research direction
Read PopConsumerCache.run(), cleanupRecords(), and clearStagedRecords(), then trace PopConsumerService::revive and its CompletableFuture result. Add the regression test with enablePopBufferMerge=true and an asynchronously failed revive; done means the checkpoint remains retryable.
Written by the indexing model from the issue text.
Assessment
- Tech stack
- java
- Domain
- distributed-systems, testing
- Issue type
- Bug
- Difficulty
- 4/5
- Estimated time
- 3-5 days
- Activity status
- Quiet
- Clarity
- Mostly clear
- Newbie friendliness
- 52/100