From fb071a3c0a863f54f1be631016cee19686e0029d Mon Sep 17 00:00:00 2001 From: liuhy Date: Fri, 31 Jul 2026 01:08:23 -0700 Subject: [PATCH 1/3] [ISSUE #10718] Summarize removed client settings logs --- .../v2/common/GrpcClientSettingsManager.java | 13 +++++++- .../common/GrpcClientSettingsManagerTest.java | 33 +++++++++++++++++++ 2 files changed, 45 insertions(+), 1 deletion(-) diff --git a/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/v2/common/GrpcClientSettingsManager.java b/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/v2/common/GrpcClientSettingsManager.java index ac87da8c244..65a676179fa 100644 --- a/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/v2/common/GrpcClientSettingsManager.java +++ b/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/v2/common/GrpcClientSettingsManager.java @@ -282,7 +282,8 @@ protected void onWaitEnd() { consumerGroup ); if (consumerGroupInfo == null || consumerGroupInfo.findChannel(clientId) == null) { - log.info("remove unused grpc client settings. group:{}, settings:{}", consumerGroupInfo, settings); + log.info("remove unused grpc client settings. group:{}, clientId:{}, settingsSummary:{}", + consumerGroup, clientId, summarizeClientSettings(settings)); return null; } return settings; @@ -292,4 +293,14 @@ protected void onWaitEnd() { } } } + + static String summarizeClientSettings(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); + } } diff --git a/proxy/src/test/java/org/apache/rocketmq/proxy/grpc/v2/common/GrpcClientSettingsManagerTest.java b/proxy/src/test/java/org/apache/rocketmq/proxy/grpc/v2/common/GrpcClientSettingsManagerTest.java index 4d0037a272a..ad318f24240 100644 --- a/proxy/src/test/java/org/apache/rocketmq/proxy/grpc/v2/common/GrpcClientSettingsManagerTest.java +++ b/proxy/src/test/java/org/apache/rocketmq/proxy/grpc/v2/common/GrpcClientSettingsManagerTest.java @@ -40,8 +40,10 @@ import org.junit.Test; import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; import static org.junit.Assert.assertNotEquals; import static org.junit.Assert.assertNull; +import static org.junit.Assert.assertTrue; import static org.mockito.ArgumentMatchers.any; import static org.mockito.ArgumentMatchers.anyLong; import static org.mockito.ArgumentMatchers.anyString; @@ -156,6 +158,37 @@ public void testOfflineClientLiteSubscription_SettingsNotNull_NotLiteConsumer() verify(messagingProcessor, never()).syncLiteSubscription(any(), any(), anyLong()); } + @Test + public void testSummarizeClientSettingsDoesNotExposeResourceNames() { + 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() + .setTopic(Resource.newBuilder().setName("sensitive-subscription-topic").build()) + .build()) + .build()) + .build(); + + String producerSummary = GrpcClientSettingsManager.summarizeClientSettings(producerSettings); + String consumerSummary = GrpcClientSettingsManager.summarizeClientSettings(consumerSettings); + + assertTrue(producerSummary.contains("clientType=PRODUCER")); + assertTrue(producerSummary.contains("publishingTopicCount=1")); + assertFalse(producerSummary.contains("sensitive-publish-topic")); + + assertTrue(consumerSummary.contains("clientType=PUSH_CONSUMER")); + assertTrue(consumerSummary.contains("subscriptionCount=1")); + assertFalse(consumerSummary.contains("sensitive-subscription-topic")); + assertFalse(consumerSummary.contains("sensitive-group")); + } + @Test public void testOfflineClientLiteSubscription_ValidLiteConsumer_Success() { Subscription subscription = Subscription.newBuilder() From 7dce94ea054d45c2f04a4d4194c68ae6e62ae052 Mon Sep 17 00:00:00 2001 From: liuhy Date: Thu, 30 Jul 2026 23:51:35 -0700 Subject: [PATCH 2/3] [ISSUE #10710] Sanitize lite offline Settings failure logs --- .../v2/common/GrpcClientSettingsManager.java | 25 ++++++++++++++++- .../common/GrpcClientSettingsManagerTest.java | 27 +++++++++++++++++++ 2 files changed, 51 insertions(+), 1 deletion(-) diff --git a/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/v2/common/GrpcClientSettingsManager.java b/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/v2/common/GrpcClientSettingsManager.java index 65a676179fa..a96ff3e8a93 100644 --- a/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/v2/common/GrpcClientSettingsManager.java +++ b/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/v2/common/GrpcClientSettingsManager.java @@ -25,6 +25,7 @@ import apache.rocketmq.v2.ExponentialBackoff; import apache.rocketmq.v2.Metric; import apache.rocketmq.v2.Settings; +import apache.rocketmq.v2.Subscription; import com.google.protobuf.Duration; import com.google.protobuf.util.Durations; import java.util.Arrays; @@ -253,10 +254,32 @@ public void offlineClientLiteSubscription(ProxyContext ctx, String clientId, Set } }); } catch (Exception e) { - log.error("offlineClientLiteSubscription error, clientId:{}, settings:{}", clientId, settings, e); + log.error("offlineClientLiteSubscription error, clientId:{}, settings:{}", + clientId, summarizeLiteSettings(settings), e); } } + protected static String summarizeLiteSettings(Settings settings) { + if (settings == null) { + return "null"; + } + String group = ""; + String topic = ""; + int subscriptionCount = 0; + if (settings.hasSubscription()) { + Subscription subscription = settings.getSubscription(); + group = subscription.getGroup().getName(); + subscriptionCount = subscription.getSubscriptionsCount(); + if (subscriptionCount > 0) { + topic = subscription.getSubscriptions(0).getTopic().getName(); + } + } + return "clientType:" + settings.getClientType() + + ", group:" + group + + ", topic:" + topic + + ", subscriptionCount:" + subscriptionCount; + } + @Override public void run() { while (!this.isStopped()) { diff --git a/proxy/src/test/java/org/apache/rocketmq/proxy/grpc/v2/common/GrpcClientSettingsManagerTest.java b/proxy/src/test/java/org/apache/rocketmq/proxy/grpc/v2/common/GrpcClientSettingsManagerTest.java index ad318f24240..d9f2176eda5 100644 --- a/proxy/src/test/java/org/apache/rocketmq/proxy/grpc/v2/common/GrpcClientSettingsManagerTest.java +++ b/proxy/src/test/java/org/apache/rocketmq/proxy/grpc/v2/common/GrpcClientSettingsManagerTest.java @@ -20,6 +20,7 @@ import apache.rocketmq.v2.ClientType; import apache.rocketmq.v2.CustomizedBackoff; import apache.rocketmq.v2.ExponentialBackoff; +import apache.rocketmq.v2.FilterExpression; import apache.rocketmq.v2.Publishing; import apache.rocketmq.v2.Resource; import apache.rocketmq.v2.RetryPolicy; @@ -234,4 +235,30 @@ public void testOfflineClientLiteSubscription_ValidLiteConsumer_SyncThrowsExcept verify(messagingProcessor, times(1)).syncLiteSubscription(any(), any(LiteSubscriptionDTO.class), anyLong()); } + + @Test + public void testSummarizeLiteSettingsDoesNotLeakSubscriptionDetails() { + Settings settings = Settings.newBuilder() + .setClientType(ClientType.LITE_PUSH_CONSUMER) + .setSubscription(Subscription.newBuilder() + .setGroup(Resource.newBuilder().setName("testGroup").build()) + .addSubscriptions(SubscriptionEntry.newBuilder() + .setTopic(Resource.newBuilder().setName("testTopic").build()) + .setExpression(FilterExpression.newBuilder().setExpression("sensitive-filter").build()) + .build()) + .addSubscriptions(SubscriptionEntry.newBuilder() + .setTopic(Resource.newBuilder().setName("otherTopic").build()) + .build()) + .build()) + .build(); + + String summary = GrpcClientSettingsManager.summarizeLiteSettings(settings); + + assertTrue(summary.contains("clientType:LITE_PUSH_CONSUMER")); + assertTrue(summary.contains("group:testGroup")); + assertTrue(summary.contains("topic:testTopic")); + assertTrue(summary.contains("subscriptionCount:2")); + assertFalse(summary.contains("sensitive-filter")); + assertFalse(summary.contains("otherTopic")); + } } From 81e7b001c2468f63e8d65abdc121b1f9fc3c1fe3 Mon Sep 17 00:00:00 2001 From: liuhy Date: Sun, 16 Aug 2026 01:13:48 -0700 Subject: [PATCH 3/3] fix: skip offline Lite cleanup without subscriptions --- .../grpc/v2/common/GrpcClientSettingsManager.java | 5 +++++ .../v2/common/GrpcClientSettingsManagerTest.java | 14 ++++++++++++++ 2 files changed, 19 insertions(+) diff --git a/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/v2/common/GrpcClientSettingsManager.java b/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/v2/common/GrpcClientSettingsManager.java index a96ff3e8a93..8158380043c 100644 --- a/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/v2/common/GrpcClientSettingsManager.java +++ b/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/v2/common/GrpcClientSettingsManager.java @@ -237,6 +237,11 @@ public void offlineClientLiteSubscription(ProxyContext ctx, String clientId, Set && ClientType.LITE_SIMPLE_CONSUMER != settings.getClientType()) { return; } + if (!settings.hasSubscription() || settings.getSubscription().getSubscriptionsCount() == 0) { + log.warn("skip offline lite subscription cleanup because subscriptions are missing. clientId:{}, settings:{}", + clientId, summarizeLiteSettings(settings)); + return; + } try { String topic = settings.getSubscription().getSubscriptions(0).getTopic().getName(); String group = settings.getSubscription().getGroup().getName(); diff --git a/proxy/src/test/java/org/apache/rocketmq/proxy/grpc/v2/common/GrpcClientSettingsManagerTest.java b/proxy/src/test/java/org/apache/rocketmq/proxy/grpc/v2/common/GrpcClientSettingsManagerTest.java index d9f2176eda5..d04e8141774 100644 --- a/proxy/src/test/java/org/apache/rocketmq/proxy/grpc/v2/common/GrpcClientSettingsManagerTest.java +++ b/proxy/src/test/java/org/apache/rocketmq/proxy/grpc/v2/common/GrpcClientSettingsManagerTest.java @@ -159,6 +159,20 @@ public void testOfflineClientLiteSubscription_SettingsNotNull_NotLiteConsumer() verify(messagingProcessor, never()).syncLiteSubscription(any(), any(), anyLong()); } + @Test + public void testOfflineClientLiteSubscription_LiteConsumerWithoutSubscriptions() { + Settings settings = Settings.newBuilder() + .setClientType(ClientType.LITE_PUSH_CONSUMER) + .setSubscription(Subscription.newBuilder() + .setGroup(Resource.newBuilder().setName("testGroup").build()) + .build()) + .build(); + + grpcClientSettingsManager.offlineClientLiteSubscription(ctx, clientId, settings); + + verify(messagingProcessor, never()).syncLiteSubscription(any(), any(), anyLong()); + } + @Test public void testSummarizeClientSettingsDoesNotExposeResourceNames() { Settings producerSettings = Settings.newBuilder()