wang-jiahua commented on issue #10667: URL: https://github.com/apache/rocketmq/issues/10667#issuecomment-5237019290
@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. -- 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]
