From e42b3fb980353433391ceee568438dc6b7a87a9a Mon Sep 17 00:00:00 2001 From: liuhy Date: Mon, 3 Aug 2026 08:34:21 -0700 Subject: [PATCH] fix: avoid logging full grpc responses --- .../proxy/grpc/v2/common/ResponseWriter.java | 38 +++++++++++-- .../grpc/v2/common/ResponseWriterTest.java | 57 +++++++++++++++++++ 2 files changed, 91 insertions(+), 4 deletions(-) create mode 100644 proxy/src/test/java/org/apache/rocketmq/proxy/grpc/v2/common/ResponseWriterTest.java diff --git a/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/v2/common/ResponseWriter.java b/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/v2/common/ResponseWriter.java index 3ac3d48410e..65ebf3d9dec 100644 --- a/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/v2/common/ResponseWriter.java +++ b/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 boolean writeResponse(StreamObserver 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 boolean isCancelled(StreamObserver 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('}'); + } + } +} diff --git a/proxy/src/test/java/org/apache/rocketmq/proxy/grpc/v2/common/ResponseWriterTest.java b/proxy/src/test/java/org/apache/rocketmq/proxy/grpc/v2/common/ResponseWriterTest.java new file mode 100644 index 00000000000..46d62047cb1 --- /dev/null +++ b/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}"); + } +}