wang-jiahua opened a new pull request, #10614: URL: https://github.com/apache/rocketmq/pull/10614
### 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). -- 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]
