From 156b3cfe9d66d317faeffa8ace038800e01b2692 Mon Sep 17 00:00:00 2001 From: liuhy Date: Fri, 31 Jul 2026 01:06:03 -0700 Subject: [PATCH 1/2] [ISSUE #10716] Summarize synced client settings logs --- .../proxy/grpc/v2/client/ClientActivity.java | 13 ++++++- .../grpc/v2/client/ClientActivityTest.java | 36 +++++++++++++++++++ 2 files changed, 48 insertions(+), 1 deletion(-) diff --git a/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/v2/client/ClientActivity.java b/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/v2/client/ClientActivity.java index abc23a53a3e..5fe8ed3500e 100644 --- a/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/v2/client/ClientActivity.java +++ b/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/v2/client/ClientActivity.java @@ -90,6 +90,16 @@ public ClientActivity(MessagingProcessor messagingProcessor, this.init(); } + static String summarizeSettings(Settings settings) { + if (settings == null) { + return "null"; + } + int publishingTopicCount = settings.hasPublishing() ? settings.getPublishing().getTopicsCount() : 0; + int subscriptionCount = settings.hasSubscription() ? settings.getSubscription().getSubscriptionsCount() : 0; + return String.format("clientType=%s, publishingTopicCount=%d, subscriptionCount=%d", + settings.getClientType(), publishingTopicCount, subscriptionCount); + } + protected void init() { this.messagingProcessor.registerConsumerListener(new ConsumerIdsChangeListenerImpl()); this.messagingProcessor.registerProducerListener(new ProducerChangeListenerImpl()); @@ -597,7 +607,8 @@ protected void processClientRegister(String group, Object... args) { if (ChannelHelper.isRemote(channel)) { // save settings from channel sync from other proxy Settings settings = GrpcClientChannel.parseChannelExtendAttribute(channel); - log.debug("save client settings sync from other proxy. group:{}, channelInfo:{}, settings:{}", group, clientChannelInfo, settings); + log.debug("save client settings sync from other proxy. group:{}, channelInfo:{}, settingsSummary:{}", + group, clientChannelInfo, summarizeSettings(settings)); if (settings == null) { return; } diff --git a/proxy/src/test/java/org/apache/rocketmq/proxy/grpc/v2/client/ClientActivityTest.java b/proxy/src/test/java/org/apache/rocketmq/proxy/grpc/v2/client/ClientActivityTest.java index e215c6efaba..eae3e59601f 100644 --- a/proxy/src/test/java/org/apache/rocketmq/proxy/grpc/v2/client/ClientActivityTest.java +++ b/proxy/src/test/java/org/apache/rocketmq/proxy/grpc/v2/client/ClientActivityTest.java @@ -184,6 +184,42 @@ protected HeartbeatResponse sendConsumerHeartbeat(ProxyContext context) throws T .build()).get(); } + @Test + public void testSummarizeSettingsDoesNotExposeResourceNames() { + Settings producerSettings = Settings.newBuilder() + .setClientType(ClientType.PRODUCER) + .setPublishing(Publishing.newBuilder() + .addTopics(Resource.newBuilder().setName("sensitive-publish-topic").build()) + .build()) + .build(); + Settings consumerSettings = Settings.newBuilder() + .setClientType(ClientType.PUSH_CONSUMER) + .setSubscription(Subscription.newBuilder() + .setGroup(Resource.newBuilder().setName("sensitive-group").build()) + .addSubscriptions(SubscriptionEntry.newBuilder() + .setExpression(FilterExpression.newBuilder() + .setExpression("sensitive-tag") + .setType(FilterType.TAG) + .build()) + .setTopic(Resource.newBuilder().setName("sensitive-subscription-topic").build()) + .build()) + .build()) + .build(); + + String producerSummary = ClientActivity.summarizeSettings(producerSettings); + String consumerSummary = ClientActivity.summarizeSettings(consumerSettings); + + assertThat(producerSummary).contains("clientType=PRODUCER"); + assertThat(producerSummary).contains("publishingTopicCount=1"); + assertThat(producerSummary).doesNotContain("sensitive-publish-topic"); + + assertThat(consumerSummary).contains("clientType=PUSH_CONSUMER"); + assertThat(consumerSummary).contains("subscriptionCount=1"); + assertThat(consumerSummary).doesNotContain("sensitive-subscription-topic"); + assertThat(consumerSummary).doesNotContain("sensitive-group"); + assertThat(consumerSummary).doesNotContain("sensitive-tag"); + } + @Test public void testConsumerHeartbeat() throws Throwable { ProxyContext context = createContext(); From 5214e17cc239a54d792c34d750dd85e04fba2469 Mon Sep 17 00:00:00 2001 From: liuhy Date: Tue, 28 Jul 2026 20:50:15 -0700 Subject: [PATCH 2/2] [ISSUE #10676] Sanitize telemetry exception logs --- .../proxy/grpc/v2/client/ClientActivity.java | 30 ++++++++++++++++++- .../grpc/v2/client/ClientActivityTest.java | 20 +++++++++++++ 2 files changed, 49 insertions(+), 1 deletion(-) diff --git a/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/v2/client/ClientActivity.java b/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/v2/client/ClientActivity.java index 5fe8ed3500e..a62493c75c7 100644 --- a/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/v2/client/ClientActivity.java +++ b/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/v2/client/ClientActivity.java @@ -372,11 +372,39 @@ protected void processTelemetryException(TelemetryCommand request, Throwable t, } } if (exception.getStatus().getCode().equals(io.grpc.Status.Code.INTERNAL)) { - log.warn("process client telemetryCommand failed. request:{}", request, t); + log.warn("process client telemetryCommand failed. request:{}", summarizeTelemetryCommand(request), t); } responseObserver.onError(exception); } + static String summarizeTelemetryCommand(TelemetryCommand request) { + if (request == null) { + return "null"; + } + + StringBuilder builder = new StringBuilder(request.getCommandCase().name()); + if (request.hasStatus()) { + builder.append(", statusCode=").append(request.getStatus().getCode()); + } + switch (request.getCommandCase()) { + case SETTINGS: + builder.append(", clientType=").append(request.getSettings().getClientType()) + .append(", pubSubCase=").append(request.getSettings().getPubSubCase()); + break; + case THREAD_STACK_TRACE: + builder.append(", nonce=").append(request.getThreadStackTrace().getNonce()) + .append(", details omitted"); + break; + case VERIFY_MESSAGE_RESULT: + builder.append(", nonce=").append(request.getVerifyMessageResult().getNonce()); + break; + default: + // Add explicit cases for future telemetry requests that carry sensitive details. + break; + } + return builder.toString(); + } + protected void processAndWriteClientSettings(ProxyContext ctx, TelemetryCommand request, StreamObserver responseObserver) { GrpcClientChannel grpcClientChannel = null; diff --git a/proxy/src/test/java/org/apache/rocketmq/proxy/grpc/v2/client/ClientActivityTest.java b/proxy/src/test/java/org/apache/rocketmq/proxy/grpc/v2/client/ClientActivityTest.java index eae3e59601f..d042dd74002 100644 --- a/proxy/src/test/java/org/apache/rocketmq/proxy/grpc/v2/client/ClientActivityTest.java +++ b/proxy/src/test/java/org/apache/rocketmq/proxy/grpc/v2/client/ClientActivityTest.java @@ -71,6 +71,7 @@ import static org.assertj.core.api.Assertions.assertThat; import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; import static org.junit.Assert.assertTrue; import static org.junit.Assert.fail; import static org.mockito.ArgumentMatchers.any; @@ -452,6 +453,25 @@ public void onCompleted() { assertThat(result.getResult().getConsumeResult()).isEqualTo(CMResult.CR_SUCCESS); } + @Test + public void testSummarizeTelemetryCommandDoesNotIncludeThreadStackTrace() { + TelemetryCommand command = TelemetryCommand.newBuilder() + .setThreadStackTrace(ThreadStackTrace.newBuilder() + .setNonce("nonce-1") + .setThreadStackTrace("secret-stack-trace") + .build()) + .setStatus(ResponseBuilder.getInstance().buildStatus(Code.OK, Code.OK.name())) + .build(); + + String summary = ClientActivity.summarizeTelemetryCommand(command); + + assertTrue(summary.contains("THREAD_STACK_TRACE")); + assertTrue(summary.contains("statusCode=OK")); + assertTrue(summary.contains("nonce-1")); + assertTrue(summary.contains("details omitted")); + assertFalse(summary.contains("secret-stack-trace")); + } + protected CompletableFuture sendClientTelemetry(ProxyContext ctx, Settings settings) { when(grpcClientSettingsManager.getClientSettings(any())).thenReturn(settings);