opensource/rocketmq
3
0
Fork 0
mirror of https://github.com/apache/rocketmq.git synced 2026-08-28 20:09:14 +08:00
Code Issues Packages Projects Releases Wiki Activity

[ISSUE #10575] Fix race condition between scanResponseTable and processResponseCommand (#10576)

Browse Source 
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.wjh@alibaba-inc.com>
This commit is contained in:
Jiahua Wang
2026-08-26 15:12:46 +08:00
committed by GitHub
parent 484b7b812c
commit e348efa66b
1 changed files with 7 additions and 7 deletions
Download Patch File Download Diff File Expand all files Collapse all files
remoting/src/main/java/org/apache/rocketmq/remoting/netty/NettyRemotingAbstract.java
+7 -7
View File
@@ -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);
}
}
}