[ISSUE 3019] Fixed bug with batch-sent message statistics

Signed-off-by: zhangyang <Git_Yang@163.com>
apache · Jun 21, 2021 · 2b236c3 · 2b236c3
1 parent e126310
commit 2b236c3
Showing 1 changed file with 4 additions and 2 deletions.
diff --git a/store/src/main/java/org/apache/rocketmq/store/dledger/DLedgerCommitLog.java b/store/src/main/java/org/apache/rocketmq/store/dledger/DLedgerCommitLog.java
@@ -553,7 +553,7 @@ public PutMessageResult putMessages(final MessageExtBatch messageExtBatch) {
         msgIdBuilder.setLength(0);
         long elapsedTimeInLock;
         long queueOffset;
-        long msgNum = 0;
+        int msgNum = 0;
         try {
             beginTimeInDledgerLock = this.defaultMessageStore.getSystemClock().now();
             queueOffset = getQueueOffsetByKey(encodeResult.queueOffsetKey, tranType);
@@ -592,6 +592,7 @@ public PutMessageResult putMessages(final MessageExtBatch messageExtBatch) {
             elapsedTimeInLock = this.defaultMessageStore.getSystemClock().now() - beginTimeInDledgerLock;
             appendResult = new AppendMessageResult(AppendMessageStatus.PUT_OK, firstWroteOffset, encodeResult.totalMsgLen,
                     msgIdBuilder.toString(), System.currentTimeMillis(), queueOffset, elapsedTimeInLock);
+            appendResult.setMsgNum(msgNum);
             DLedgerCommitLog.this.topicQueueTable.put(encodeResult.queueOffsetKey, queueOffset + msgNum);
         } catch (Exception e) {
             log.error("Put message error", e);
@@ -779,7 +780,7 @@ public CompletableFuture<PutMessageResult> asyncPutMessages(MessageExtBatch mess
         msgIdBuilder.setLength(0);
         long elapsedTimeInLock;
         long queueOffset;
-        long msgNum = 0;
+        int msgNum = 0;
         try {
             beginTimeInDledgerLock = this.defaultMessageStore.getSystemClock().now();
             queueOffset = getQueueOffsetByKey(encodeResult.queueOffsetKey, tranType);
@@ -818,6 +819,7 @@ public CompletableFuture<PutMessageResult> asyncPutMessages(MessageExtBatch mess
             elapsedTimeInLock = this.defaultMessageStore.getSystemClock().now() - beginTimeInDledgerLock;
             appendResult = new AppendMessageResult(AppendMessageStatus.PUT_OK, firstWroteOffset, encodeResult.totalMsgLen,
                     msgIdBuilder.toString(), System.currentTimeMillis(), queueOffset, elapsedTimeInLock);
+            appendResult.setMsgNum(msgNum);
             DLedgerCommitLog.this.topicQueueTable.put(encodeResult.queueOffsetKey, queueOffset + msgNum);
         } catch (Exception e) {
             log.error("Put message error", e);