[ISSUE #10724] Avoid raw gRPC response logs

proxy/src/main/java/org/apache/rocketmq/proxy/grpc/v2/common/ResponseWriter.java

@@ -17,6 +17,8 @@
 
 package org.apache.rocketmq.proxy.grpc.v2.common;
 
+import com.google.protobuf.Descriptors;
+import com.google.protobuf.MessageOrBuilder;
 import io.grpc.Status;
 import io.grpc.StatusRuntimeException;
 import io.grpc.stub.ServerCallStreamObserver;
@@ -52,23 +54,60 @@ public <T> boolean writeResponse(StreamObserver<T> observer, final T response) {
         if (null == response) {
             return false;
         }
-        log.debug("start to write response. response: {}", response);
+        if (log.isDebugEnabled()) {
+            log.debug("start to write response. responseSummary: {}", summarizeResponse(response));
+        }
         if (isCancelled(observer)) {
-            log.warn("client has cancelled the request. response to write: {}", response);
+            log.warn("client has cancelled the request. responseSummary: {}", summarizeResponse(response));
             return false;
         }
         try {
             observer.onNext(response);
         } catch (StatusRuntimeException statusRuntimeException) {
             if (Status.CANCELLED.equals(statusRuntimeException.getStatus())) {
-                log.warn("client has cancelled the request. response to write: {}", response);
+                log.warn("client has cancelled the request. responseSummary: {}", summarizeResponse(response));
                 return false;
             }
             throw statusRuntimeException;
         }
         return true;
     }
 
+    static String summarizeResponse(Object response) {
+        if (!(response instanceof MessageOrBuilder)) {
+            return response.getClass().getSimpleName();
+        }
+        MessageOrBuilder message = (MessageOrBuilder) response;
+        StringBuilder summary = new StringBuilder()
+            .append("type=")
+            .append(message.getDescriptorForType().getFullName());
+        appendStatusCode(summary, message);
+        return summary.toString();
+    }
+
+    private static void appendStatusCode(StringBuilder summary, MessageOrBuilder message) {
+        Descriptors.FieldDescriptor statusField = message.getDescriptorForType().findFieldByName("status");
+        if (statusField == null || statusField.isRepeated()
+            || statusField.getJavaType() != Descriptors.FieldDescriptor.JavaType.MESSAGE
+            || !message.hasField(statusField)) {
+            return;
+        }
+        Object status = message.getField(statusField);
+        if (!(status instanceof MessageOrBuilder)) {
+            return;
+        }
+        MessageOrBuilder statusMessage = (MessageOrBuilder) status;
+        Descriptors.FieldDescriptor codeField = statusMessage.getDescriptorForType().findFieldByName("code");
+        if (codeField == null || codeField.isRepeated()
+            || codeField.getJavaType() != Descriptors.FieldDescriptor.JavaType.ENUM) {
+            return;
+        }
+        Object code = statusMessage.getField(codeField);
+        if (code instanceof Descriptors.EnumValueDescriptor) {
+            summary.append(", statusCode=").append(((Descriptors.EnumValueDescriptor) code).getName());
+        }
+    }
+
     public <T> boolean isCancelled(StreamObserver<T> observer) {
         if (observer instanceof ServerCallStreamObserver) {
             final ServerCallStreamObserver<T> serverCallStreamObserver = (ServerCallStreamObserver<T>) observer;
@@ -77,4 +116,3 @@ public <T> boolean isCancelled(StreamObserver<T> observer) {
         return false;
     }
 }
-

proxy/src/test/java/org/apache/rocketmq/proxy/grpc/v2/common/ResponseWriterTest.java

@@ -0,0 +1,55 @@
+/*
+ * 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.proxy.grpc.v2.common;
+
+import apache.rocketmq.v2.Code;
+import apache.rocketmq.v2.Message;
+import apache.rocketmq.v2.ReceiveMessageResponse;
+import apache.rocketmq.v2.Status;
+import com.google.protobuf.ByteString;
+import org.junit.Test;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+public class ResponseWriterTest {
+
+    @Test
+    public void testSummarizeResponseDoesNotExposeMessagePayload() {
+        ReceiveMessageResponse response = ReceiveMessageResponse.newBuilder()
+            .setMessage(Message.newBuilder().setBody(ByteString.copyFromUtf8("secret-payload")).build())
+            .build();
+
+        String summary = ResponseWriter.summarizeResponse(response);
+
+        assertThat(summary).contains("ReceiveMessageResponse");
+        assertThat(summary).doesNotContain("secret-payload");
+        assertThat(summary).doesNotContain("body");
+    }
+
+    @Test
+    public void testSummarizeResponseIncludesStatusCode() {
+        ReceiveMessageResponse response = ReceiveMessageResponse.newBuilder()
+            .setStatus(Status.newBuilder().setCode(Code.OK).setMessage("OK").build())
+            .build();
+
+        String summary = ResponseWriter.summarizeResponse(response);
+
+        assertThat(summary).contains("ReceiveMessageResponse");
+        assertThat(summary).contains("statusCode=OK");
+    }
+}