[rocketmq] branch develop updated: [ISSUE #5415] Fix duplicate records (#5416)

[jinrongtong]

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

jinrongtong 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 6af6fedd0 [ISSUE #5415] Fix duplicate records (#5416)
6af6fedd0 is described below

commit 6af6fedd05247b3683b730e75ac6f3e9f1ce0c49
Author: RapperCL <44110731+rappe...@users.noreply.github.com>
AuthorDate: Mon Oct 31 10:44:41 2022 +0800

    [ISSUE #5415] Fix duplicate records (#5416)
    
    * fix duplicate records
    
    * code update
---
 .../broker/transaction/queue/TransactionalMessageBridge.java | 12 +++---------
 1 file changed, 3 insertions(+), 9 deletions(-)

diff --git 
a/broker/src/main/java/org/apache/rocketmq/broker/transaction/queue/TransactionalMessageBridge.java
 
b/broker/src/main/java/org/apache/rocketmq/broker/transaction/queue/TransactionalMessageBridge.java
index 6869d3833..a3bda9b59 100644
--- 
a/broker/src/main/java/org/apache/rocketmq/broker/transaction/queue/TransactionalMessageBridge.java
+++ 
b/broker/src/main/java/org/apache/rocketmq/broker/transaction/queue/TransactionalMessageBridge.java
@@ -302,23 +302,17 @@ public class TransactionalMessageBridge {
     }
 
     public boolean writeOp(Integer queueId,Message message) {
-        MessageQueue opQueue;
-        if (opQueueMap.containsKey(queueId)) {
-            opQueue = opQueueMap.get(queueId);
-        } else {
+        MessageQueue opQueue = opQueueMap.get(queueId);
+        if (opQueue == null) {
             opQueue = getOpQueueByHalf(queueId, 
this.brokerController.getBrokerConfig().getBrokerName());
             MessageQueue oldQueue = opQueueMap.putIfAbsent(queueId, opQueue);
             if (oldQueue != null) {
                 opQueue = oldQueue;
             }
         }
-
+    
         PutMessageResult result = 
putMessageReturnResult(makeOpMessageInner(message, opQueue));
         if (result != null && result.getPutMessageStatus() == 
PutMessageStatus.PUT_OK) {
-            
this.brokerController.getBrokerStatsManager().incTopicPutNums(message.getTopic());
-            
this.brokerController.getBrokerStatsManager().incTopicPutSize(message.getTopic(),
-                    result.getAppendMessageResult().getWroteBytes());
-            this.brokerController.getBrokerStatsManager().incBrokerPutNums();
             return true;
         }