(rocketmq) branch develop updated: [ISSUE #10575] Fix race condition between scanResponseTable and processResponseCommand (#10576)

[jinrongtong]

This is an automated email from the ASF dual-hosted git repository.

RongtongJin 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 e348efa66b [ISSUE #10575] Fix race condition between scanResponseTable 
and processResponseCommand (#10576)
e348efa66b is described below

commit e348efa66b08eb645ee123706ea6492fa9a3ad35
Author: Jiahua Wang <90542846+wang-jia...@users.noreply.github.com>
AuthorDate: Wed Aug 26 15:12:46 2026 +0800

    [ISSUE #10575] Fix race condition between scanResponseTable and 
processResponseCommand (#10576)
    
    processResponseCommand used get+remove (non-atomic) to retrieve 
ResponseFuture
    from responseTable. scanResponseTable used iterator.remove(). If the 
response
    arrived between scanResponseTable's remove and processResponseCommand's get,
    the response would be logged as 'not matched any request' and silently 
dropped.
    The callback would fire with timeout instead of success, even though the 
broker
    had successfully processed the request.
    
    Fix: Use ConcurrentHashMap.remove(key) in both methods. This is atomic:
    either processResponseCommand gets the future (and executes success 
callback),
    or scanResponseTable gets it (and executes timeout callback), but never both
    and never neither.
    
    Co-authored-by: wangjiahua.wjh <wangjiahua....@alibaba-inc.com>
---
 .../rocketmq/remoting/netty/NettyRemotingAbstract.java     | 14 +++++++-------
 1 file changed, 7 insertions(+), 7 deletions(-)

diff --git 
a/remoting/src/main/java/org/apache/rocketmq/remoting/netty/NettyRemotingAbstract.java
 
b/remoting/src/main/java/org/apache/rocketmq/remoting/netty/NettyRemotingAbstract.java
index a735f8455d..a40aab3ae5 100644
--- 
a/remoting/src/main/java/org/apache/rocketmq/remoting/netty/NettyRemotingAbstract.java
+++ 
b/remoting/src/main/java/org/apache/rocketmq/remoting/netty/NettyRemotingAbstract.java
@@ -467,12 +467,10 @@ public abstract class NettyRemotingAbstract {
      */
     public void processResponseCommand(ChannelHandlerContext ctx, 
RemotingCommand cmd) {
         final int opaque = cmd.getOpaque();
-        final ResponseFuture responseFuture = responseTable.get(opaque);
+        final ResponseFuture responseFuture = responseTable.remove(opaque);
         if (responseFuture != null) {
             responseFuture.setResponseCommand(cmd);
 
-            responseTable.remove(opaque);
-
             if (responseFuture.getInvokeCallback() != null) {
                 executeInvokeCallback(responseFuture);
             } else {
@@ -564,10 +562,12 @@ public abstract class NettyRemotingAbstract {
             ResponseFuture rep = next.getValue();
 
             if ((rep.getBeginTimestamp() + rep.getTimeoutMillis() + 1000) <= 
System.currentTimeMillis()) {
-                rep.release();
-                it.remove();
-                rfList.add(rep);
-                log.warn("remove timeout request, " + rep);
+                ResponseFuture removed = responseTable.remove(next.getKey());
+                if (removed != null) {
+                    removed.release();
+                    rfList.add(removed);
+                    log.warn("remove timeout request, " + removed);
+                }
             }
         }