From 9fe5950fb79d15e08aff323322e4f4acb72022ee Mon Sep 17 00:00:00 2001 From: Rui <1685901819@qq.com> Date: Sat, 15 Aug 2026 22:01:11 +0800 Subject: [PATCH 1/2] [ISSUE #10936] Avoid temporary buffers during batch message encoding Signed-off-by: Rui <1685901819@qq.com> --- .../common/message/MessageDecoder.java | 54 ++++++++++------- .../common/MessageEncodeDecodeTest.java | 58 +++++++++++++++++++ 2 files changed, 91 insertions(+), 21 deletions(-) diff --git a/common/src/main/java/org/apache/rocketmq/common/message/MessageDecoder.java b/common/src/main/java/org/apache/rocketmq/common/message/MessageDecoder.java index 713f9405ea9..b2cf4eaf6c8 100644 --- a/common/src/main/java/org/apache/rocketmq/common/message/MessageDecoder.java +++ b/common/src/main/java/org/apache/rocketmq/common/message/MessageDecoder.java @@ -661,20 +661,29 @@ public static Map string2messageProperties(final String properti } public static byte[] encodeMessage(Message message) { - //only need flag, body, properties byte[] body = message.getBody(); - int bodyLen = body.length; String properties = messageProperties2String(message.getProperties()); byte[] propertiesBytes = properties.getBytes(CHARSET_UTF8); - //note properties length must not more than Short.MAX + int storeSize = encodedMessageSize(body, propertiesBytes); + ByteBuffer byteBuffer = ByteBuffer.allocate(storeSize); + encodeMessage(message, body, propertiesBytes, storeSize, byteBuffer); + return byteBuffer.array(); + } + + private static int encodedMessageSize(byte[] body, byte[] propertiesBytes) { + // Note properties length must not be greater than Short.MAX_VALUE. short propertiesLength = (short) propertiesBytes.length; - int storeSize = 4 // 1 TOTALSIZE + return 4 // 1 TOTALSIZE + 4 // 2 MAGICCOD + 4 // 3 BODYCRC + 4 // 4 FLAG - + 4 + bodyLen // 4 BODY + + 4 + body.length // 5 BODY + 2 + propertiesLength; - ByteBuffer byteBuffer = ByteBuffer.allocate(storeSize); + } + + private static void encodeMessage(Message message, byte[] body, byte[] propertiesBytes, int storeSize, + ByteBuffer byteBuffer) { + short propertiesLength = (short) propertiesBytes.length; // 1 TOTALSIZE byteBuffer.putInt(storeSize); @@ -689,14 +698,12 @@ public static byte[] encodeMessage(Message message) { byteBuffer.putInt(flag); // 5 BODY - byteBuffer.putInt(bodyLen); + byteBuffer.putInt(body.length); byteBuffer.put(body); // 6 properties byteBuffer.putShort(propertiesLength); byteBuffer.put(propertiesBytes); - - return byteBuffer.array(); } public static Message decodeMessage(ByteBuffer byteBuffer) throws Exception { @@ -731,21 +738,26 @@ public static Message decodeMessage(ByteBuffer byteBuffer) throws Exception { } public static byte[] encodeMessages(List messages) { - //TO DO refactor, accumulate in one buffer, avoid copies - List encodedMessages = new ArrayList<>(messages.size()); - int allSize = 0; + byte[][] bodies = new byte[messages.size()][]; + byte[][] propertiesBytes = new byte[messages.size()][]; + int totalSize = 0; + int index = 0; for (Message message : messages) { - byte[] tmp = encodeMessage(message); - encodedMessages.add(tmp); - allSize += tmp.length; + byte[] body = message.getBody(); + byte[] encodedProperties = messageProperties2String(message.getProperties()).getBytes(CHARSET_UTF8); + bodies[index] = body; + propertiesBytes[index++] = encodedProperties; + totalSize += encodedMessageSize(body, encodedProperties); } - byte[] allBytes = new byte[allSize]; - int pos = 0; - for (byte[] bytes : encodedMessages) { - System.arraycopy(bytes, 0, allBytes, pos, bytes.length); - pos += bytes.length; + + ByteBuffer byteBuffer = ByteBuffer.allocate(totalSize); + index = 0; + for (Message message : messages) { + byte[] body = bodies[index]; + byte[] encodedProperties = propertiesBytes[index++]; + encodeMessage(message, body, encodedProperties, encodedMessageSize(body, encodedProperties), byteBuffer); } - return allBytes; + return byteBuffer.array(); } public static List decodeMessages(ByteBuffer byteBuffer) throws Exception { diff --git a/common/src/test/java/org/apache/rocketmq/common/MessageEncodeDecodeTest.java b/common/src/test/java/org/apache/rocketmq/common/MessageEncodeDecodeTest.java index 7c73147e0c7..0c941385d34 100644 --- a/common/src/test/java/org/apache/rocketmq/common/MessageEncodeDecodeTest.java +++ b/common/src/test/java/org/apache/rocketmq/common/MessageEncodeDecodeTest.java @@ -24,6 +24,7 @@ import org.apache.rocketmq.common.message.MessageDecoder; import org.junit.Test; +import static org.junit.Assert.assertArrayEquals; import static org.junit.Assert.assertTrue; public class MessageEncodeDecodeTest { @@ -72,4 +73,61 @@ public void testEncodeDecodeList() throws Exception { } } + + @Test + public void testMessageEncodingWireFormatGoldenBytes() { + Message message = new Message(); + message.setBody(new byte[] {0x01, 0x02}); + message.setFlag(0x01020304); + + byte[] encodedProperties = new byte[0]; + int storeSize = 4 + 4 + 4 + 4 + 4 + message.getBody().length + 2 + encodedProperties.length; + ByteBuffer expected = ByteBuffer.allocate(storeSize); + expected.putInt(storeSize); + expected.putInt(0); + expected.putInt(0); + expected.putInt(message.getFlag()); + expected.putInt(message.getBody().length); + expected.put(message.getBody()); + expected.putShort((short) encodedProperties.length); + expected.put(encodedProperties); + + assertArrayEquals(expected.array(), MessageDecoder.encodeMessage(message)); + assertArrayEquals(expected.array(), MessageDecoder.encodeMessages(Arrays.asList(message))); + } + + @Test + public void testBatchEncodingMatchesConcatenatedSingleMessages() { + Message emptyMessage = new Message(); + emptyMessage.setBody(new byte[0]); + emptyMessage.setFlag(-1); + + Message unicodeMessage = new Message("topic", "payload".getBytes(MessageDecoder.CHARSET_UTF8)); + unicodeMessage.setFlag(7); + unicodeMessage.putUserProperty("region", "华东"); + + byte[] largeBody = new byte[4096]; + Arrays.fill(largeBody, (byte) 0x5A); + Message largeMessage = new Message("topic", largeBody); + largeMessage.setFlag(Integer.MAX_VALUE); + largeMessage.putUserProperty("key", "value"); + largeMessage.putUserProperty("trace", "enabled"); + + List messages = Arrays.asList(emptyMessage, unicodeMessage, largeMessage); + List individuallyEncoded = new ArrayList<>(messages.size()); + int expectedLength = 0; + for (Message message : messages) { + byte[] encoded = MessageDecoder.encodeMessage(message); + individuallyEncoded.add(encoded); + expectedLength += encoded.length; + } + + ByteBuffer expected = ByteBuffer.allocate(expectedLength); + for (byte[] encoded : individuallyEncoded) { + expected.put(encoded); + } + + assertArrayEquals(expected.array(), MessageDecoder.encodeMessages(messages)); + assertArrayEquals(new byte[0], MessageDecoder.encodeMessages(new ArrayList())); + } } From d66b4f88d8195f2ddcd60e0e8cfed1ca36b34e80 Mon Sep 17 00:00:00 2001 From: Rui <1685901819@qq.com> Date: Sat, 15 Aug 2026 22:06:10 +0800 Subject: [PATCH 2/2] [ISSUE #10936] Use Unicode escapes in encoding test Signed-off-by: Rui <1685901819@qq.com> --- .../org/apache/rocketmq/common/MessageEncodeDecodeTest.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/common/src/test/java/org/apache/rocketmq/common/MessageEncodeDecodeTest.java b/common/src/test/java/org/apache/rocketmq/common/MessageEncodeDecodeTest.java index 0c941385d34..1a52458c1b2 100644 --- a/common/src/test/java/org/apache/rocketmq/common/MessageEncodeDecodeTest.java +++ b/common/src/test/java/org/apache/rocketmq/common/MessageEncodeDecodeTest.java @@ -104,7 +104,7 @@ public void testBatchEncodingMatchesConcatenatedSingleMessages() { Message unicodeMessage = new Message("topic", "payload".getBytes(MessageDecoder.CHARSET_UTF8)); unicodeMessage.setFlag(7); - unicodeMessage.putUserProperty("region", "华东"); + unicodeMessage.putUserProperty("region", "\u534e\u4e1c"); byte[] largeBody = new byte[4096]; Arrays.fill(largeBody, (byte) 0x5A);