wang-jiahua opened a new issue, #10667: URL: https://github.com/apache/rocketmq/issues/10667
### 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<PopConsumerRecord>`: - expired records staged in `removeTreeMap` are handed to `consumer.accept(record)`, and the returned `CompletableFuture<Boolean>` 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. -- 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]
