This is an automated email from the ASF dual-hosted git repository.
RongtongJin pushed a commit to branch develop
in repository https://gitbox.apache.org/repos/asf/rocketmq.git
The following commit(s) were added to refs/heads/develop by this push:
new e1eb99c852 [ISSUE #11203] Fix incorrect consumption mode reported for
wildcard Lite groups (#11204)
e1eb99c852 is described below
commit e1eb99c85296faba93ac43dd8edc8929f34fa126
Author: Quan <[email protected]>
AuthorDate: Mon Oct 12 11:07:25 2026 +0800
[ISSUE #11203] Fix incorrect consumption mode reported for wildcard Lite
groups (#11204)
Wildcard Lite consumer groups consume with clustering semantics, but the
registration path reported them as LITE_SELECTIVE, so the admin/console view
showed an incorrect consumption mode.
Instead of adding a new MessageModel value, derive the Lite identity from
the
client type carried on ProxyContext and normalize the reported model:
- ClientActivity#registerConsumer writes the real client type into the
context
via ctx.setClientType, so downstream no longer infers Lite identity from
the
display-only LITE_SELECTIVE value.
- ClientProcessor replaces validateLiteMode with validateLiteConsumer, which
reads ctx.isLiteConsumer() plus the real (remapped)
SubscriptionGroupConfig to
enforce that Lite clients only register to Lite groups and vice versa, and
normalizes a wildcard Lite group's model to CLUSTERING before
registration.
The overridden method signatures kept by the commercial subclass
(registerConsumer, validateLiteSubTopic) are unchanged.
---
.../proxy/grpc/v2/client/ClientActivity.java | 1 +
.../rocketmq/proxy/processor/ClientProcessor.java | 43 ++++++++--------
.../proxy/processor/ClientProcessorTest.java | 58 ++++++++++++++++++----
3 files changed, 68 insertions(+), 34 deletions(-)
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 abc23a53a3..d704108fb3 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
@@ -430,6 +430,7 @@ public class ClientActivity extends
AbstractMessagingActivity {
protected GrpcClientChannel registerConsumer(ProxyContext ctx, String
consumerGroup, ClientType clientType,
List<SubscriptionEntry> subscriptionEntryList, boolean
updateSubscription) {
+ ctx.setClientType(clientType.name());
String clientId = ctx.getClientID();
LanguageCode languageCode = LanguageCode.valueOf(ctx.getLanguage());
diff --git
a/proxy/src/main/java/org/apache/rocketmq/proxy/processor/ClientProcessor.java
b/proxy/src/main/java/org/apache/rocketmq/proxy/processor/ClientProcessor.java
index c73e66416d..e2373baa03 100644
---
a/proxy/src/main/java/org/apache/rocketmq/proxy/processor/ClientProcessor.java
+++
b/proxy/src/main/java/org/apache/rocketmq/proxy/processor/ClientProcessor.java
@@ -96,10 +96,7 @@ public class ClientProcessor extends AbstractProcessor {
Set<SubscriptionData> subList,
boolean updateSubscription
) {
- validateLiteMode(ctx, consumerGroup, messageModel);
- if (MessageModel.LITE_SELECTIVE == messageModel) {
- validateLiteSubTopic(ctx, consumerGroup, subList);
- }
+ messageModel = validateLiteConsumer(ctx, consumerGroup, messageModel,
subList);
this.serviceManager.getConsumerManager().registerConsumer(
consumerGroup,
clientChannelInfo,
@@ -169,28 +166,28 @@ public class ClientProcessor extends AbstractProcessor {
}
/**
- * Validates the message model for a given consumer group.
- * Ensures that regular groups do not use LITE mode and LITE groups use
LITE mode.
- *
- * @param ctx the proxy context
- * @param group the consumer group name
- * @param messageModel the message model to validate
+ * Ensures the lite/regular consumer identity matches the group binding
and resolves the effective
+ * message model.
*/
- protected void validateLiteMode(ProxyContext ctx, String group,
MessageModel messageModel) {
- String bindTopic = getGroupOrException(ctx, group).getLiteBindTopic();
- if (StringUtils.isEmpty(bindTopic)) {
- // regular group
- if (MessageModel.LITE_SELECTIVE == messageModel) {
- throw new GrpcProxyException(Code.ILLEGAL_CONSUMER_GROUP,
- "regular group cannot use LITE mode: " + group);
- }
- } else {
- // lite group
- if (MessageModel.LITE_SELECTIVE != messageModel) {
- throw new GrpcProxyException(Code.ILLEGAL_CONSUMER_GROUP,
- "lite group must use LITE mode: " + group);
+ protected MessageModel validateLiteConsumer(ProxyContext ctx, String
consumerGroup,
+ MessageModel messageModel, Set<SubscriptionData> subList) {
+ SubscriptionGroupConfig groupConfig = getGroupOrException(ctx,
consumerGroup);
+ boolean liteConsumer = ctx.isLiteConsumer();
+ boolean liteGroup =
StringUtils.isNotEmpty(groupConfig.getLiteBindTopic());
+ if (liteConsumer != liteGroup) {
+ throw new GrpcProxyException(Code.ILLEGAL_CONSUMER_GROUP, liteGroup
+ ? "lite group must use LITE mode: " + consumerGroup
+ : "regular group cannot use LITE mode: " + consumerGroup);
+ }
+ if (liteGroup) {
+ validateLiteSubTopic(ctx, consumerGroup, subList);
+ // A wildcard lite group consumes with clustering semantics, so
report it as CLUSTERING.
+ // The message model is only a display value here, thus safe to
override.
+ if (groupConfig.isWildcardLiteGroup()) {
+ messageModel = MessageModel.CLUSTERING;
}
}
+ return messageModel;
}
protected void validateLiteSubTopic(ProxyContext ctx, String group,
Set<SubscriptionData> subList) {
diff --git
a/proxy/src/test/java/org/apache/rocketmq/proxy/processor/ClientProcessorTest.java
b/proxy/src/test/java/org/apache/rocketmq/proxy/processor/ClientProcessorTest.java
index 6644341e55..8a0d1caaff 100644
---
a/proxy/src/test/java/org/apache/rocketmq/proxy/processor/ClientProcessorTest.java
+++
b/proxy/src/test/java/org/apache/rocketmq/proxy/processor/ClientProcessorTest.java
@@ -17,6 +17,7 @@
package org.apache.rocketmq.proxy.processor;
+import java.util.Collections;
import java.util.HashSet;
import java.util.Set;
import org.apache.rocketmq.proxy.common.ProxyContext;
@@ -62,51 +63,86 @@ public class ClientProcessorTest {
}
@Test
- public void
testValidateLiteMode_regularGroupWithLiteMode_throwsException() {
+ public void
testValidateLiteConsumer_regularGroupWithLiteConsumer_throwsException() {
String group = "regularGroup";
+ when(ctx.isLiteConsumer()).thenReturn(true);
when(groupConfig.getLiteBindTopic()).thenReturn("");
when(messagingProcessor.getSubscriptionGroupConfig(ctx,
group)).thenReturn(groupConfig);
GrpcProxyException exception = assertThrows(GrpcProxyException.class,
() -> {
- clientProcessor.validateLiteMode(ctx, group,
MessageModel.LITE_SELECTIVE);
+ clientProcessor.validateLiteConsumer(ctx, group,
MessageModel.LITE_SELECTIVE, Collections.emptySet());
});
assertEquals("regular group cannot use LITE mode: " + group,
exception.getMessage());
}
@Test
- public void
testValidateLiteMode_liteGroupWithoutLiteMode_throwsException() {
+ public void
testValidateLiteConsumer_liteGroupWithRegularConsumer_throwsException() {
String group = "liteGroup";
+ when(ctx.isLiteConsumer()).thenReturn(false);
when(groupConfig.getLiteBindTopic()).thenReturn("topic1");
when(messagingProcessor.getSubscriptionGroupConfig(ctx,
group)).thenReturn(groupConfig);
GrpcProxyException exception = assertThrows(GrpcProxyException.class,
() -> {
- clientProcessor.validateLiteMode(ctx, group,
MessageModel.CLUSTERING);
+ clientProcessor.validateLiteConsumer(ctx, group,
MessageModel.CLUSTERING, Collections.emptySet());
});
assertEquals("lite group must use LITE mode: " + group,
exception.getMessage());
}
@Test
- public void testValidateLiteMode_regularGroupWithoutLiteMode_noException()
{
+ public void
testValidateLiteConsumer_regularGroupWithRegularConsumer_returnsSameModel() {
String group = "regularGroup";
+ when(ctx.isLiteConsumer()).thenReturn(false);
when(groupConfig.getLiteBindTopic()).thenReturn("");
when(messagingProcessor.getSubscriptionGroupConfig(ctx,
group)).thenReturn(groupConfig);
- assertDoesNotThrow(() -> {
- clientProcessor.validateLiteMode(ctx, group,
MessageModel.CLUSTERING);
- });
+ MessageModel result = clientProcessor.validateLiteConsumer(ctx, group,
MessageModel.CLUSTERING, Collections.emptySet());
+ assertEquals(MessageModel.CLUSTERING, result);
}
@Test
- public void testValidateLiteMode_liteGroupWithLiteMode_noException() {
+ public void
testValidateLiteConsumer_liteGroupWithLiteConsumer_keepsLiteSelective() {
String group = "liteGroup";
+ when(ctx.isLiteConsumer()).thenReturn(true);
when(groupConfig.getLiteBindTopic()).thenReturn("topic1");
+ when(groupConfig.isWildcardLiteGroup()).thenReturn(false);
when(messagingProcessor.getSubscriptionGroupConfig(ctx,
group)).thenReturn(groupConfig);
- assertDoesNotThrow(() -> {
- clientProcessor.validateLiteMode(ctx, group,
MessageModel.LITE_SELECTIVE);
+ MessageModel result = clientProcessor.validateLiteConsumer(ctx, group,
MessageModel.LITE_SELECTIVE, Collections.emptySet());
+ assertEquals(MessageModel.LITE_SELECTIVE, result);
+ }
+
+ @Test
+ public void
testValidateLiteConsumer_wildcardLiteGroup_normalizesToClustering() {
+ String group = "wildcardLiteGroup";
+ when(ctx.isLiteConsumer()).thenReturn(true);
+ when(groupConfig.getLiteBindTopic()).thenReturn("topic1");
+ when(groupConfig.isWildcardLiteGroup()).thenReturn(true);
+ when(messagingProcessor.getSubscriptionGroupConfig(ctx,
group)).thenReturn(groupConfig);
+
+ MessageModel result = clientProcessor.validateLiteConsumer(ctx, group,
MessageModel.LITE_SELECTIVE, Collections.emptySet());
+ assertEquals(MessageModel.CLUSTERING, result);
+ }
+
+ @Test
+ public void
testValidateLiteConsumer_liteGroupWithMismatchedSubTopic_throwsException() {
+ String group = "liteGroup";
+ String bindTopic = "topic1";
+ SubscriptionData subscriptionData = new SubscriptionData();
+ subscriptionData.setTopic("otherTopic");
+ Set<SubscriptionData> subList = new HashSet<>();
+ subList.add(subscriptionData);
+
+ when(ctx.isLiteConsumer()).thenReturn(true);
+ when(groupConfig.getLiteBindTopic()).thenReturn(bindTopic);
+ when(messagingProcessor.getSubscriptionGroupConfig(ctx,
group)).thenReturn(groupConfig);
+
+ GrpcProxyException exception = assertThrows(GrpcProxyException.class,
() -> {
+ clientProcessor.validateLiteConsumer(ctx, group,
MessageModel.LITE_SELECTIVE, subList);
});
+
+ assertTrue(exception.getMessage().contains("expected to bind topic"));
}
@Test
