From ce9c4bca2cf8b05b310f333a9c91a9b1dbf4b477 Mon Sep 17 00:00:00 2001 From: zjncs <18910855655@163.com> Date: Fri, 11 Sep 2026 13:53:19 +0800 Subject: [PATCH] fix(proxy): offline all lite subscriptions and guard empty subscription list offlineClientLiteSubscription only took subscriptions(0) to build the COMPLETE_REMOVE LiteSubscriptionDTO, so a lite consumer subscribing to multiple topics left stale subscriptions on the broker for every topic after the first one. When the subscription list was empty, getSubscriptions(0) threw IndexOutOfBoundsException which was swallowed by the catch block, so the cleanup was silently skipped entirely and the stale lite subscription could trigger LITE_SUBSCRIPTION_QUOTA_EXCEEDED on the next connect. Iterate over all subscription entries and send one COMPLETE_REMOVE per topic; an empty list now simply sends nothing. --- .../v2/common/GrpcClientSettingsManager.java | 35 +++++++++------- .../common/GrpcClientSettingsManagerTest.java | 41 +++++++++++++++++++ 2 files changed, 61 insertions(+), 15 deletions(-) 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..571ee41cbb1 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,8 @@ import apache.rocketmq.v2.ExponentialBackoff; import apache.rocketmq.v2.Metric; import apache.rocketmq.v2.Settings; +import apache.rocketmq.v2.Subscription; +import apache.rocketmq.v2.SubscriptionEntry; import com.google.protobuf.Duration; import com.google.protobuf.util.Durations; import java.util.Arrays; @@ -237,21 +239,24 @@ public void offlineClientLiteSubscription(ProxyContext ctx, String clientId, Set return; } try { - String topic = settings.getSubscription().getSubscriptions(0).getTopic().getName(); - String group = settings.getSubscription().getGroup().getName(); - log.info("offlineClientLiteSubscription, topic:{}, group:{}, clientId:{}", topic, group, clientId); - LiteSubscriptionDTO liteSubscriptionDTO = new LiteSubscriptionDTO() - .setAction(LiteSubscriptionAction.COMPLETE_REMOVE) - .setClientId(clientId) - .setGroup(group) - .setTopic(topic); - this.messagingProcessor.syncLiteSubscription(ctx, liteSubscriptionDTO, java.time.Duration.ofSeconds(2).toMillis()) - .whenComplete((result, throwable) -> { - if (throwable != null) { - log.error("offlineClientLiteSubscription failed, topic:{}, group:{}, clientId:{}", - topic, group, clientId, throwable); - } - }); + Subscription subscription = settings.getSubscription(); + String group = subscription.getGroup().getName(); + for (SubscriptionEntry subscriptionEntry : subscription.getSubscriptionsList()) { + String topic = subscriptionEntry.getTopic().getName(); + log.info("offlineClientLiteSubscription, topic:{}, group:{}, clientId:{}", topic, group, clientId); + LiteSubscriptionDTO liteSubscriptionDTO = new LiteSubscriptionDTO() + .setAction(LiteSubscriptionAction.COMPLETE_REMOVE) + .setClientId(clientId) + .setGroup(group) + .setTopic(topic); + this.messagingProcessor.syncLiteSubscription(ctx, liteSubscriptionDTO, java.time.Duration.ofSeconds(2).toMillis()) + .whenComplete((result, throwable) -> { + if (throwable != null) { + log.error("offlineClientLiteSubscription failed, topic:{}, group:{}, clientId:{}", + topic, group, clientId, throwable); + } + }); + } } catch (Exception e) { log.error("offlineClientLiteSubscription error, clientId:{}, settings:{}", clientId, settings, e); } 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..4e870dfbe6a 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 @@ -201,4 +201,45 @@ public void testOfflineClientLiteSubscription_ValidLiteConsumer_SyncThrowsExcept verify(messagingProcessor, times(1)).syncLiteSubscription(any(), any(LiteSubscriptionDTO.class), anyLong()); } + + @Test + public void testOfflineClientLiteSubscription_MultipleTopics_AllSubscriptionsRemoved() { + Subscription subscription = Subscription.newBuilder() + .setGroup(Resource.newBuilder().setName("testGroup").build()) + .addSubscriptions(SubscriptionEntry.newBuilder() + .setTopic(Resource.newBuilder().setName("testTopic0").build()) + .build()) + .addSubscriptions(SubscriptionEntry.newBuilder() + .setTopic(Resource.newBuilder().setName("testTopic1").build()) + .build()) + .build(); + + Settings settings = Settings.newBuilder() + .setClientType(ClientType.LITE_PUSH_CONSUMER) + .setSubscription(subscription) + .build(); + + when(messagingProcessor.syncLiteSubscription(any(), any(LiteSubscriptionDTO.class), anyLong())) + .thenReturn(CompletableFuture.completedFuture(null)); + + grpcClientSettingsManager.offlineClientLiteSubscription(ctx, clientId, settings); + + verify(messagingProcessor, times(2)).syncLiteSubscription(any(), any(LiteSubscriptionDTO.class), anyLong()); + } + + @Test + public void testOfflineClientLiteSubscription_EmptySubscriptions_NoException() { + Subscription subscription = Subscription.newBuilder() + .setGroup(Resource.newBuilder().setName("testGroup").build()) + .build(); + + Settings settings = Settings.newBuilder() + .setClientType(ClientType.LITE_SIMPLE_CONSUMER) + .setSubscription(subscription) + .build(); + + grpcClientSettingsManager.offlineClientLiteSubscription(ctx, clientId, settings); + + verify(messagingProcessor, never()).syncLiteSubscription(any(), any(), anyLong()); + } }