Skip to content

Commit c100a62

Browse files
author
wangjiahua.wjh
committed
[ISSUE #10575] Fix race condition between scanResponseTable and processResponseCommand
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.
1 parent 8242c1e commit c100a62

1 file changed

Lines changed: 7 additions & 7 deletions

File tree

remoting/src/main/java/org/apache/rocketmq/remoting/netty/NettyRemotingAbstract.java

Lines changed: 7 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -467,12 +467,10 @@ private Runnable buildProcessRequestHandler(ChannelHandlerContext ctx, RemotingC
467467
*/
468468
public void processResponseCommand(ChannelHandlerContext ctx, RemotingCommand cmd) {
469469
final int opaque = cmd.getOpaque();
470-
final ResponseFuture responseFuture = responseTable.get(opaque);
470+
final ResponseFuture responseFuture = responseTable.remove(opaque);
471471
if (responseFuture != null) {
472472
responseFuture.setResponseCommand(cmd);
473473

474-
responseTable.remove(opaque);
475-
476474
if (responseFuture.getInvokeCallback() != null) {
477475
executeInvokeCallback(responseFuture);
478476
} else {
@@ -564,10 +562,12 @@ public void scanResponseTable() {
564562
ResponseFuture rep = next.getValue();
565563

566564
if ((rep.getBeginTimestamp() + rep.getTimeoutMillis() + 1000) <= System.currentTimeMillis()) {
567-
rep.release();
568-
it.remove();
569-
rfList.add(rep);
570-
log.warn("remove timeout request, " + rep);
565+
ResponseFuture removed = responseTable.remove(next.getKey());
566+
if (removed != null) {
567+
removed.release();
568+
rfList.add(removed);
569+
log.warn("remove timeout request, " + removed);
570+
}
571571
}
572572
}
573573

0 commit comments

Comments
 (0)