[ISSUE #11087] Validate lite.bind.topic for LiteTopic groups (#11088)

* [ISSUE #11087] Validate lite.bind.topic for LiteTopic groups

* Update

broker/src/main/java/org/apache/rocketmq/broker/lite/LiteMetadataUtil.java

@@ -22,6 +22,7 @@ import java.util.Map;
 import java.util.Set;
 import java.util.concurrent.ConcurrentMap;
 import java.util.stream.Collectors;
+import org.apache.commons.lang3.StringUtils;
 import org.apache.rocketmq.broker.BrokerController;
 import org.apache.rocketmq.common.TopicConfig;
 import org.apache.rocketmq.common.attribute.TopicMessageType;
@@ -52,7 +53,7 @@ public class LiteMetadataUtil {
         }
         SubscriptionGroupConfig groupConfig =
             brokerController.getSubscriptionGroupManager().findSubscriptionGroupConfig(group);
-        return null != groupConfig && groupConfig.getLiteBindTopic() != null;
+        return null != groupConfig && StringUtils.isNotBlank(groupConfig.getLiteBindTopic());
     }
 
     public static String getLiteBindTopic(String group, BrokerController brokerController) {
@@ -135,7 +136,7 @@ public class LiteMetadataUtil {
             brokerController.getSubscriptionGroupManager().getSubscriptionGroupTable();
 
         return groupTable.entrySet().stream()
-            .filter(entry -> entry.getValue().getLiteBindTopic() != null)
+            .filter(entry -> StringUtils.isNotBlank(entry.getValue().getLiteBindTopic()))
             .collect(Collectors.groupingBy(
                 entry -> entry.getValue().getLiteBindTopic(),
                 Collectors.mapping(Map.Entry::getKey, Collectors.toSet())

broker/src/test/java/org/apache/rocketmq/broker/lite/LiteMetadataUtilTest.java

@@ -0,0 +1,112 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.rocketmq.broker.lite;
+
+import java.util.Collections;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
+import org.apache.rocketmq.broker.BrokerController;
+import org.apache.rocketmq.broker.subscription.SubscriptionGroupManager;
+import org.apache.rocketmq.remoting.protocol.subscription.SubscriptionGroupConfig;
+import org.junit.Before;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.mockito.Mock;
+import org.mockito.junit.MockitoJUnitRunner;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
+import static org.mockito.Mockito.when;
+
+@RunWith(MockitoJUnitRunner.class)
+public class LiteMetadataUtilTest {
+
+    @Mock
+    private BrokerController brokerController;
+
+    @Mock
+    private SubscriptionGroupManager subscriptionGroupManager;
+
+    @Before
+    public void setUp() {
+        when(brokerController.getSubscriptionGroupManager()).thenReturn(subscriptionGroupManager);
+    }
+
+    @Test
+    public void testIsLiteGroupTypeTreatsEmptyBindTopicAsNonLite() {
+        SubscriptionGroupConfig emptyBindGroup = new SubscriptionGroupConfig();
+        emptyBindGroup.setGroupName("emptyBindGroup");
+        emptyBindGroup.setLiteBindTopic("");
+
+        SubscriptionGroupConfig blankBindGroup = new SubscriptionGroupConfig();
+        blankBindGroup.setGroupName("blankBindGroup");
+        blankBindGroup.setLiteBindTopic(" ");
+
+        SubscriptionGroupConfig liteGroup = new SubscriptionGroupConfig();
+        liteGroup.setGroupName("liteGroup");
+        liteGroup.setLiteBindTopic("parentTopic");
+
+        when(subscriptionGroupManager.findSubscriptionGroupConfig("normalGroup"))
+            .thenReturn(new SubscriptionGroupConfig());
+        when(subscriptionGroupManager.findSubscriptionGroupConfig("emptyBindGroup"))
+            .thenReturn(emptyBindGroup);
+        when(subscriptionGroupManager.findSubscriptionGroupConfig("blankBindGroup"))
+            .thenReturn(blankBindGroup);
+        when(subscriptionGroupManager.findSubscriptionGroupConfig("liteGroup"))
+            .thenReturn(liteGroup);
+
+        assertFalse(LiteMetadataUtil.isLiteGroupType("missingGroup", brokerController));
+        assertFalse(LiteMetadataUtil.isLiteGroupType("normalGroup", brokerController));
+        assertFalse(LiteMetadataUtil.isLiteGroupType("emptyBindGroup", brokerController));
+        assertFalse(LiteMetadataUtil.isLiteGroupType("blankBindGroup", brokerController));
+        assertTrue(LiteMetadataUtil.isLiteGroupType("liteGroup", brokerController));
+    }
+
+    @Test
+    public void testGetSubscriberGroupMapSkipsEmptyBindTopic() {
+        ConcurrentMap<String, SubscriptionGroupConfig> groupTable = new ConcurrentHashMap<>();
+        groupTable.put("normalGroup", new SubscriptionGroupConfig());
+
+        SubscriptionGroupConfig emptyBindGroup = new SubscriptionGroupConfig();
+        emptyBindGroup.setGroupName("emptyBindGroup");
+        emptyBindGroup.setLiteBindTopic("");
+        groupTable.put("emptyBindGroup", emptyBindGroup);
+
+        SubscriptionGroupConfig blankBindGroup = new SubscriptionGroupConfig();
+        blankBindGroup.setGroupName("blankBindGroup");
+        blankBindGroup.setLiteBindTopic(" ");
+        groupTable.put("blankBindGroup", blankBindGroup);
+
+        SubscriptionGroupConfig liteGroup = new SubscriptionGroupConfig();
+        liteGroup.setGroupName("liteGroup");
+        liteGroup.setLiteBindTopic("parentTopic");
+        groupTable.put("liteGroup", liteGroup);
+
+        when(subscriptionGroupManager.getSubscriptionGroupTable()).thenReturn(groupTable);
+
+        Map<String, Set<String>> result = LiteMetadataUtil.getSubscriberGroupMap(brokerController);
+
+        assertFalse(result.containsKey(""));
+        assertFalse(result.containsKey(" "));
+        assertFalse(result.containsKey(null));
+        assertEquals(Collections.singleton("liteGroup"), result.get("parentTopic"));
+    }
+}

broker/src/test/java/org/apache/rocketmq/broker/processor/AdminBrokerProcessorTest.java

@@ -86,6 +86,7 @@ import org.apache.rocketmq.remoting.protocol.body.GroupList;
 import org.apache.rocketmq.remoting.protocol.body.HARuntimeInfo;
 import org.apache.rocketmq.remoting.protocol.body.LockBatchRequestBody;
 import org.apache.rocketmq.remoting.protocol.body.QueryCorrectionOffsetBody;
+import org.apache.rocketmq.remoting.protocol.body.SubscriptionGroupList;
 import org.apache.rocketmq.remoting.protocol.body.SubscriptionGroupWrapper;
 import org.apache.rocketmq.remoting.protocol.body.TopicConfigSerializeWrapper;
 import org.apache.rocketmq.remoting.protocol.body.UnlockBatchRequestBody;
@@ -146,6 +147,7 @@ import org.apache.rocketmq.store.timer.TimerMessageStore;
 import org.apache.rocketmq.store.timer.TimerMetrics;
 import org.apache.rocketmq.store.util.LibC;
 import org.junit.After;
+import org.junit.Assert;
 import org.junit.Before;
 import org.junit.Test;
 import org.junit.runner.RunWith;
@@ -659,6 +661,41 @@ public class AdminBrokerProcessorTest {
         assertThat(response.getCode()).isEqualTo(ResponseCode.SUCCESS);
     }
 
+    @Test
+    public void testDeleteSubscriptionGroupWithEmptyLiteBindTopicDoesNotCleanOffset() throws Exception {
+        String groupName = "GID-EMPTY-LITE-BIND";
+        SubscriptionGroupConfig groupConfig = new SubscriptionGroupConfig();
+        groupConfig.setGroupName(groupName);
+        groupConfig.setLiteBindTopic("");
+        brokerController.getSubscriptionGroupManager().getSubscriptionGroupTable().put(groupName, groupConfig);
+        brokerController.setConsumerOffsetManager(consumerOffsetManager);
+
+        RemotingCommand request = RemotingCommand.createRequestCommand(RequestCode.DELETE_SUBSCRIPTIONGROUP, null);
+        request.addExtField("groupName", groupName);
+        request.addExtField("cleanOffset", "false");
+        RemotingCommand response = adminBrokerProcessor.processRequest(handlerContext, request);
+
+        assertThat(response.getCode()).isEqualTo(ResponseCode.SUCCESS);
+        verify(consumerOffsetManager, never()).removeOffset(groupName);
+    }
+
+    @Test
+    public void testDeleteSubscriptionGroupListWithEmptyLiteBindTopicDoesNotCleanOffset() throws Exception {
+        brokerController.getBrokerConfig().setBatchDeleteSubscriptionGroupMaxRate(0);
+        String groupName = "GID-BATCH-EMPTY-LITE-BIND";
+        SubscriptionGroupConfig groupConfig = new SubscriptionGroupConfig();
+        groupConfig.setGroupName(groupName);
+        groupConfig.setLiteBindTopic("");
+        brokerController.getSubscriptionGroupManager().getSubscriptionGroupTable().put(groupName, groupConfig);
+        brokerController.setConsumerOffsetManager(consumerOffsetManager);
+
+        RemotingCommand request = buildDeleteSubscriptionGroupListRequest(Collections.singletonList(groupName), false);
+        RemotingCommand response = adminBrokerProcessor.processRequest(handlerContext, request);
+
+        assertThat(response.getCode()).isEqualTo(ResponseCode.SUCCESS);
+        verify(consumerOffsetManager, never()).removeOffset(groupName);
+    }
+
     @Test
     public void testDeleteTopicListWithPopRetryTopics() throws Exception {
         // When clearRetryTopicWhenDeleteTopic=true, POP retry topics should be collected and deleted
@@ -1077,6 +1114,42 @@ public class AdminBrokerProcessorTest {
         assertThat(response.getCode()).isEqualTo(ResponseCode.SUCCESS);
     }
 
+    @Test
+    public void testUpdateAndCreateSubscriptionGroupRejectsEmptyLiteBindTopic() {
+        String groupName = "GID-EMPTY-LITE-BIND";
+        SubscriptionGroupConfig subscriptionGroupConfig = new SubscriptionGroupConfig();
+        subscriptionGroupConfig.setGroupName(groupName);
+        subscriptionGroupConfig.setAttributes(ImmutableMap.of("+lite.bind.topic", ""));
+
+        RemotingCommand request = RemotingCommand.createRequestCommand(RequestCode.UPDATE_AND_CREATE_SUBSCRIPTIONGROUP, null);
+        request.setBody(JSON.toJSON(subscriptionGroupConfig).toString().getBytes(StandardCharsets.UTF_8));
+
+        RuntimeException exception = Assert.assertThrows(RuntimeException.class,
+            () -> adminBrokerProcessor.processRequest(handlerContext, request));
+
+        assertThat(exception).hasMessageContaining("The specified topic is blank");
+        assertFalse(brokerController.getSubscriptionGroupManager().getSubscriptionGroupTable().containsKey(groupName));
+    }
+
+    @Test
+    public void testUpdateAndCreateSubscriptionGroupListRejectsEmptyLiteBindTopic() {
+        String groupName = "GID-LIST-EMPTY-LITE-BIND";
+        SubscriptionGroupConfig subscriptionGroupConfig = new SubscriptionGroupConfig();
+        subscriptionGroupConfig.setGroupName(groupName);
+        subscriptionGroupConfig.setAttributes(ImmutableMap.of("+lite.bind.topic", ""));
+
+        SubscriptionGroupList subscriptionGroupList =
+            new SubscriptionGroupList(Collections.singletonList(subscriptionGroupConfig));
+        RemotingCommand request = RemotingCommand.createRequestCommand(RequestCode.UPDATE_AND_CREATE_SUBSCRIPTIONGROUP_LIST, null);
+        request.setBody(subscriptionGroupList.encode());
+
+        RuntimeException exception = Assert.assertThrows(RuntimeException.class,
+            () -> adminBrokerProcessor.processRequest(handlerContext, request));
+
+        assertThat(exception).hasMessageContaining("The specified topic is blank");
+        assertFalse(brokerController.getSubscriptionGroupManager().getSubscriptionGroupTable().containsKey(groupName));
+    }
+
     @Test
     public void testGetAllSubscriptionGroupInRocksdb() throws Exception {
         initRocksdbSubscriptionManager();

common/src/main/java/org/apache/rocketmq/common/SubscriptionGroupAttributes.java

@@ -23,9 +23,10 @@ import java.util.Map;
 import org.apache.rocketmq.common.attribute.Attribute;
 import org.apache.rocketmq.common.attribute.BooleanAttribute;
 import org.apache.rocketmq.common.attribute.EnumAttribute;
+import org.apache.rocketmq.common.attribute.LiteSubModel;
 import org.apache.rocketmq.common.attribute.LongRangeAttribute;
 import org.apache.rocketmq.common.attribute.StringAttribute;
-import org.apache.rocketmq.common.attribute.LiteSubModel;
+import org.apache.rocketmq.common.topic.TopicValidator;
 
 public class SubscriptionGroupAttributes {
 
@@ -40,7 +41,13 @@ public class SubscriptionGroupAttributes {
 
     public static final StringAttribute LITE_BIND_TOPIC_ATTRIBUTE = new StringAttribute(
         "lite.bind.topic",
-        true
+        true,
+        value -> {
+            TopicValidator.ValidateResult result = TopicValidator.validateTopic(value);
+            if (!result.isValid()) {
+                throw new RuntimeException(result.getRemark());
+            }
+        }
     );
 
     public static final EnumAttribute LITE_SUB_MODEL_ATTRIBUTE = new EnumAttribute(
@@ -97,4 +104,4 @@ public class SubscriptionGroupAttributes {
         ALL.put(LITE_SUB_CLIENT_MAX_EVENT_COUNT_ATTRIBUTE.getName(), LITE_SUB_CLIENT_MAX_EVENT_COUNT_ATTRIBUTE);
         ALL.put(LITE_SUB_WILDCARD_ATTRIBUTE.getName(), LITE_SUB_WILDCARD_ATTRIBUTE);
     }
-}
+}

common/src/main/java/org/apache/rocketmq/common/attribute/StringAttribute.java

@@ -17,16 +17,23 @@
 
 package org.apache.rocketmq.common.attribute;
 
-import static com.google.common.base.Preconditions.checkNotNull;
+import com.google.common.base.Preconditions;
+import java.util.function.Consumer;
 
 public class StringAttribute extends Attribute {
+    private final Consumer<String> validator;
 
     public StringAttribute(String name, boolean changeable) {
+        this(name, changeable, Preconditions::checkNotNull);
+    }
+
+    public StringAttribute(String name, boolean changeable, Consumer<String> validator) {
         super(name, changeable);
+        this.validator = validator;
     }
 
     @Override
     public void verify(String value) {
-        checkNotNull(value);
+        validator.accept(value);
     }
 }

common/src/test/java/org/apache/rocketmq/common/SubscriptionGroupAttributesTest.java

@@ -0,0 +1,43 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.rocketmq.common;
+
+import org.junit.Assert;
+import org.junit.Test;
+
+public class SubscriptionGroupAttributesTest {
+
+    @Test
+    public void testLiteBindTopicAttributeValidatesTopicName() {
+        SubscriptionGroupAttributes.LITE_BIND_TOPIC_ATTRIBUTE.verify("parentTopic");
+        SubscriptionGroupAttributes.LITE_BIND_TOPIC_ATTRIBUTE.verify("parent_topic");
+
+        Assert.assertThrows(RuntimeException.class,
+            () -> SubscriptionGroupAttributes.LITE_BIND_TOPIC_ATTRIBUTE.verify(null));
+        Assert.assertThrows(RuntimeException.class,
+            () -> SubscriptionGroupAttributes.LITE_BIND_TOPIC_ATTRIBUTE.verify(""));
+        Assert.assertThrows(RuntimeException.class,
+            () -> SubscriptionGroupAttributes.LITE_BIND_TOPIC_ATTRIBUTE.verify(" "));
+        Assert.assertThrows(RuntimeException.class,
+            () -> SubscriptionGroupAttributes.LITE_BIND_TOPIC_ATTRIBUTE.verify("parent topic"));
+    }
+
+    @Test
+    public void testLiteSubWildcardAttributeStillAllowsEmptyValue() {
+        SubscriptionGroupAttributes.LITE_SUB_WILDCARD_ATTRIBUTE.verify("");
+    }
+}