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

[wang-jiahua]

Before Creating the Enhancement Request

• 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 (#10659 (comment)). 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.

[RockteMQ-AI]

Issue Evaluation

Category: type/bug | Status: Confirmed

The reported issue has been verified against the current codebase (develop branch, commit 00e45b8a6).

Root Cause: In PopConsumerCache.cleanupRecords(), expired records with visibilityTimeout <= currentTime are passed to consumer.accept(record) (the revive callback), but:

1. The return value (future) is discarded — asynchronous failures are silently ignored
2. These records are not added to writeConsumerRecords, so they are not persisted to the store
3. clearStagedRecords() then unconditionally clears the entire removeTreeMap

If the revive future completes exceptionally (e.g., reviveRetry throwing inside the chain), the record has already been removed from cache and was never persisted, resulting in silent message loss.

Impact: POP consumer checkpoints can be lost when enablePopBufferMerge=true and an asynchronous revive fails. Affected consumers lose their un-acked message checkpoints.

Severity: High — potential message loss in production under specific failure conditions.

An automated fix proposal will be generated. Reply /approve to proceed with PR generation.

---

Automated evaluation by github-manager-bot

[RockteMQ-AI]

Fix Spec Generated

A detailed fix specification has been prepared for this issue.

Strategy: Make the revive-failure handling of the cache path identical to the KV-store path — observe the future; on failure, re-persist the record to the KV store with backoff so the durable revive loop retries it later.

Key Changes:

1. PopConsumerService.java — Extract buildBackoffRetryRecord() helper from existing inline retry logic (pure refactor)
2. PopConsumerService.java — Add reviveBufferedRecord(PopConsumerRecord) method that wraps revive() with .whenComplete() to observe async results. On false/exception: re-persist to KV store with exponential backoff. Change constructor to bind this::reviveBufferedRecord instead of this::revive
3. PopConsumerCache.java — No functional change needed; the existing clearStagedRecords() becomes correct once the callback guarantees durable failure handling

Behavior Change:

Revive outcome	Before	After
Future completes true	record dropped ✓	record dropped ✓
Future completes false	record lost	re-persisted with backoff
Future completes exceptionally	record lost, error swallowed	error logged + re-persisted with backoff

Tests: 6 new tests covering failed future, false result, success, exhausted attempts, backoff math, and end-to-end cache sweep.

Risk: No API/protocol change. Re-persisted records flow through existing durable revive loop. Happy path adds negligible overhead (one whenComplete per expired record).

Reply /approve to proceed with PR generation, /revise <feedback> to request changes, or /reject to decline.

---

Automated fix proposal by github-manager-bot

[ai-yang]

Hi, I would like to work on this issue once #10659 reaches a stable merged state. I reproduced the failure path against the current develop branch: an expired checkpoint is staged, an exceptionally completed revive future is discarded, cleanup clears that checkpoint, and the cache can then commit the next offset even though no durable retry record exists.

I plan to preserve the flush-before-offset invariant introduced by #9627:

1. Change the cache callback from Consumer to Function<PopConsumerRecord, CompletableFuture>.
2. Dispatch all revive futures for a cleanup sweep before waiting, normalize each result independently, and remove a staged checkpoint only after the callback reports that it is safe to do so. A false, null, synchronous throw, or exceptional completion keeps the checkpoint staged and prevents the minimum offset from advancing past it.
3. Bind the cache to a reviveBufferedRecord wrapper. On revive failure, the wrapper will synchronously persist the existing backoff retry record before reporting success to the cache. If that write fails, the original checkpoint remains staged.
4. Reuse the final backoff construction from [ISSUE #10658] Fix PopConsumerService.revive aborting the whole batch on a single failed record #10659, preserving attemptTimes, attemptId, and suspend.

This intentionally differs from clearing the cache immediately and attaching only a completion callback: that smaller approach still leaves a broker-crash window between the clear/offset advance and durable retry persistence.

The regression coverage will include exceptional and false futures, synchronous throws, persistence failure, concurrent dispatch before waiting, offset gating, backoff/max-attempt behavior, suspend preservation, and an enablePopBufferMerge=true cache-to-RocksDB path.

I will wait for #10659 to merge, then recheck all open and closed PRs before creating a branch so the work does not overlap. Please let me know if the maintainers prefer a different reliability boundary.

[RockteMQ-AI]

⏰ Spec Feedback Timeout

No formal /approve was received within 72 hours of the fix spec being posted (2026-07-28T09:10:47Z).

Spec summary: Fix for PopConsumerCache.cleanupRecords clearing staged checkpoints regardless of the revive result — the async revive callback's return value (CompletableFuture<Boolean>) was being discarded.

Community discussion observed:

• @RongtJin (committer) acknowledged the spec and directed @chenxu80 to submit a fix
• @chenxu80 agreed to submit a fix shortly

Since no formal approval was received and the 72h feedback window has elapsed, this issue workspace is being closed. If the community or maintainers would like to proceed with this fix, please comment and the workspace can be reopened.

---

This is an automated message from the github-manager bot. The fix spec remains available for reference.

[wang-jiahua]

@ai-yang Thanks for picking this up, and please go ahead. Your reliability boundary is the right one, and it is stronger than what I had locally.

I had a smaller fix on a local branch that binds the cache to a wrapper which persists the backoff record after the revive future fails. You correctly identified its weakness: the checkpoint is cleared before the retry record is durable, so a broker crash in that window loses it. I had noted the window but treated it as acceptable because in-flight cache records are not persisted anyway. Your point about the offset is what changes that judgement, and I had missed it: cleanupRecords commits getMinOffsetInBuffer() after clearStagedRecords(), so once a staged checkpoint is dropped without a durable retry record, the committed offset can advance past it and the message is unrecoverable rather than merely delayed. Gating the minimum offset until the callback reports durability is the part my approach did not cover. I am not opening a competing PR.

A few things I verified while working on this, in case any of them save you time.

Both failure shapes have to be normalized, and they behave differently. I probed the JDK semantics with a small program rather than reasoning about it:

• A thenCompose lambda that throws on an already-completed upstream does not throw synchronously; the caller receives an exceptionally completed future.
• An upstream that throws before returning a future does throw synchronously at the call site. DefaultMessageStore.getMessageAsync is completedFuture(getMessage(...)), so a local read failure takes this path, while the tiered store path completes exceptionally instead.
• An unobserved exceptionally completed future is entirely silent in the JVM; nothing is logged when it is dropped.

The practical consequence for the cache path today is that these two shapes have opposite outcomes: an exceptional completion lets accept() return normally, so the clear runs and the checkpoint is lost, whereas a synchronous throw propagates out of accept() and aborts cleanupRecords before the clear, which accidentally preserves the checkpoint and is also the only case the catch in run() can see. Normalizing both, as you described, removes that asymmetry.

On the persistence layer. Following @lizhimins's note on this issue about RocksDB WAL atomicity, one detail worth keeping in mind: writeRecords and deleteRecords are two separate WriteBatch instances and two db.write calls, each atomic on its own but not jointly. The batch revive path relies on the ordering rather than joint atomicity, writing the retry records before deleting the originals, so a crash between the two can only produce duplicates, never loss. If the cache path keeps the same ordering, the same reasoning holds.

Reusing the backoff construction. The nine-argument PopConsumerRecord constructor that preserves attemptTimes, attemptId, and suspend is already in #10659 as of 771afdcd4, so it should be available to build on once that merges. The eight-argument constructor hard-codes suspend to false when it delegates, which silently discards what the client set through changeInvisibilityDuration(..., suspend=true) and makes a later successful reviveRetry increment reconsumeTimes unexpectedly.

One test detail. When driving cleanupRecords directly in a unit test, the consumer has to look online or the sweep takes the offline eviction branch instead: PopConsumerLockService.isLockTimeout returns true for a key that was never locked, so a tryLock followed by unlock on the group and topic is needed first.

Happy to review once your PR is up.