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..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 @@ -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; @@ -236,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(); @@ -253,10 +259,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()) { @@ -282,7 +310,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 +321,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..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 @@ -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; @@ -40,8 +41,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 +159,51 @@ 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() + .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() @@ -201,4 +249,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")); + } }