This is an automated email from the ASF dual-hosted git repository.
fuyou001 pushed a commit to branch develop
in repository https://gitbox.apache.org/repos/asf/rocketmq.git
The following commit(s) were added to refs/heads/develop by this push:
new 3373f284e6 [ISSUE #10613] Fix async request-reply RequestCallback
firing multiple times (#10614)
3373f284e6 is described below
commit 3373f284e6d8e5101f63a38be28f1fe3ec9f6773
Author: Jiahua Wang <[email protected]>
AuthorDate: Fri Jul 17 18:43:41 2026 +0800
[ISSUE #10613] Fix async request-reply RequestCallback firing multiple
times (#10614)
- DefaultMQProducerImpl#request(Message, RequestCallback, long): drop the
executeRequestCallback() call in the async send onSuccess so a send
success
no longer delivers a premature onSuccess(null); align with the other async
request overloads which only set sendRequestOk here.
- RequestResponseFuture: add an AtomicBoolean executeCallbackOnlyOnce guard
so
the callback fires at most once even if the reply and timeout paths race.
- RequestFutureHolder#scanExpiredRequest: use ConcurrentHashMap.remove(key)
to
atomically claim ownership instead of iterator.remove(); also fix the log
placeholder concatenation.
- ClientRemotingProcessor#processReplyMessage: use atomic
remove(correlationId)
and route the reply through executeRequestCallback so the single-shot
guard
covers the reply-success path too.
- Add RequestResponseFutureTest cases for success-then-timeout and
concurrent
single-callback semantics.
Co-authored-by: wangjiahua.wjh <[email protected]>
---
.../client/impl/ClientRemotingProcessor.java | 13 ++---
.../impl/producer/DefaultMQProducerImpl.java | 5 +-
.../client/producer/RequestFutureHolder.java | 10 +++-
.../client/producer/RequestResponseFuture.java | 20 +++++--
.../client/producer/RequestResponseFutureTest.java | 66 ++++++++++++++++++++++
5 files changed, 97 insertions(+), 17 deletions(-)
diff --git
a/client/src/main/java/org/apache/rocketmq/client/impl/ClientRemotingProcessor.java
b/client/src/main/java/org/apache/rocketmq/client/impl/ClientRemotingProcessor.java
index 0a187751f2..567198e3a8 100644
---
a/client/src/main/java/org/apache/rocketmq/client/impl/ClientRemotingProcessor.java
+++
b/client/src/main/java/org/apache/rocketmq/client/impl/ClientRemotingProcessor.java
@@ -272,15 +272,14 @@ public class ClientRemotingProcessor implements
NettyRequestProcessor {
private void processReplyMessage(MessageExt replyMsg) {
final String correlationId =
replyMsg.getUserProperty(MessageConst.PROPERTY_CORRELATION_ID);
- final RequestResponseFuture requestResponseFuture =
RequestFutureHolder.getInstance().getRequestFutureTable().get(correlationId);
+ // Atomically remove so that only one of the reply-arrival path and
the timeout-scan path
+ // (RequestFutureHolder#scanExpiredRequest) can take ownership of this
request.
+ final RequestResponseFuture requestResponseFuture =
RequestFutureHolder.getInstance().getRequestFutureTable().remove(correlationId);
if (requestResponseFuture != null) {
requestResponseFuture.putResponseMessage(replyMsg);
-
-
RequestFutureHolder.getInstance().getRequestFutureTable().remove(correlationId);
-
- if (requestResponseFuture.getRequestCallback() != null) {
- requestResponseFuture.getRequestCallback().onSuccess(replyMsg);
- }
+ // Route through executeRequestCallback so the single-callback
guard applies to the
+ // reply-success path too; sync callers (null callback) are woken
by putResponseMessage.
+ requestResponseFuture.executeRequestCallback();
} else {
String bornHost = replyMsg.getBornHostString();
logger.warn("receive reply message, but not matched any request,
CorrelationId: {} , reply from host: {}",
diff --git
a/client/src/main/java/org/apache/rocketmq/client/impl/producer/DefaultMQProducerImpl.java
b/client/src/main/java/org/apache/rocketmq/client/impl/producer/DefaultMQProducerImpl.java
index f68742949f..9ad5fcef4d 100644
---
a/client/src/main/java/org/apache/rocketmq/client/impl/producer/DefaultMQProducerImpl.java
+++
b/client/src/main/java/org/apache/rocketmq/client/impl/producer/DefaultMQProducerImpl.java
@@ -1662,8 +1662,11 @@ public class DefaultMQProducerImpl implements
MQProducerInner {
this.sendDefaultImpl(msg, CommunicationMode.ASYNC, new SendCallback() {
@Override
public void onSuccess(SendResult sendResult) {
+ // Only mark the request as sent here. The user callback must
fire when the reply
+ // arrives (processReplyMessage), on timeout
(scanExpiredRequest), or on send
+ // failure (requestFail). Invoking it here delivers a
premature onSuccess(null)
+ // and, combined with the later reply/timeout callback, causes
a double callback.
requestResponseFuture.setSendRequestOk(true);
- requestResponseFuture.executeRequestCallback();
}
@Override
diff --git
a/client/src/main/java/org/apache/rocketmq/client/producer/RequestFutureHolder.java
b/client/src/main/java/org/apache/rocketmq/client/producer/RequestFutureHolder.java
index 00f5bb6e6e..b489873c79 100644
---
a/client/src/main/java/org/apache/rocketmq/client/producer/RequestFutureHolder.java
+++
b/client/src/main/java/org/apache/rocketmq/client/producer/RequestFutureHolder.java
@@ -54,9 +54,13 @@ public class RequestFutureHolder {
RequestResponseFuture rep = next.getValue();
if (rep.isTimeout()) {
- it.remove();
- rfList.add(rep);
- log.warn("remove timeout request, CorrelationId={}" +
rep.getCorrelationId());
+ // Atomically remove so the timeout path and the reply-arrival
path in
+ // ClientRemotingProcessor#processReplyMessage cannot both
handle the same request.
+ RequestResponseFuture removed =
requestFutureTable.remove(next.getKey());
+ if (removed != null) {
+ rfList.add(removed);
+ log.warn("remove timeout request, CorrelationId={}",
removed.getCorrelationId());
+ }
}
}
diff --git
a/client/src/main/java/org/apache/rocketmq/client/producer/RequestResponseFuture.java
b/client/src/main/java/org/apache/rocketmq/client/producer/RequestResponseFuture.java
index e66c08fdc5..8d13dc7f52 100644
---
a/client/src/main/java/org/apache/rocketmq/client/producer/RequestResponseFuture.java
+++
b/client/src/main/java/org/apache/rocketmq/client/producer/RequestResponseFuture.java
@@ -19,6 +19,7 @@ package org.apache.rocketmq.client.producer;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
import org.apache.rocketmq.common.message.Message;
public class RequestResponseFuture {
@@ -31,6 +32,7 @@ public class RequestResponseFuture {
private volatile Message responseMsg = null;
private volatile boolean sendRequestOk = true;
private volatile Throwable cause = null;
+ private final AtomicBoolean executeCallbackOnlyOnce = new
AtomicBoolean(false);
public RequestResponseFuture(String correlationId, long timeoutMillis,
RequestCallback requestCallback) {
this.correlationId = correlationId;
@@ -39,12 +41,18 @@ public class RequestResponseFuture {
}
public void executeRequestCallback() {
- if (requestCallback != null) {
- if (sendRequestOk && cause == null) {
- requestCallback.onSuccess(responseMsg);
- } else {
- requestCallback.onException(cause);
- }
+ if (requestCallback == null) {
+ return;
+ }
+ // Ensure the callback fires exactly once even if the reply-arrival
path and the
+ // timeout-scan path race on the same request. Whoever wins the CAS
runs the callback.
+ if (!this.executeCallbackOnlyOnce.compareAndSet(false, true)) {
+ return;
+ }
+ if (sendRequestOk && cause == null) {
+ requestCallback.onSuccess(responseMsg);
+ } else {
+ requestCallback.onException(cause);
}
}
diff --git
a/client/src/test/java/org/apache/rocketmq/client/producer/RequestResponseFutureTest.java
b/client/src/test/java/org/apache/rocketmq/client/producer/RequestResponseFutureTest.java
index 68615f3b39..f5c3d7709c 100644
---
a/client/src/test/java/org/apache/rocketmq/client/producer/RequestResponseFutureTest.java
+++
b/client/src/test/java/org/apache/rocketmq/client/producer/RequestResponseFutureTest.java
@@ -18,6 +18,7 @@
package org.apache.rocketmq.client.producer;
import java.util.UUID;
+import java.util.concurrent.CountDownLatch;
import java.util.concurrent.atomic.AtomicInteger;
import org.apache.rocketmq.common.message.Message;
import org.junit.Test;
@@ -44,4 +45,69 @@ public class RequestResponseFutureTest {
assertThat(cc.get()).isEqualTo(1);
}
+ @Test
+ public void testExecuteRequestCallbackSuccessThenTimeoutFiresOnce() {
+ final AtomicInteger success = new AtomicInteger(0);
+ final AtomicInteger exception = new AtomicInteger(0);
+ RequestResponseFuture future = new
RequestResponseFuture(UUID.randomUUID().toString(), 3 * 1000L, new
RequestCallback() {
+ @Override
+ public void onSuccess(Message message) {
+ success.incrementAndGet();
+ }
+
+ @Override
+ public void onException(Throwable e) {
+ exception.incrementAndGet();
+ }
+ });
+
+ // Reply-arrival path wins first (success), then the timeout-scan path
tries to fire again.
+ future.setSendRequestOk(true);
+ future.executeRequestCallback();
+
+ future.setCause(new RuntimeException("request timeout, no reply
message."));
+ future.executeRequestCallback();
+
+ // The same request must never deliver both a success and a timeout
callback.
+ assertThat(success.get()).isEqualTo(1);
+ assertThat(exception.get()).isEqualTo(0);
+ }
+
+ @Test
+ public void testExecuteRequestCallbackConcurrentlyFiresOnce() throws
Exception {
+ final AtomicInteger total = new AtomicInteger(0);
+ final RequestResponseFuture future = new
RequestResponseFuture(UUID.randomUUID().toString(), 3 * 1000L, new
RequestCallback() {
+ @Override
+ public void onSuccess(Message message) {
+ total.incrementAndGet();
+ }
+
+ @Override
+ public void onException(Throwable e) {
+ total.incrementAndGet();
+ }
+ });
+ future.setSendRequestOk(true);
+
+ int threads = 16;
+ final CountDownLatch start = new CountDownLatch(1);
+ final CountDownLatch done = new CountDownLatch(threads);
+ for (int i = 0; i < threads; i++) {
+ new Thread(() -> {
+ try {
+ start.await();
+ future.executeRequestCallback();
+ } catch (InterruptedException ignored) {
+ } finally {
+ done.countDown();
+ }
+ }).start();
+ }
+ start.countDown();
+ done.await();
+
+ // Under concurrent contention the CAS guard still allows exactly one
callback.
+ assertThat(total.get()).isEqualTo(1);
+ }
+
}
