[ISSUE #10613] Fix async request-reply RequestCallback firing multiple times

[wang-jiahua]

Which Issue(s) This PR Fixes

Fixes #10613

Brief Description

The async request-reply API DefaultMQProducer#request(Message, RequestCallback, long) could invoke the user RequestCallback more than once for a single request, due to two combined problems:

1. Premature onSuccess(null) — the async send onSuccess called executeRequestCallback() right after the request message was sent, so the caller received onSuccess(null) before any reply arrived. The other async overloads (request(msg, selector, arg, callback, timeout), request(msg, mq, callback, timeout)) do not do this.
2. reply-vs-timeout race — ClientRemotingProcessor#processReplyMessage (non-atomic get + remove) and RequestFutureHolder#scanExpiredRequest (iterator.remove) could both take ownership of the same request, and RequestResponseFuture#executeRequestCallback had no single-shot guard.

How Did You Fix It

• DefaultMQProducerImpl#request(Message, RequestCallback, long): drop executeRequestCallback() from the async send onSuccess; only set sendRequestOk, consistent with the other async overloads. The user callback now fires only on reply arrival, timeout, or send failure.
• RequestResponseFuture: add an AtomicBoolean executeCallbackOnlyOnce guard so the callback runs at most once regardless of which path wins.
• RequestFutureHolder#scanExpiredRequest: use ConcurrentHashMap.remove(key) to atomically claim ownership instead of iterator.remove(); also fix the malformed log placeholder.
• ClientRemotingProcessor#processReplyMessage: use atomic remove(correlationId) and route the reply through executeRequestCallback so the single-shot guard also covers the reply-success path.

How to Verify

Unit tests (RequestResponseFutureTest): success-then-timeout fires exactly once; 16-thread concurrent invocation fires exactly once. Full client module test suite passes.

End-to-end on a 4-node cluster (2000 async requests, 16 threads, timeout=1000ms, consumer reply delay=200ms), identical reply flow (consumed=1312, replySent=1280):

Metric	before (develop)	after (this PR)
successReply (real reply callback)	284	286
successNull (premature onSuccess(null))	2000	0
doubleCallback	2000	0
nullThenReply / successAndException	284 / 1716	0 / 0

After the fix each request delivers exactly one callback (286 onSuccess(reply) + 1714 onException(timeout) = 2000).

[Copilot]

Copilot was unable to review this pull request because the user who requested the review has reached their quota limit.

[RockteMQ-AI]

Review by github-manager-bot

Summary

Fixes async request-reply RequestCallback firing multiple times due to (1) premature onSuccess(null) in the send callback and (2) a race between the reply-arrival and timeout-scan paths. Well-structured fix with defense-in-depth.

Findings

• [Info] ClientRemotingProcessor.java:275 — Atomic remove(correlationId) correctly eliminates the TOCTOU race between reply arrival and timeout scan. Clean fix.
• [Info] DefaultMQProducerImpl.java:1665 — Removing executeRequestCallback() from the send onSuccess is correct and consistent with the other async overloads (request(msg, selector, arg, callback, timeout), request(msg, mq, callback, timeout)). The callback now fires only on reply, timeout, or send failure.
• [Info] RequestFutureHolder.java:57-62 — Good catch on the malformed log placeholder. The original "CorrelationId={}" + rep.getCorrelationId() would produce CorrelationId={}abc123 instead of CorrelationId=abc123. The remove(key) atomic claim pattern is correct.
• [Info] RequestResponseFuture.java:44-53 — AtomicBoolean CAS guard is a solid defense-in-depth measure. Even if both paths somehow bypass the atomic remove(), the callback still fires exactly once.
• [Info] RequestResponseFutureTest.java — Tests cover both sequential double-call (success → timeout) and 16-thread concurrent contention. Good coverage of the fix.

Suggestions

• No blocking concerns. The fix is minimal, focused, and well-tested. The end-to-end verification table in the PR description (2000 requests, doubleCallback: 2000 → 0) is convincing.

---

Automated review by github-manager-bot

[RongtongJin]

LGTM～

[RockteMQ-AI]

Review by github-manager-bot

PR: Fix async request-reply RequestCallback firing multiple times
Verdict: ✅ Approved

Summary

Fixes a race condition where RequestCallback could be invoked multiple times in the async request-reply pattern. The root cause: the reply-arrival path (processReplyMessage) and the timeout-scan path (scanExpiredRequest) could both access the same RequestResponseFuture without atomic ownership, and the send-success path also fired a premature callback.

Analysis

Correctness ✅

• Atomic ownership via remove(): Replacing get() + later remove() with a single remove() in processReplyMessage ensures only one path (reply-arrival OR timeout-scan) can claim the future. This is the correct fix.
• AtomicBoolean guard: The executeCallbackOnlyOnce in RequestResponseFuture provides a second layer of defense, ensuring the callback fires exactly once even under race conditions.
• Removed premature callback: Removing executeRequestCallback() from onSuccess(SendResult) is correct — the user callback should fire when the reply arrives, not when the send succeeds. The previous behavior delivered a premature onSuccess(null).
• Timeout path fix: scanExpiredRequest now uses requestFutureTable.remove(key) instead of it.remove() + separate handling, ensuring atomic claim of the future.

Performance ✅

• ConcurrentHashMap.remove() and AtomicBoolean.compareAndSet() are both lock-free, minimal overhead.

Compatibility ✅

• No public API changes. The behavioral fix (no more double callbacks) is what users would expect.

Tests ✅

• New test testAsyncRequestCallbackFiredOnce validates the single-callback guarantee.

Verdict

Well-designed fix for a subtle concurrency bug. The dual-layer approach (atomic remove + AtomicBoolean guard) is robust. Clear comments explain the reasoning at each decision point.