[ISSUE #10976] Reduce per-message allocation on the proxy gRPC path

.gitignore

@@ -20,3 +20,8 @@ bazel-testlogs
 .vscode
 MODULE.bazel.lock
 *.flattened-pom.xml
+.qoder/
+profiling-docs/
+profiling-tools/
+.gorepro/
+.gotmp/

common/src/main/java/org/apache/rocketmq/common/utils/BinaryUtil.java

@@ -23,13 +23,21 @@
 import org.apache.commons.codec.binary.Hex;
 
 public class BinaryUtil {
-    public static byte[] calculateMd5(byte[] binaryData) {
-        MessageDigest messageDigest = null;
+    /**
+     * MessageDigest is not thread safe, so keep one instance per thread instead of
+     * looking it up from the provider on every call.
+     */
+    private static final ThreadLocal<MessageDigest> MD5_DIGEST = ThreadLocal.withInitial(() -> {
         try {
-            messageDigest = MessageDigest.getInstance("MD5");
+            return MessageDigest.getInstance("MD5");
         } catch (NoSuchAlgorithmException e) {
-            throw new RuntimeException("MD5 algorithm not found.");
+            throw new RuntimeException("MD5 algorithm not found.", e);
         }
+    });
+
+    public static byte[] calculateMd5(byte[] binaryData) {
+        MessageDigest messageDigest = MD5_DIGEST.get();
+        messageDigest.reset();
         messageDigest.update(binaryData);
         return messageDigest.digest();
     }

common/src/test/java/org/apache/rocketmq/common/utils/BinaryUtilTest.java

@@ -0,0 +1,49 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.rocketmq.common.utils;
+
+import java.nio.charset.StandardCharsets;
+import java.security.MessageDigest;
+import org.apache.commons.codec.binary.Hex;
+import org.junit.Test;
+
+import static org.junit.Assert.assertEquals;
+
+public class BinaryUtilTest {
+
+    @Test
+    public void testGenerateMd5MatchesFreshDigestAndIsStableAcrossCalls() throws Exception {
+        byte[] payload = "rocketmq-md5-test".getBytes(StandardCharsets.UTF_8);
+        String expected = Hex.encodeHexString(MessageDigest.getInstance("MD5").digest(payload), false);
+
+        String first = BinaryUtil.generateMd5(payload);
+        // interleave another digest to verify the reused per-thread instance is reset between calls
+        BinaryUtil.generateMd5("another-payload".getBytes(StandardCharsets.UTF_8));
+        String second = BinaryUtil.generateMd5(payload);
+
+        assertEquals(expected, first);
+        assertEquals(expected, second);
+    }
+
+    @Test
+    public void testGenerateMd5FromString() throws Exception {
+        String body = "string-body-\u4e2d\u6587";
+        String expected = Hex.encodeHexString(
+            MessageDigest.getInstance("MD5").digest(body.getBytes(StandardCharsets.UTF_8)), false);
+        assertEquals(expected, BinaryUtil.generateMd5(body));
+    }
+}

proxy/src/main/java/org/apache/rocketmq/proxy/grpc/v2/producer/SendMessageActivity.java

@@ -29,7 +29,6 @@
 import com.google.protobuf.Timestamp;
 import com.google.protobuf.util.Durations;
 import com.google.protobuf.util.Timestamps;
-import java.nio.charset.StandardCharsets;
 import java.util.ArrayList;
 import java.util.HashSet;
 import java.util.List;
@@ -166,7 +165,7 @@ protected void validateMessageGroup(String messageGroup) {
             if (maxSize <= 0) {
                 return;
             }
-            if (messageGroup.getBytes(StandardCharsets.UTF_8).length >= maxSize) {
+            if (utf8Length(messageGroup) >= maxSize) {
                 throw new GrpcProxyException(Code.ILLEGAL_MESSAGE_GROUP, "message group exceed the max size " + maxSize);
             }
             if (GrpcValidator.getInstance().containControlCharacter(messageGroup)) {
@@ -214,22 +213,22 @@ protected Map<String, String> buildMessageProperty(ProxyContext context, apache.
             if (GrpcValidator.getInstance().containControlCharacter(userPropertiesEntry.getValue())) {
                 throw new GrpcProxyException(Code.ILLEGAL_MESSAGE_PROPERTY_KEY, "the value of property cannot contain control character");
             }
-            userPropertySize += userPropertiesEntry.getKey().getBytes(StandardCharsets.UTF_8).length;
-            userPropertySize += userPropertiesEntry.getValue().getBytes(StandardCharsets.UTF_8).length;
+            userPropertySize += utf8Length(userPropertiesEntry.getKey());
+            userPropertySize += utf8Length(userPropertiesEntry.getValue());
         }
         MessageAccessor.setProperties(messageWithHeader, Maps.newHashMap(userProperties));
 
         // set tag
         String tag = message.getSystemProperties().getTag();
         GrpcValidator.getInstance().validateTag(tag);
         messageWithHeader.setTags(tag);
-        userPropertySize += tag.getBytes(StandardCharsets.UTF_8).length;
+        userPropertySize += utf8Length(tag);
 
         // set keys
         List<String> keysList = message.getSystemProperties().getKeysList();
         for (String key : keysList) {
             validateMessageKey(key);
-            userPropertySize += key.getBytes(StandardCharsets.UTF_8).length;
+            userPropertySize += utf8Length(key);
         }
         if (keysList.size() > 0) {
             messageWithHeader.setKeys(keysList);
@@ -311,6 +310,35 @@ protected Map<String, String> buildMessageProperty(ProxyContext context, apache.
         return messageWithHeader.getProperties();
     }
 
+    /**
+     * Length of the UTF-8 encoding of {@code str} without materializing the byte array.
+     * Matches {@code str.getBytes(StandardCharsets.UTF_8).length}, including the
+     * single-byte replacement for unpaired surrogates.
+     */
+    static int utf8Length(String str) {
+        int len = 0;
+        for (int i = 0; i < str.length(); i++) {
+            char c = str.charAt(i);
+            if (c < 0x80) {
+                len += 1;
+            } else if (c < 0x800) {
+                len += 2;
+            } else if (Character.isHighSurrogate(c)) {
+                if (i + 1 < str.length() && Character.isLowSurrogate(str.charAt(i + 1))) {
+                    len += 4;
+                    i++;
+                } else {
+                    len += 1;
+                }
+            } else if (Character.isLowSurrogate(c)) {
+                len += 1;
+            } else {
+                len += 3;
+            }
+        }
+        return len;
+    }
+
     protected void fillDelayMessageProperty(apache.rocketmq.v2.Message message, org.apache.rocketmq.common.message.Message messageWithHeader) {
         if (message.getSystemProperties().hasDeliveryTimestamp()) {
             Timestamp deliveryTimestamp = message.getSystemProperties().getDeliveryTimestamp();

proxy/src/test/java/org/apache/rocketmq/proxy/grpc/v2/producer/SendMessageActivityTest.java

@@ -28,6 +28,7 @@
 import com.google.protobuf.ByteString;
 import com.google.protobuf.util.Durations;
 import com.google.protobuf.util.Timestamps;
+import java.nio.charset.StandardCharsets;
 import java.time.Duration;
 import java.util.HashMap;
 import java.util.Map;
@@ -89,6 +90,23 @@ public void before() throws Throwable {
         this.sendMessageActivity = new SendMessageActivity(messagingProcessor, grpcClientSettingsManager, grpcChannelManager);
     }
 
+    @Test
+    public void testUtf8Length() {
+        String[] samples = {
+            "",
+            "abc123",
+            "\u4e2d\u6587\u5c5e\u6027\u503c",
+            "mixed\u4e2d\u6587and😀emoji",
+            "\uD83D\uDE00",
+            "\uD800",
+            "abc\uDC00def",
+            "߿ࠀ"
+        };
+        for (String sample : samples) {
+            assertEquals(sample, sample.getBytes(StandardCharsets.UTF_8).length, SendMessageActivity.utf8Length(sample));
+        }
+    }
+
     @Test
     public void sendMessage() throws Exception {
         String msgId = MessageClientIDSetter.createUniqID();