[ISSUE #9416] Fix batch send messages have the same message id when consumed (#9417)

Author: qianye1001

store/src/main/java/org/apache/rocketmq/store/MessageExtEncoder.java

@@ -290,15 +290,6 @@ public ByteBuffer encode(final MessageExtBatch messageExtBatch, PutMessageContex
             throw new RuntimeException("message body size exceeded");
         }
 
-        // properties from MessageExtBatch
-        String batchPropStr = MessageDecoder.messageProperties2String(messageExtBatch.getProperties());
-        final byte[] batchPropData = batchPropStr.getBytes(MessageDecoder.CHARSET_UTF8);
-        int batchPropDataLen = batchPropData.length;
-        if (batchPropDataLen > Short.MAX_VALUE) {
-            CommitLog.log.warn("Properties size of messageExtBatch exceeded, properties size: {}, maxSize: {}.", batchPropDataLen, Short.MAX_VALUE);
-            throw new RuntimeException("Properties size of messageExtBatch exceeded!");
-        }
-        final short batchPropLen = (short) batchPropDataLen;
 
         int batchSize = 0;
         while (messagesByteBuff.hasRemaining()) {
@@ -320,14 +311,11 @@ public ByteBuffer encode(final MessageExtBatch messageExtBatch, PutMessageContex
             short propertiesLen = messagesByteBuff.getShort();
             int propertiesPos = messagesByteBuff.position();
             messagesByteBuff.position(propertiesPos + propertiesLen);
-            boolean needAppendLastPropertySeparator = propertiesLen > 0 && batchPropLen > 0
-                && messagesByteBuff.get(messagesByteBuff.position() - 1) != MessageDecoder.PROPERTY_SEPARATOR;
 
             final byte[] topicData = messageExtBatch.getTopic().getBytes(MessageDecoder.CHARSET_UTF8);
 
             final int topicLength = topicData.length;
-            int totalPropLen = needAppendLastPropertySeparator ?
-                propertiesLen + batchPropLen + 1 : propertiesLen + batchPropLen;
+            int totalPropLen = propertiesLen;
 
             // properties need to add crc32
             totalPropLen += crc32ReservedLength;
@@ -386,12 +374,6 @@ public ByteBuffer encode(final MessageExtBatch messageExtBatch, PutMessageContex
             if (propertiesLen > 0) {
                 this.byteBuf.writeBytes(messagesByteBuff.array(), propertiesPos, propertiesLen);
             }
-            if (batchPropLen > 0) {
-                if (needAppendLastPropertySeparator) {
-                    this.byteBuf.writeByte((byte) MessageDecoder.PROPERTY_SEPARATOR);
-                }
-                this.byteBuf.writeBytes(batchPropData, 0, batchPropLen);
-            }
             this.byteBuf.writerIndex(this.byteBuf.writerIndex() + crc32ReservedLength);
         }
         putMessageContext.setBatchSize(batchSize);

store/src/test/java/org/apache/rocketmq/store/AppendCallbackTest.java

@@ -17,14 +17,16 @@
 
 package org.apache.rocketmq.store;
 
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
+
 import java.io.File;
 import java.net.InetSocketAddress;
 import java.nio.ByteBuffer;
 import java.util.ArrayList;
 import java.util.HashSet;
 import java.util.List;
 import java.util.Set;
-
 import java.util.concurrent.ConcurrentHashMap;
 import org.apache.rocketmq.common.BrokerConfig;
 import org.apache.rocketmq.common.UtilAll;
@@ -37,9 +39,6 @@
 import org.junit.Before;
 import org.junit.Test;
 
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertTrue;
-
 public class AppendCallbackTest {
 
     AppendMessageCallback callback;

store/src/test/java/org/apache/rocketmq/store/BatchPutMessageTest.java

@@ -17,6 +17,12 @@
 
 package org.apache.rocketmq.store;
 
+import static java.util.concurrent.TimeUnit.SECONDS;
+import static org.apache.rocketmq.common.message.MessageDecoder.messageProperties2String;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.awaitility.Awaitility.await;
+import static org.junit.Assert.assertTrue;
+
 import java.io.File;
 import java.net.InetSocketAddress;
 import java.nio.charset.Charset;
@@ -39,12 +45,6 @@
 import org.junit.Before;
 import org.junit.Test;
 
-import static java.util.concurrent.TimeUnit.SECONDS;
-import static org.apache.rocketmq.common.message.MessageDecoder.messageProperties2String;
-import static org.assertj.core.api.Assertions.assertThat;
-import static org.awaitility.Awaitility.await;
-import static org.junit.Assert.assertTrue;
-
 public class BatchPutMessageTest {
 
     private MessageStore messageStore;
@@ -108,7 +108,7 @@ public void testPutMessages() throws Exception {
             short propertiesLength = (short) propertiesBytes.length;
             final byte[] topicData = msg.getTopic().getBytes(MessageDecoder.CHARSET_UTF8);
             final int topicLength = topicData.length;
-            msgLengthArr[j] = calMsgLength(msg.getBody().length, topicLength, propertiesLength + batchPropLen) + msgLengthArr[j - 1];
+            msgLengthArr[j] = calMsgLength(msg.getBody().length, topicLength, propertiesLength) + msgLengthArr[j - 1];
             j++;
         }
         byte[] batchMessageBody = MessageDecoder.encodeMessages(messages);