[ISSUE #10724] Avoid logging full gRPC responses in ResponseWriter

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.Message;
 import io.grpc.Status;
 import io.grpc.StatusRuntimeException;
 import io.grpc.stub.ServerCallStreamObserver;
@@ -52,16 +54,17 @@ public <T> boolean writeResponse(StreamObserver<T> observer, final T response) {
         if (null == response) {
             return false;
         }
-        log.debug("start to write response. response: {}", response);
+        String responseSummary = summarizeResponse(response);
+        log.debug("start to write response. response: {}", responseSummary);
         if (isCancelled(observer)) {
-            log.warn("client has cancelled the request. response to write: {}", response);
+            log.warn("client has cancelled the request. response to write: {}", responseSummary);
             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. response to write: {}", responseSummary);
                 return false;
             }
             throw statusRuntimeException;
@@ -76,5 +79,32 @@ public <T> boolean isCancelled(StreamObserver<T> observer) {
         }
         return false;
     }
-}
 
+    static String summarizeResponse(Object response) {
+        if (response == null) {
+            return "null";
+        }
+        StringBuilder summary = new StringBuilder(response.getClass().getSimpleName());
+        if (response instanceof Message) {
+            appendStatusCode(summary, (Message) response);
+        }
+        return summary.toString();
+    }
+
+    private static void appendStatusCode(StringBuilder summary, Message response) {
+        Descriptors.FieldDescriptor statusField = response.getDescriptorForType().findFieldByName("status");
+        if (statusField == null || !response.hasField(statusField)) {
+            return;
+        }
+        Object status = response.getField(statusField);
+        if (!(status instanceof Message)) {
+            summary.append("{status=").append(status).append('}');
+            return;
+        }
+        Message statusMessage = (Message) status;
+        Descriptors.FieldDescriptor codeField = statusMessage.getDescriptorForType().findFieldByName("code");
+        if (codeField != null) {
+            summary.append("{statusCode=").append(statusMessage.getField(codeField)).append('}');
+        }
+    }
+}

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

@@ -0,0 +1,57 @@
+/*
+ * 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.Resource;
+import apache.rocketmq.v2.SystemProperties;
+import com.google.protobuf.ByteString;
+import org.junit.Test;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+public class ResponseWriterTest {
+
+    @Test
+    public void summarizeResponseShouldNotExposeMessagePayload() {
+        ReceiveMessageResponse response = ReceiveMessageResponse.newBuilder()
+            .setMessage(Message.newBuilder()
+                .setTopic(Resource.newBuilder().setName("topic").build())
+                .setSystemProperties(SystemProperties.newBuilder().setMessageId("message-id").build())
+                .setBody(ByteString.copyFromUtf8("sensitive-body"))
+                .build())
+            .build();
+
+        String summary = ResponseWriter.summarizeResponse(response);
+
+        assertThat(summary).isEqualTo("ReceiveMessageResponse");
+        assertThat(summary).doesNotContain("sensitive-body", "message-id", "topic");
+    }
+
+    @Test
+    public void summarizeResponseShouldKeepStatusCodeWhenAvailable() {
+        ReceiveMessageResponse response = ReceiveMessageResponse.newBuilder()
+            .setStatus(ResponseBuilder.getInstance().buildStatus(Code.OK, Code.OK.name()))
+            .build();
+
+        String summary = ResponseWriter.summarizeResponse(response);
+
+        assertThat(summary).isEqualTo("ReceiveMessageResponse{statusCode=OK}");
+    }
+}