[ISSUE apache#8426]Fix when SendMessage asyncSendEnable is false will…

… write response twice
mxsm · Jul 22, 2024 · c0980e8 · c0980e8
1 parent 86d59d2
commit c0980e8
Showing 1 changed file with 5 additions and 4 deletions.
diff --git a/broker/src/main/java/org/apache/rocketmq/broker/processor/SendMessageProcessor.java b/broker/src/main/java/org/apache/rocketmq/broker/processor/SendMessageProcessor.java
@@ -353,13 +353,14 @@ public RemotingCommand sendMessage(final ChannelHandlerContext ctx,
             } else {
                 putMessageResult = this.brokerController.getMessageStore().putMessage(msgInner);
             }
-            handlePutMessageResult(putMessageResult, response, request, msgInner, responseHeader, sendMessageContext, ctx, queueIdInt, beginTimeMillis, mappingContext, BrokerMetricsManager.getMessageType(requestHeader));
+            final RemotingCommand handledResponse = handlePutMessageResult(putMessageResult, response, request, msgInner, responseHeader,
+                sendMessageContext, ctx, queueIdInt, beginTimeMillis, mappingContext, BrokerMetricsManager.getMessageType(requestHeader));
             // record the transaction metrics
             if (putMessageResult.getPutMessageStatus() == PutMessageStatus.PUT_OK && putMessageResult.getAppendMessageResult().isOk()) {
                 this.brokerController.getTransactionalMessageService().getTransactionMetrics().addAndGet(msgInner.getProperty(MessageConst.PROPERTY_REAL_TOPIC), 1);
             }
             sendMessageCallback.onComplete(sendMessageContext, response);
-            return response;
+            return handledResponse;
         }
     }
 
@@ -640,10 +641,10 @@ private RemotingCommand sendBatchMessage(final ChannelHandlerContext ctx,
             } else {
                 putMessageResult = this.brokerController.getMessageStore().putMessages(messageExtBatch);
             }
-            handlePutMessageResult(putMessageResult, response, request, messageExtBatch, responseHeader,
+            final RemotingCommand handledResponse = handlePutMessageResult(putMessageResult, response, request, messageExtBatch, responseHeader,
                 sendMessageContext, ctx, queueIdInt, beginTimeMillis, mappingContext, BrokerMetricsManager.getMessageType(requestHeader));
             sendMessageCallback.onComplete(sendMessageContext, response);
-            return response;
+            return handledResponse;
         }
     }