apache / apache/rocketmq

PopConsumerCache.cleanupRecords clears staged checkpoints regardless of the revive result, losing messages when an asynchronous revive fails

Open
#10,667 5 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

### 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

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

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.