>
- * mapping each group to its subscribers
+ * This method merges results from two sources:
+ * 1. Exact subscriptions from liteTopic2ClientGroup
+ * 2. Wildcard subscriptions from wildcardGroupMap
+ *
+ * When group is specified, returns a single-entry map for that group.
+ * When group is null, returns a map of all groups to their subscribers.
*/
@Override
- public SubscriberWrapper getAllSubscriber(String group, String lmqName) {
- String topic = LiteUtil.getParentTopic(lmqName);
-
- if (group != null) {
- if (LiteMetadataUtil.isWildcardGroup(group, brokerController)) {
- return getWildcardSubscriber(group, topic);
- }
- SubscriberWrapper.ListWrapper wrapper = new SubscriberWrapper.ListWrapper();
- Set subscribers = liteTopic2Group.get(lmqName);
- if (subscribers != null) {
- wrapper.getClients().addAll(subscribers.stream()
- .filter(clientGroup -> group.equals(clientGroup.group))
- .collect(Collectors.toSet()));
- }
- return wrapper;
+ public Map> getAllSubscribers(String group, String lmqName) {
+ Map> result = new HashMap<>();
+ if (LiteMetadataUtil.isWildcardGroup(group, brokerController)) {
+ collectWildcardSubscribers(result, group, lmqName);
} else {
- SubscriberWrapper.MapWrapper wrapper = new SubscriberWrapper.MapWrapper();
- Set subscribers = liteTopic2Group.get(lmqName);
- if (subscribers != null) {
- for (ClientGroup clientGroup : subscribers) {
- wrapper.getGroupMap().computeIfAbsent(clientGroup.group, k -> new ArrayList<>()).add(clientGroup);
- }
+ collectExactSubscribers(result, group, lmqName);
+ if (group == null) {
+ collectWildcardSubscribers(result, null, lmqName);
}
- Set wildcardGroups = wildcardGroupMap.get(topic);
- if (wildcardGroups != null) {
- for (String wildcardGroup : wildcardGroups) {
- List wildcardClients = getWildcardGroupClients(topic, wildcardGroup);
- if (CollectionUtils.isNotEmpty(wildcardClients)) {
- wrapper.getGroupMap().putIfAbsent(wildcardGroup, wildcardClients);
- }
+ }
+ return result;
+ }
+
+ @Override
+ public List getWildcardGroupClients(String group) {
+ List list = null;
+ try {
+ list = wildcardClientCache.get(group, () -> {
+ String topic = LiteMetadataUtil.getLiteBindTopic(group, brokerController);
+ if (topic == null) {
+ return Collections.emptyList();
}
+ Set clientSet = liteTopic2ClientGroup.get(mockLmqNameForWildcardGroup(topic, group));
+ return clientSet != null ? new ArrayList<>(clientSet) : Collections.emptyList();
+ });
+ } catch (ExecutionException ignored) {
+ }
+ return list;
+ }
+
+ private void collectExactSubscribers(Map> result, String group, String lmqName) {
+ Set subscribers = liteTopic2ClientGroup.get(lmqName);
+ if (subscribers == null) {
+ return;
+ }
+ for (ClientGroup cg : subscribers) {
+ if (group == null || group.equals(cg.group)) {
+ result.computeIfAbsent(cg.group, k -> new ArrayList<>()).add(cg);
}
- return wrapper;
}
}
- @Override
- public SubscriberWrapper.ListWrapper getWildcardSubscriber(String group, String topic) {
- return new SubscriberWrapper.ListWrapper(getWildcardGroupClients(topic, group));
+ private void collectWildcardSubscribers(Map> result, String group, String lmqName) {
+ if (group == null) {
+ String topic = LiteUtil.getParentTopic(lmqName);
+ Set wildcardGroups = wildcardGroupMap.get(topic);
+ if (wildcardGroups == null) {
+ return;
+ }
+ for (String wildcardGroup : wildcardGroups) {
+ List wildcardClients = getWildcardGroupClients(wildcardGroup);
+ if (CollectionUtils.isNotEmpty(wildcardClients)) {
+ result.putIfAbsent(wildcardGroup, wildcardClients);
+ }
+ }
+ } else if (LiteMetadataUtil.isWildcardGroup(group, brokerController)) {
+ List wildcardClients = getWildcardGroupClients(group);
+ if (CollectionUtils.isNotEmpty(wildcardClients)) {
+ result.put(group, wildcardClients);
+ }
+ }
}
/**
@@ -251,7 +272,7 @@ public SubscriberWrapper.ListWrapper getWildcardSubscriber(String group, String
*/
@Override
public void cleanSubscription(String lmqName, boolean notifyClient) {
- Set topicGroupSet = liteTopic2Group.remove(lmqName);
+ Set topicGroupSet = liteTopic2ClientGroup.remove(lmqName);
if (CollectionUtils.isEmpty(topicGroupSet)) {
return;
}
@@ -260,7 +281,7 @@ public void cleanSubscription(String lmqName, boolean notifyClient) {
if (liteSubscription == null) {
continue;
}
- if (liteSubscription.removeLiteTopic(lmqName)) {
+ if (liteSubscription.removeLmq(lmqName)) {
if (notifyClient) {
notifyUnsubscribeLite(topicGroup.clientId, topicGroup.group, lmqName);
}
@@ -269,20 +290,20 @@ public void cleanSubscription(String lmqName, boolean notifyClient) {
}
}
- protected void addTopicGroup(ClientGroup clientGroup, String lmqName) {
- Set topicGroupSet = liteTopic2Group
- .computeIfAbsent(lmqName, k -> ConcurrentHashMap.newKeySet());
- if (topicGroupSet.add(clientGroup)) {
- activeNum.incrementAndGet();
- invalidateWildcardCacheIfNecessary(clientGroup.group);
- for (LiteCtlListener listener : listeners) {
- listener.onRegister(clientGroup.clientId, clientGroup.group, lmqName);
- }
+ protected void addClientGroup(LiteSubscription subscription, ClientGroup clientGroup, String lmqName) {
+ if (!subscription.addLmq(lmqName)) {
+ return;
+ }
+ liteTopic2ClientGroup.computeIfAbsent(lmqName, k -> ConcurrentHashMap.newKeySet()).add(clientGroup);
+ activeNum.incrementAndGet();
+ invalidateWildcardCacheIfNecessary(clientGroup.group);
+ for (LiteCtlListener listener : listeners) {
+ listener.onRegister(clientGroup.clientId, clientGroup.group, lmqName);
}
}
- protected void removeTopicGroup(ClientGroup clientGroup, String lmqName, boolean resetOffset) {
- Set topicGroupSet = liteTopic2Group.get(lmqName);
+ protected void removeClientGroup(ClientGroup clientGroup, String lmqName, boolean resetOffset) {
+ Set topicGroupSet = liteTopic2ClientGroup.get(lmqName);
if (topicGroupSet == null) {
return;
}
@@ -298,7 +319,7 @@ protected void removeTopicGroup(ClientGroup clientGroup, String lmqName, boolean
}
}
if (topicGroupSet.isEmpty()) {
- liteTopic2Group.remove(lmqName);
+ liteTopic2ClientGroup.remove(lmqName);
unmarkWildcardGroupIfNecessary(lmqName);
}
}
@@ -307,7 +328,7 @@ protected void removeTopicGroup(ClientGroup clientGroup, String lmqName, boolean
* Remove clients that subscribe to the same liteTopic under the same group
*/
protected void excludeClientByLmqName(String newClientId, String group, String lmqName) {
- Set clientSet = liteTopic2Group.get(lmqName);
+ Set clientSet = liteTopic2ClientGroup.get(lmqName);
if (CollectionUtils.isEmpty(clientSet)) {
return;
}
@@ -318,9 +339,9 @@ protected void excludeClientByLmqName(String newClientId, String group, String l
toRemove.forEach(clientGroup -> {
LiteSubscription liteSubscription = client2Subscription.get(clientGroup.clientId);
if (liteSubscription != null) {
- liteSubscription.removeLiteTopic(lmqName);
+ liteSubscription.removeLmq(lmqName);
// remove client if no more liteTopic
- if (liteSubscription.getLiteTopicSet().isEmpty()) {
+ if (liteSubscription.getLmqSet().isEmpty()) {
client2Subscription.remove(clientGroup.clientId);
}
}
@@ -329,7 +350,7 @@ protected void excludeClientByLmqName(String newClientId, String group, String l
boolean resetOffset = LiteMetadataUtil.isResetOffsetInExclusiveMode(group, brokerController);
LOGGER.info("excludeClientByLmqName group:{}, lmqName:{}, resetOffset:{}, clientId:{} -> {}",
group, lmqName, resetOffset, clientGroup.clientId, newClientId);
- removeTopicGroup(clientGroup, lmqName, resetOffset);
+ removeClientGroup(clientGroup, lmqName, resetOffset);
});
}
@@ -441,18 +462,6 @@ private String mockLmqNameForWildcardGroup(String topic, String group) {
return topic + "@" + group;
}
- private List getWildcardGroupClients(String topic, String group) {
- List list = null;
- try {
- list = wildcardClientCache.get(group, () -> {
- Set clientSet = liteTopic2Group.get(mockLmqNameForWildcardGroup(topic, group));
- return clientSet != null ? new ArrayList<>(clientSet) : Collections.emptyList();
- });
- } catch (ExecutionException ignored) {
- }
- return list;
- }
-
@Override
public void run() {
LOGGER.info("Start checking lite subscription.");
diff --git a/broker/src/main/java/org/apache/rocketmq/broker/lite/LmqPrefixIndex.java b/broker/src/main/java/org/apache/rocketmq/broker/lite/LmqPrefixIndex.java
new file mode 100644
index 00000000000..f073bb9e590
--- /dev/null
+++ b/broker/src/main/java/org/apache/rocketmq/broker/lite/LmqPrefixIndex.java
@@ -0,0 +1,109 @@
+/*
+ * 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.SortedMap;
+import java.util.concurrent.locks.ReadWriteLock;
+import java.util.concurrent.locks.ReentrantReadWriteLock;
+import java.util.function.Function;
+
+import org.apache.commons.collections4.trie.PatriciaTrie;
+import org.apache.commons.lang3.StringUtils;
+import org.apache.rocketmq.common.lite.LiteUtil;
+
+/**
+ * Global prefix index over lmqName, backed by {@link PatriciaTrie}.
+ *
+ * A single instance is shared across all parentTopics: every lmqName starts with
+ * {@link LiteUtil#LITE_TOPIC_PREFIX} followed by its parentTopic, so lmqs of the same
+ * parentTopic form a contiguous subtree, and a prefix lookup only walks that subtree.
+ *
+ *
Used to accelerate prefix-subscription full dispatch in
+ * {@link LiteEventDispatcher#doFullDispatchForClient(String, String)}.
+ *
+ *
A {@link ReadWriteLock} guards the trie; reads dominate writes by ~10000x in steady state.
+ * Empty prefix / parentTopic is rejected to avoid an unintended full-table scan.
+ */
+public class LmqPrefixIndex {
+
+ private final PatriciaTrie trie = new PatriciaTrie<>();
+ private final ReadWriteLock rwLock = new ReentrantReadWriteLock();
+
+ /**
+ * Insert lmqName into the trie. Idempotent. Returns {@code true} if newly added.
+ */
+ public boolean add(String lmqName) {
+ if (lmqName == null) {
+ return false;
+ }
+ rwLock.writeLock().lock();
+ try {
+ return trie.put(lmqName, Boolean.TRUE) == null;
+ } finally {
+ rwLock.writeLock().unlock();
+ }
+ }
+
+ /**
+ * Remove lmqName from the trie. Returns {@code true} if an entry was removed.
+ */
+ public boolean remove(String lmqName) {
+ rwLock.writeLock().lock();
+ try {
+ return trie.remove(lmqName) != null;
+ } finally {
+ rwLock.writeLock().unlock();
+ }
+ }
+
+ /**
+ * Iterate all lmqs whose name starts with the given lmqName prefix.
+ * The visitor returns {@code false} to break iteration early.
+ * Empty prefix is rejected to avoid a full scan.
+ *
+ * @return {@code true} if iteration completed; {@code false} on early break or invalid input.
+ */
+ public boolean forEachLmqByPrefix(String lmqPrefix, Function visitor) {
+ if (StringUtils.isEmpty(lmqPrefix) || visitor == null) {
+ return false;
+ }
+ rwLock.readLock().lock();
+ try {
+ SortedMap sub = trie.prefixMap(lmqPrefix);
+ for (String lmqName : sub.keySet()) {
+ if (!visitor.apply(lmqName)) {
+ return false;
+ }
+ }
+ } finally {
+ rwLock.readLock().unlock();
+ }
+ return true;
+ }
+
+ /**
+ * Best-effort size / emptiness probes for monitoring; intentionally lock-free.
+ */
+ public boolean isEmpty() {
+ return trie.isEmpty();
+ }
+
+ public int size() {
+ return trie.size();
+ }
+}
diff --git a/broker/src/main/java/org/apache/rocketmq/broker/lite/RocksDBLiteLifecycleManager.java b/broker/src/main/java/org/apache/rocketmq/broker/lite/RocksDBLiteLifecycleManager.java
index a0adb7216cf..2fa968c56d0 100644
--- a/broker/src/main/java/org/apache/rocketmq/broker/lite/RocksDBLiteLifecycleManager.java
+++ b/broker/src/main/java/org/apache/rocketmq/broker/lite/RocksDBLiteLifecycleManager.java
@@ -17,11 +17,9 @@
package org.apache.rocketmq.broker.lite;
-import org.apache.commons.lang3.StringUtils;
import org.apache.commons.lang3.reflect.FieldUtils;
import org.apache.commons.lang3.tuple.Triple;
import org.apache.rocketmq.broker.BrokerController;
-import org.apache.rocketmq.common.Pair;
import org.apache.rocketmq.common.constant.LoggerName;
import org.apache.rocketmq.common.lite.LiteUtil;
import org.apache.rocketmq.logging.org.slf4j.Logger;
@@ -32,10 +30,7 @@
import org.apache.rocketmq.store.queue.RocksDBConsumeQueueStore;
import org.apache.rocketmq.tieredstore.TieredMessageStore;
-import java.util.ArrayList;
import java.util.Collections;
-import java.util.Iterator;
-import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentMap;
import java.util.function.Function;
@@ -54,43 +49,6 @@ public long getMaxOffsetInQueue(String lmqName) {
return maxCqOffsetTable.getOrDefault(lmqName + "-0", -1L) + 1;
}
- @Override
- public List collectByParentTopic(String parentTopic) {
- if (StringUtils.isEmpty(parentTopic)) {
- return Collections.emptyList();
- }
- List resultList = new ArrayList<>();
- Iterator> iterator = maxCqOffsetTable.entrySet().iterator();
- while (iterator.hasNext()) {
- Map.Entry entry = iterator.next();
- String queueAndQid = entry.getKey();
- String lmqName = queueAndQid.substring(0, queueAndQid.lastIndexOf("-"));
- if (LiteUtil.belongsTo(lmqName, parentTopic)) {
- resultList.add(lmqName);
- }
- }
- return resultList;
- }
-
- @Override
- public List> collectExpiredLiteTopic() {
- List> lmqToDelete = new ArrayList<>();
- Iterator> iterator = maxCqOffsetTable.entrySet().iterator();
- while (iterator.hasNext()) {
- Map.Entry entry = iterator.next();
- String queueAndQid = entry.getKey();
- String lmqName = queueAndQid.substring(0, queueAndQid.lastIndexOf("-"));
- String parentTopic = LiteUtil.getParentTopic(lmqName);
- if (null == parentTopic) {
- continue;
- }
- if (isLiteTopicExpired(parentTopic, lmqName, entry.getValue() + 1)) {
- lmqToDelete.add(new Pair<>(parentTopic, lmqName));
- }
- }
- return lmqToDelete;
- }
-
@Override
public boolean init() {
super.init();
diff --git a/broker/src/main/java/org/apache/rocketmq/broker/lite/SubscriberWrapper.java b/broker/src/main/java/org/apache/rocketmq/broker/lite/SubscriberWrapper.java
deleted file mode 100644
index 97c02e52825..00000000000
--- a/broker/src/main/java/org/apache/rocketmq/broker/lite/SubscriberWrapper.java
+++ /dev/null
@@ -1,64 +0,0 @@
-/*
- * 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 org.apache.rocketmq.common.entity.ClientGroup;
-
-import java.util.ArrayList;
-import java.util.HashMap;
-import java.util.List;
-import java.util.Map;
-
-public abstract class SubscriberWrapper {
-
- public static class ListWrapper extends SubscriberWrapper {
- private final List clients;
-
- public ListWrapper() {
- this.clients = new ArrayList<>();
- }
-
- public ListWrapper(List clients) {
- this.clients = clients;
- }
-
- public List getClients() {
- return this.clients;
- }
- }
-
- public static class MapWrapper extends SubscriberWrapper {
- private final Map> groupMap = new HashMap<>();
-
- public MapWrapper() {
- }
-
- public Map> getGroupMap() {
- return groupMap;
- }
- }
-
- public ListWrapper asListWrapper() {
- return this instanceof ListWrapper ? (ListWrapper) this : null;
- }
-
- public MapWrapper asMapWrapper() {
- return this instanceof MapWrapper ? (MapWrapper) this : null;
- }
-
-}
diff --git a/broker/src/main/java/org/apache/rocketmq/broker/processor/LiteManagerProcessor.java b/broker/src/main/java/org/apache/rocketmq/broker/processor/LiteManagerProcessor.java
index d1b2a35b20b..57f3be55d96 100644
--- a/broker/src/main/java/org/apache/rocketmq/broker/processor/LiteManagerProcessor.java
+++ b/broker/src/main/java/org/apache/rocketmq/broker/processor/LiteManagerProcessor.java
@@ -20,15 +20,14 @@
import com.google.common.annotations.VisibleForTesting;
import io.netty.channel.ChannelHandlerContext;
-import java.util.Collections;
import java.util.List;
+import java.util.Map;
import org.apache.commons.lang3.StringUtils;
import org.apache.rocketmq.broker.BrokerController;
import org.apache.rocketmq.broker.lite.AbstractLiteLifecycleManager;
import org.apache.rocketmq.broker.lite.LiteMetadataUtil;
import org.apache.rocketmq.broker.lite.LiteSharding;
-import org.apache.rocketmq.broker.lite.SubscriberWrapper;
import org.apache.rocketmq.common.Pair;
import org.apache.rocketmq.common.TopicConfig;
import org.apache.rocketmq.common.attribute.TopicMessageType;
@@ -62,7 +61,6 @@
import org.apache.rocketmq.store.queue.ConsumeQueueStoreInterface;
import java.util.HashSet;
-import java.util.Map;
import java.util.Set;
import java.util.stream.Collectors;
import java.util.stream.Stream;
@@ -254,8 +252,8 @@ protected RemotingCommand getLiteClientInfo(ChannelHandlerContext ctx,
Set returnSet = null;
int liteTopicCount = 0;
LiteSubscription liteSubscription = brokerController.getLiteSubscriptionRegistry().getLiteSubscription(clientId);
- if (liteSubscription != null && liteSubscription.getLiteTopicSet() != null) {
- Set liteTopicSet = liteSubscription.getLiteTopicSet();
+ if (liteSubscription != null && liteSubscription.getLmqSet() != null) {
+ Set liteTopicSet = liteSubscription.getLmqSet();
liteTopicCount = liteTopicSet.size();
if (maxCount >= liteTopicCount) {
returnSet = liteTopicSet;
@@ -393,12 +391,9 @@ protected RemotingCommand triggerLiteDispatch(ChannelHandlerContext ctx, Remotin
@VisibleForTesting
public Set getSubscriber(String lmqName) {
- SubscriberWrapper.MapWrapper wrapper =
- brokerController.getLiteSubscriptionRegistry().getAllSubscriber(null, lmqName).asMapWrapper();
- if (null == wrapper) {
- return Collections.emptySet();
- }
- return wrapper.getGroupMap().entrySet().stream()
+ Map> subscriberMap =
+ brokerController.getLiteSubscriptionRegistry().getAllSubscribers(null, lmqName);
+ return subscriberMap.entrySet().stream()
.flatMap(entry -> {
String group = entry.getKey();
if (LiteMetadataUtil.isWildcardGroup(group, brokerController)) {
diff --git a/broker/src/test/java/org/apache/rocketmq/broker/lite/AbstractLiteLifecycleManagerTest.java b/broker/src/test/java/org/apache/rocketmq/broker/lite/AbstractLiteLifecycleManagerTest.java
index ddc140013c0..b3eb91c3734 100644
--- a/broker/src/test/java/org/apache/rocketmq/broker/lite/AbstractLiteLifecycleManagerTest.java
+++ b/broker/src/test/java/org/apache/rocketmq/broker/lite/AbstractLiteLifecycleManagerTest.java
@@ -17,10 +17,14 @@
package org.apache.rocketmq.broker.lite;
+import java.util.ArrayList;
+import java.util.Arrays;
import java.util.Collections;
+import java.util.HashSet;
import java.util.List;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
+import java.util.concurrent.TimeUnit;
import java.util.function.Function;
import org.apache.commons.lang3.tuple.Triple;
@@ -31,7 +35,6 @@
import org.apache.rocketmq.broker.subscription.SubscriptionGroupManager;
import org.apache.rocketmq.broker.topic.TopicConfigManager;
import org.apache.rocketmq.common.BrokerConfig;
-import org.apache.rocketmq.common.Pair;
import org.apache.rocketmq.common.TopicAttributes;
import org.apache.rocketmq.common.TopicConfig;
import org.apache.rocketmq.common.attribute.TopicMessageType;
@@ -44,6 +47,7 @@
import org.junit.Test;
import org.junit.runner.RunWith;
import org.mockito.Mock;
+import org.mockito.ArgumentCaptor;
import org.mockito.Mockito;
import org.mockito.junit.MockitoJUnitRunner;
@@ -89,6 +93,7 @@ public class AbstractLiteLifecycleManagerTest {
private final TopicConfig topicConfig = new TopicConfig(PARENT_TOPIC, 1, 1);
private final SubscriptionGroupConfig groupConfig = new SubscriptionGroupConfig();
private final ConcurrentMap> offsetTable = new ConcurrentHashMap<>();
+ private final ConcurrentMap topicConfigTable = new ConcurrentHashMap<>();
@Before
public void setUp() {
@@ -104,7 +109,7 @@ public void setUp() {
topicConfig.getAttributes().put(
TopicAttributes.TOPIC_MESSAGE_TYPE_ATTRIBUTE.getName(), TopicMessageType.LITE.getValue());
- ConcurrentMap topicConfigTable = new ConcurrentHashMap<>();
+ topicConfig.setLiteTopicExpiration(1);
topicConfigTable.put(PARENT_TOPIC, topicConfig);
when(topicConfigManager.getTopicConfigTable()).thenReturn(topicConfigTable);
when(topicConfigManager.selectTopicConfig(PARENT_TOPIC)).thenReturn(topicConfig);
@@ -121,6 +126,7 @@ public void setUp() {
TestLiteLifecycleManager testObject = new TestLiteLifecycleManager(brokerController, liteSharding);
lifecycleManager = Mockito.spy(testObject);
lifecycleManager.init();
+ lifecycleManager.lmqPrefixIndex.add(EXIST_LMQ_NAME);
}
@After
@@ -128,6 +134,7 @@ public void reset() {
topicConfig.getAttributes().clear();
groupConfig.getAttributes().clear();
offsetTable.clear();
+ topicConfigTable.clear();
}
@Test
@@ -154,10 +161,23 @@ public void testIsLmqExist() {
@Test
public void testGetLiteTopicCount() {
Assert.assertEquals(1, lifecycleManager.getLiteTopicCount(PARENT_TOPIC));
- verify(lifecycleManager).collectByParentTopic(PARENT_TOPIC);
-
Assert.assertEquals(0, lifecycleManager.getLiteTopicCount("whatever"));
- verify(lifecycleManager, never()).collectByParentTopic("whatever");
+
+ // parentTopic1: 2 liteTopics, parentTopic2: 3 liteTopics
+ String parent1 = "parentTopic1";
+ String parent2 = "parentTopic2";
+ registerLiteTopicConfig(parent1);
+ registerLiteTopicConfig(parent2);
+ lifecycleManager.lmqPrefixIndex.add(LiteUtil.toLmqName(parent1, "sub1"));
+ lifecycleManager.lmqPrefixIndex.add(LiteUtil.toLmqName(parent1, "sub2"));
+ lifecycleManager.lmqPrefixIndex.add(LiteUtil.toLmqName(parent2, "sub1"));
+ lifecycleManager.lmqPrefixIndex.add(LiteUtil.toLmqName(parent2, "sub2"));
+ lifecycleManager.lmqPrefixIndex.add(LiteUtil.toLmqName(parent2, "sub3"));
+
+ Assert.assertEquals(2, lifecycleManager.getLiteTopicCount(parent1));
+ Assert.assertEquals(3, lifecycleManager.getLiteTopicCount(parent2));
+ // PARENT_TOPIC count unchanged
+ Assert.assertEquals(1, lifecycleManager.getLiteTopicCount(PARENT_TOPIC));
}
@Test
@@ -238,6 +258,9 @@ public void testDeleteLmq() {
public void testCleanExpiredLiteTopic() {
String removeKey = EXIST_LMQ_NAME + TOPIC_GROUP_SEPARATOR + GROUP;
when(liteSharding.shardingByLmqName(PARENT_TOPIC, EXIST_LMQ_NAME)).thenReturn(brokerConfig.getBrokerName());
+ brokerConfig.setMinLiteTTl(0);
+ when(messageStore.getMessageStoreTimeStamp(anyString(), anyInt(), anyLong()))
+ .thenReturn(System.currentTimeMillis() - TimeUnit.MINUTES.toMillis(10));
lifecycleManager.cleanExpiredLiteTopic();
verify(consumerOffsetManager).removeConsumerOffset(removeKey);
@@ -247,22 +270,84 @@ public void testCleanExpiredLiteTopic() {
@Test
public void testCleanByParentTopic() {
- String removeKey = EXIST_LMQ_NAME + TOPIC_GROUP_SEPARATOR + GROUP;
- when(liteSharding.shardingByLmqName(PARENT_TOPIC, EXIST_LMQ_NAME)).thenReturn(brokerConfig.getBrokerName());
+ String lmq1 = LiteUtil.toLmqName(PARENT_TOPIC, "sub1");
+ String lmq2 = LiteUtil.toLmqName(PARENT_TOPIC, "sub2");
+ String lmq3 = LiteUtil.toLmqName(PARENT_TOPIC, "sub3");
+
+ String otherLmq1 = LiteUtil.toLmqName("otherParentTopic", "sub1");
+ String otherLmq2 = LiteUtil.toLmqName("otherParentTopic", "sub2");
+
+ // multiple LMQs: deleteLmq called only for LMQs under PARENT_TOPIC
+ lifecycleManager.lmqPrefixIndex.remove(EXIST_LMQ_NAME);
+ lifecycleManager.lmqPrefixIndex.add(lmq1);
+ lifecycleManager.lmqPrefixIndex.add(lmq2);
+ lifecycleManager.lmqPrefixIndex.add(lmq3);
+ lifecycleManager.lmqPrefixIndex.add(otherLmq1);
+ lifecycleManager.lmqPrefixIndex.add(otherLmq2);
+
+ ArgumentCaptor parentCaptor = ArgumentCaptor.forClass(String.class);
+ ArgumentCaptor lmqCaptor = ArgumentCaptor.forClass(String.class);
+ lifecycleManager.cleanByParentTopic(PARENT_TOPIC);
+ verify(lifecycleManager, times(3)).deleteLmq(parentCaptor.capture(), lmqCaptor.capture());
+ Assert.assertTrue(parentCaptor.getAllValues().stream().allMatch(PARENT_TOPIC::equals));
+ Assert.assertEquals(new HashSet<>(Arrays.asList(lmq1, lmq2, lmq3)), new HashSet<>(lmqCaptor.getAllValues()));
+ // other parent's LMQs remain untouched
+ List otherResult = lifecycleManager.collectByParentTopic("otherParentTopic");
+ Assert.assertEquals(new HashSet<>(Arrays.asList(otherLmq1, otherLmq2)), new HashSet<>(otherResult));
+
+ // zero LMQs: deleteLmq not called
+ Mockito.clearInvocations(lifecycleManager);
lifecycleManager.cleanByParentTopic(PARENT_TOPIC);
- verify(consumerOffsetManager).removeConsumerOffset(removeKey);
- verify(messageStore).deleteTopics(Collections.singleton(EXIST_LMQ_NAME));
- verify(liteSubscriptionRegistry).cleanSubscription(EXIST_LMQ_NAME, false);
+ verify(lifecycleManager, never()).deleteLmq(anyString(), anyString());
+
+ // guard: non-lite topic and null both return early
+ Mockito.clearInvocations(lifecycleManager);
+ lifecycleManager.lmqPrefixIndex.add(EXIST_LMQ_NAME);
+ lifecycleManager.cleanByParentTopic("nonExistentTopic");
+ verify(lifecycleManager, never()).deleteLmq(anyString(), anyString());
+ lifecycleManager.cleanByParentTopic(null);
+ verify(lifecycleManager, never()).deleteLmq(anyString(), anyString());
+ }
- lifecycleManager.cleanByParentTopic("whatever");
- verify(lifecycleManager, never()).collectByParentTopic("whatever");
+ @Test
+ public void testCollectByParentTopic() {
+ String lmq1 = LiteUtil.toLmqName(PARENT_TOPIC, "sub1");
+ String lmq2 = LiteUtil.toLmqName(PARENT_TOPIC, "sub2");
+ String lmq3 = LiteUtil.toLmqName(PARENT_TOPIC, "sub3");
+
+ String otherLmq1 = LiteUtil.toLmqName("otherParentTopic", "sub1");
+ String otherLmq2 = LiteUtil.toLmqName("otherParentTopic", "sub2");
+
+ lifecycleManager.lmqPrefixIndex.remove(EXIST_LMQ_NAME);
+ lifecycleManager.lmqPrefixIndex.add(lmq1);
+ lifecycleManager.lmqPrefixIndex.add(lmq2);
+ lifecycleManager.lmqPrefixIndex.add(lmq3);
+ lifecycleManager.lmqPrefixIndex.add(otherLmq1);
+ lifecycleManager.lmqPrefixIndex.add(otherLmq2);
+
+ // multiple LMQs: returns only those under PARENT_TOPIC, excluding other parent's
+ List result = lifecycleManager.collectByParentTopic(PARENT_TOPIC);
+ Assert.assertEquals(new HashSet<>(Arrays.asList(lmq1, lmq2, lmq3)), new HashSet<>(result));
+
+ // no LMQs under parent: returns empty list
+ result = lifecycleManager.collectByParentTopic("nonExistentTopic");
+ Assert.assertTrue(result.isEmpty());
+
+ // guard: null and empty both return empty list
+ result = lifecycleManager.collectByParentTopic(null);
+ Assert.assertTrue(result.isEmpty());
+ result = lifecycleManager.collectByParentTopic("");
+ Assert.assertTrue(result.isEmpty());
}
@Test
public void testRun() throws InterruptedException {
brokerConfig.setLiteTtlCheckInterval(100L);
+ brokerConfig.setMinLiteTTl(0);
when(liteSharding.shardingByLmqName(PARENT_TOPIC, EXIST_LMQ_NAME)).thenReturn(brokerConfig.getBrokerName());
+ when(messageStore.getMessageStoreTimeStamp(anyString(), anyInt(), anyLong()))
+ .thenReturn(System.currentTimeMillis() - TimeUnit.MINUTES.toMillis(10));
lifecycleManager.start();
Thread.sleep(300);
lifecycleManager.shutdown();
@@ -272,29 +357,40 @@ public void testRun() throws InterruptedException {
verify(liteSubscriptionRegistry, atLeastOnce()).cleanSubscription(EXIST_LMQ_NAME, false);
}
+ private void registerLiteTopicConfig(String parentTopic) {
+ TopicConfig config = new TopicConfig(parentTopic, 1, 1);
+ config.getAttributes().put(
+ TopicAttributes.TOPIC_MESSAGE_TYPE_ATTRIBUTE.getName(), TopicMessageType.LITE.getValue());
+ topicConfigTable.put(parentTopic, config);
+ when(topicConfigManager.selectTopicConfig(parentTopic)).thenReturn(config);
+ }
+
private static class TestLiteLifecycleManager extends AbstractLiteLifecycleManager {
+
public TestLiteLifecycleManager(BrokerController brokerController, LiteSharding liteSharding) {
super(brokerController, liteSharding);
}
@Override
public long getMaxOffsetInQueue(String lmqName) {
- return EXIST_LMQ_NAME.equals(lmqName) ? 100 : -1;
- }
-
- @Override
- public List> collectExpiredLiteTopic() {
- return Collections.singletonList(new Pair<>(PARENT_TOPIC, EXIST_LMQ_NAME));
- }
-
- @Override
- public List collectByParentTopic(String parentTopic) {
- return PARENT_TOPIC.equals(parentTopic) ? Collections.singletonList(EXIST_LMQ_NAME) : Collections.emptyList();
+ return LiteUtil.isLiteTopicQueue(lmqName) ? 100 : -1;
}
@Override
public void forEachLiteTopic(Function, Boolean> function) {
-
+ List> triples = new ArrayList<>();
+ lmqPrefixIndex.forEachLmqByPrefix(LiteUtil.LITE_TOPIC_PREFIX, lmqName -> {
+ long maxOffset = getMaxOffsetInQueue(lmqName);
+ if (maxOffset > 0) {
+ triples.add(Triple.of(lmqName, maxOffset, null));
+ }
+ return true;
+ });
+ for (Triple triple : triples) {
+ if (!function.apply(triple)) {
+ break;
+ }
+ }
}
}
}
diff --git a/broker/src/test/java/org/apache/rocketmq/broker/lite/LiteEventDispatcherTest.java b/broker/src/test/java/org/apache/rocketmq/broker/lite/LiteEventDispatcherTest.java
index f96e5cb80de..da40204a1fb 100644
--- a/broker/src/test/java/org/apache/rocketmq/broker/lite/LiteEventDispatcherTest.java
+++ b/broker/src/test/java/org/apache/rocketmq/broker/lite/LiteEventDispatcherTest.java
@@ -17,6 +17,7 @@
package org.apache.rocketmq.broker.lite;
+import org.apache.commons.lang3.tuple.Triple;
import org.apache.rocketmq.broker.BrokerController;
import org.apache.rocketmq.broker.offset.ConsumerOffsetManager;
import org.apache.rocketmq.broker.processor.NotificationProcessor;
@@ -42,6 +43,7 @@
import java.util.Map;
import java.util.Set;
import java.util.HashMap;
+import java.util.function.Function;
import org.mockito.Mockito;
import org.mockito.junit.MockitoJUnitRunner;
@@ -58,10 +60,12 @@
import static org.mockito.ArgumentMatchers.anyList;
import static org.mockito.ArgumentMatchers.anyString;
import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.Mockito.doAnswer;
import static org.mockito.Mockito.doReturn;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.never;
import static org.mockito.Mockito.spy;
+import static org.mockito.Mockito.times;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
@@ -110,7 +114,7 @@ public void testInitAddsListener() {
@Test
public void testDispatchWhenQueueIdNotZero() {
liteEventDispatcher.dispatch("group", "lmqName", 1, 0L, 0L);
- verify(liteSubscriptionRegistry, never()).getAllSubscriber(anyString(), anyString());
+ verify(liteSubscriptionRegistry, never()).getAllSubscribers(anyString(), anyString());
}
@Test
@@ -123,7 +127,7 @@ public void testDispatchCallsDoDispatch() {
@Test
public void testDoDispatchWhenWrapperIsNull() {
- when(liteSubscriptionRegistry.getAllSubscriber("group", "lmqName")).thenReturn(null);
+ when(liteSubscriptionRegistry.getAllSubscribers("group", "lmqName")).thenReturn(Collections.emptyMap());
// Use reflection to access private method
try {
@@ -135,7 +139,7 @@ public void testDoDispatchWhenWrapperIsNull() {
fail("Exception should not be thrown");
}
- verify(liteSubscriptionRegistry).getAllSubscriber("group", "lmqName");
+ verify(liteSubscriptionRegistry).getAllSubscribers("group", "lmqName");
}
@Test
@@ -144,11 +148,10 @@ public void testDoDispatchWithListWrapper() {
subscriptionGroupConfig.setWildcardLiteGroup(false);
when(subscriptionGroupManager.findSubscriptionGroupConfig("group")).thenReturn(subscriptionGroupConfig);
- SubscriberWrapper.ListWrapper listWrapper = mock(SubscriberWrapper.ListWrapper.class);
List clients = Collections.singletonList(new ClientGroup("clientId", "group"));
- when(listWrapper.asListWrapper()).thenReturn(listWrapper);
- when(listWrapper.getClients()).thenReturn(clients);
- when(liteSubscriptionRegistry.getAllSubscriber("group", "lmqName")).thenReturn(listWrapper);
+ Map> subscriberMap = new HashMap<>();
+ subscriberMap.put("group", clients);
+ when(liteSubscriptionRegistry.getAllSubscribers("group", "lmqName")).thenReturn(subscriberMap);
LiteEventDispatcher spyDispatcher = Mockito.spy(liteEventDispatcher);
spyDispatcher.doDispatch("group", "lmqName", null);
@@ -157,12 +160,9 @@ public void testDoDispatchWithListWrapper() {
@Test
public void testDoDispatchWithMapWrapper() {
- SubscriberWrapper.MapWrapper mapWrapper = mock(SubscriberWrapper.MapWrapper.class);
Map> groupMap = new HashMap<>();
groupMap.put("key", Collections.singletonList(new ClientGroup("clientId", "group")));
- when(mapWrapper.getGroupMap()).thenReturn(groupMap);
- when(mapWrapper.asMapWrapper()).thenReturn(mapWrapper);
- when(liteSubscriptionRegistry.getAllSubscriber("group", "lmqName")).thenReturn(mapWrapper);
+ when(liteSubscriptionRegistry.getAllSubscribers("group", "lmqName")).thenReturn(groupMap);
LiteEventDispatcher spyDispatcher = Mockito.spy(liteEventDispatcher);
@@ -270,7 +270,7 @@ public void testDoFullDispatchForClientWhenSubscriptionHasNoTopics() {
String group = "group";
LiteSubscription subscription = mock(LiteSubscription.class);
- when(subscription.getLiteTopicSet()).thenReturn(Collections.emptySet());
+ when(subscription.getLmqSet()).thenReturn(Collections.emptySet());
when(liteSubscriptionRegistry.getLiteSubscription(clientId)).thenReturn(subscription);
liteEventDispatcher.doFullDispatchForClient(clientId, group);
@@ -515,7 +515,7 @@ public void testDoFullDispatchForClientNormalCase() {
LiteSubscription subscription = new LiteSubscription();
Set topics = new HashSet<>();
topics.add(lmqName);
- subscription.setLiteTopicSet(topics);
+ subscription.setLmqSet(topics);
when(liteSubscriptionRegistry.getLiteSubscription(clientId)).thenReturn(subscription);
when(liteLifecycleManager.getMaxOffsetInQueue(lmqName)).thenReturn(100L);
@@ -544,4 +544,54 @@ public void testScan_FullDispatch() {
liteEventDispatcher.scan();
assertTrue(liteEventDispatcher.fullDispatchSet.isEmpty());
}
+
+ @Test
+ public void testDoFullDispatchForWildcardGroup_dispatchesLmqs() {
+ String group = "wildcardGroup";
+ String parentTopic = "parentTopic";
+ String lmq1 = "%LMQ%$parentTopic$sub1";
+ String lmq2 = "%LMQ%$parentTopic$sub2";
+
+ // Make isWildcardGroup return true and getLiteBindTopic return parentTopic
+ SubscriptionGroupConfig groupConfig = new SubscriptionGroupConfig();
+ groupConfig.setWildcardLiteGroup(true);
+ groupConfig.setLiteBindTopic(parentTopic);
+ when(subscriptionGroupManager.findSubscriptionGroupConfig(group)).thenReturn(groupConfig);
+
+ List clients = Collections.singletonList(new ClientGroup("clientId", group));
+ when(liteSubscriptionRegistry.getWildcardGroupClients(group)).thenReturn(clients);
+
+ doAnswer(invocation -> {
+ Function, Boolean> func = invocation.getArgument(1);
+ func.apply(Triple.of(lmq1, 100L, null));
+ func.apply(Triple.of(lmq2, 200L, null));
+ return null;
+ }).when(liteLifecycleManager).forEachLiteTopicByParent(eq(parentTopic), any());
+
+ when(consumerOffsetManager.queryOffset(group, lmq1, 0)).thenReturn(50L);
+ when(consumerOffsetManager.queryOffset(group, lmq2, 0)).thenReturn(50L);
+
+ LiteEventDispatcher spyDispatcher = Mockito.spy(liteEventDispatcher);
+ spyDispatcher.doFullDispatchForWildcardGroup(group);
+
+ verify(spyDispatcher, times(2)).selectAndDispatch(anyString(), eq(clients), eq(null));
+ }
+
+ @Test
+ public void testDoFullDispatchByGroup_nonWildcard_delegatesToClientDispatch() {
+ String group = "testGroup";
+
+ SubscriptionGroupConfig groupConfig = new SubscriptionGroupConfig();
+ groupConfig.setWildcardLiteGroup(false);
+ when(subscriptionGroupManager.findSubscriptionGroupConfig(group)).thenReturn(groupConfig);
+
+ List clientIds = Arrays.asList("client1", "client2");
+ when(liteSubscriptionRegistry.getAllClientIdByGroup(group)).thenReturn(clientIds);
+
+ LiteEventDispatcher spyDispatcher = Mockito.spy(liteEventDispatcher);
+ spyDispatcher.doFullDispatchByGroup(group);
+
+ verify(spyDispatcher).doFullDispatchForClient("client1", group);
+ verify(spyDispatcher).doFullDispatchForClient("client2", group);
+ }
}
\ No newline at end of file
diff --git a/broker/src/test/java/org/apache/rocketmq/broker/lite/LiteLifecycleManagerTest.java b/broker/src/test/java/org/apache/rocketmq/broker/lite/LiteLifecycleManagerTest.java
index 00dcb79c8de..e936fc18054 100644
--- a/broker/src/test/java/org/apache/rocketmq/broker/lite/LiteLifecycleManagerTest.java
+++ b/broker/src/test/java/org/apache/rocketmq/broker/lite/LiteLifecycleManagerTest.java
@@ -22,11 +22,8 @@
import org.apache.rocketmq.broker.subscription.SubscriptionGroupManager;
import org.apache.rocketmq.broker.topic.TopicConfigManager;
import org.apache.rocketmq.common.BrokerConfig;
-import org.apache.rocketmq.common.Pair;
-import org.apache.rocketmq.common.TopicAttributes;
import org.apache.rocketmq.common.TopicConfig;
import org.apache.rocketmq.common.UtilAll;
-import org.apache.rocketmq.common.attribute.TopicMessageType;
import org.apache.rocketmq.common.lite.LiteUtil;
import org.apache.rocketmq.store.MessageStore;
import org.junit.AfterClass;
@@ -115,48 +112,6 @@ public void testGetMaxOffsetInQueue() {
Assert.assertEquals(0, liteLifecycleManager.getMaxOffsetInQueue(UUID.randomUUID().toString()));
}
- @Test
- public void testCollectByParentTopic() {
- int num = 3;
- String parentTopic = UUID.randomUUID().toString();
- for (int i = 0; i < num; i++) {
- messageStore.putMessage(LiteTestUtil.buildMessage(parentTopic, UUID.randomUUID().toString()));
- messageStore.putMessage(LiteTestUtil.buildMessage(UUID.randomUUID().toString(), UUID.randomUUID().toString()));
- }
- await().atMost(5, SECONDS).pollInterval(200, MILLISECONDS).until(() -> messageStore.dispatchBehindBytes() <= 0);
- List result = liteLifecycleManager.collectByParentTopic(parentTopic);
- Assert.assertEquals(num, result.size());
- for (String lmqName : result) {
- Assert.assertTrue(LiteUtil.belongsTo(lmqName, parentTopic));
- }
-
- result = liteLifecycleManager.collectByParentTopic(UUID.randomUUID().toString());
- Assert.assertEquals(0, result.size());
- }
-
- @Test
- public void testCollectExpiredLiteTopic() {
- int num = 3;
- String parentTopic = UUID.randomUUID().toString();
- for (int i = 0; i < num; i++) {
- messageStore.putMessage(LiteTestUtil.buildMessage(parentTopic, UUID.randomUUID().toString()));
- messageStore.putMessage(LiteTestUtil.buildMessage(UUID.randomUUID().toString(), null));
- }
- await().atMost(5, SECONDS).pollInterval(200, MILLISECONDS).until(() -> messageStore.dispatchBehindBytes() <= 0);
-
- when(liteLifecycleManager.isLiteTopicExpired(anyString(), anyString(), anyLong())).thenReturn(false);
- List> result = liteLifecycleManager.collectExpiredLiteTopic();
- Assert.assertEquals(0, result.size());
-
- when(liteLifecycleManager.isLiteTopicExpired(eq(parentTopic), anyString(), anyLong())).thenReturn(true);
- result = liteLifecycleManager.collectExpiredLiteTopic();
- Assert.assertEquals(num, result.size());
- for (Pair pair : result) {
- Assert.assertEquals(parentTopic, pair.getObject1());
- Assert.assertTrue(LiteUtil.belongsTo(pair.getObject2(), parentTopic));
- }
- }
-
@Ignore
@Test
public void testCleanExpiredLiteTopic() {
@@ -182,31 +137,4 @@ public void testCleanExpiredLiteTopic() {
Assert.assertFalse(messageStore.getQueueStore().getConsumeQueueTable().containsKey(lmqName));
}
}
-
- @Test
- public void testCleanByParentTopic() {
- int num = 3;
- String parentTopic = UUID.randomUUID().toString();
- mockTopicConfig.getAttributes().put(
- TopicAttributes.TOPIC_MESSAGE_TYPE_ATTRIBUTE.getName(), TopicMessageType.LITE.getValue());
-
- List liteTopics =
- IntStream.range(0, 3).mapToObj(i -> UUID.randomUUID().toString()).collect(Collectors.toList());
- for (int i = 0; i < num; i++) {
- messageStore.putMessage(LiteTestUtil.buildMessage(parentTopic, liteTopics.get(i)));
- }
- await().atMost(5, SECONDS).pollInterval(200, MILLISECONDS).until(() -> messageStore.dispatchBehindBytes() <= 0);
-
- for (int i = 0; i < num; i++) {
- String lmqName = LiteUtil.toLmqName(parentTopic, liteTopics.get(i));
- Assert.assertTrue(messageStore.getQueueStore().getConsumeQueueTable().containsKey(lmqName));
- }
-
- liteLifecycleManager.cleanByParentTopic(parentTopic);
-
- for (int i = 0; i < num; i++) {
- String lmqName = LiteUtil.toLmqName(parentTopic, liteTopics.get(i));
- Assert.assertFalse(messageStore.getQueueStore().getConsumeQueueTable().containsKey(lmqName));
- }
- }
-}
+}
\ No newline at end of file
diff --git a/broker/src/test/java/org/apache/rocketmq/broker/lite/LiteSubscriptionRegistryImplTest.java b/broker/src/test/java/org/apache/rocketmq/broker/lite/LiteSubscriptionRegistryImplTest.java
index 7645a470962..505613508a2 100644
--- a/broker/src/test/java/org/apache/rocketmq/broker/lite/LiteSubscriptionRegistryImplTest.java
+++ b/broker/src/test/java/org/apache/rocketmq/broker/lite/LiteSubscriptionRegistryImplTest.java
@@ -21,6 +21,7 @@
import java.util.Collections;
import java.util.HashSet;
import java.util.List;
+import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import org.apache.rocketmq.broker.BrokerController;
@@ -46,7 +47,6 @@
import static org.junit.Assert.assertNull;
import static org.junit.Assert.assertThrows;
import static org.junit.Assert.assertTrue;
-import static org.junit.jupiter.api.Assertions.assertInstanceOf;
import static org.mockito.ArgumentMatchers.anyString;
import static org.mockito.ArgumentMatchers.eq;
import static org.mockito.Mockito.mock;
@@ -167,7 +167,7 @@ public void testAddPartialSubscription_InactiveSubscription() {
LiteSubscription subscription = registry.getLiteSubscription(clientId);
assertNotNull(subscription);
- assertFalse(subscription.getLiteTopicSet().contains("lmq1"));
+ assertFalse(subscription.getLmqSet().contains("lmq1"));
assertEquals(0, registry.getActiveSubscriptionNum());
}
@@ -191,7 +191,7 @@ public void testAddPartialSubscription_NormalCase() {
LiteSubscription subscription = registry.getLiteSubscription(clientId);
assertNotNull(subscription);
- assertTrue(subscription.getLiteTopicSet().contains("lmq1"));
+ assertTrue(subscription.getLmqSet().contains("lmq1"));
assertEquals(1, registry.getActiveSubscriptionNum());
verify(mockListener).onRegister(clientId, group, "lmq1");
@@ -220,7 +220,7 @@ public void testAddPartialSubscription_ExclusiveMode() {
LiteSubscription subscription1 = registry.getLiteSubscription(clientId1);
assertNotNull(subscription1);
- assertTrue(subscription1.getLiteTopicSet().contains("lmq1"));
+ assertTrue(subscription1.getLmqSet().contains("lmq1"));
assertEquals(1, registry.getActiveSubscriptionNum());
// Add second client, should exclude first client
@@ -228,7 +228,7 @@ public void testAddPartialSubscription_ExclusiveMode() {
LiteSubscription subscription2 = registry.getLiteSubscription(clientId2);
assertNotNull(subscription2);
- assertTrue(subscription2.getLiteTopicSet().contains("lmq1"));
+ assertTrue(subscription2.getLmqSet().contains("lmq1"));
assertNull(registry.getLiteSubscription(clientId1));
assertEquals(1, registry.getActiveSubscriptionNum());
@@ -260,8 +260,8 @@ public void testRemovePartialSubscription_RemoveSubscription() {
LiteSubscription subscription = registry.getLiteSubscription(clientId);
assertNotNull(subscription);
- assertTrue(subscription.getLiteTopicSet().contains("lmq1"));
- assertTrue(subscription.getLiteTopicSet().contains("lmq2"));
+ assertTrue(subscription.getLmqSet().contains("lmq1"));
+ assertTrue(subscription.getLmqSet().contains("lmq2"));
assertEquals(2, registry.getActiveSubscriptionNum());
// Remove partial subscription
@@ -270,8 +270,8 @@ public void testRemovePartialSubscription_RemoveSubscription() {
subscription = registry.getLiteSubscription(clientId);
assertNotNull(subscription);
- assertFalse(subscription.getLiteTopicSet().contains("lmq1"));
- assertTrue(subscription.getLiteTopicSet().contains("lmq2"));
+ assertFalse(subscription.getLmqSet().contains("lmq1"));
+ assertTrue(subscription.getLmqSet().contains("lmq2"));
assertEquals(1, registry.getActiveSubscriptionNum());
verify(mockListener).onUnregister(clientId, group, "lmq1");
@@ -303,7 +303,7 @@ public void testAddCompleteSubscription_WildcardGroup() {
LiteSubscription subscription = registry.getLiteSubscription(clientId);
assertNotNull(subscription);
- assertTrue(subscription.getLiteTopicSet().contains(topic + "@" + group));
+ assertTrue(subscription.getLmqSet().contains(topic + "@" + group));
assertEquals(1, registry.getActiveSubscriptionNum());
}
@@ -363,8 +363,8 @@ public void testAddCompleteSubscription_UpdateSubscription() {
LiteSubscription subscription = registry.getLiteSubscription(clientId);
assertNotNull(subscription);
- assertTrue(subscription.getLiteTopicSet().contains("lmq1"));
- assertTrue(subscription.getLiteTopicSet().contains("lmq2"));
+ assertTrue(subscription.getLmqSet().contains("lmq1"));
+ assertTrue(subscription.getLmqSet().contains("lmq2"));
assertEquals(2, registry.getActiveSubscriptionNum());
// Update subscription
@@ -372,9 +372,9 @@ public void testAddCompleteSubscription_UpdateSubscription() {
subscription = registry.getLiteSubscription(clientId);
assertNotNull(subscription);
- assertFalse(subscription.getLiteTopicSet().contains("lmq1"));
- assertTrue(subscription.getLiteTopicSet().contains("lmq2"));
- assertTrue(subscription.getLiteTopicSet().contains("lmq3"));
+ assertFalse(subscription.getLmqSet().contains("lmq1"));
+ assertTrue(subscription.getLmqSet().contains("lmq2"));
+ assertTrue(subscription.getLmqSet().contains("lmq3"));
assertEquals(2, registry.getActiveSubscriptionNum());
}
@@ -402,8 +402,8 @@ public void testRemoveCompleteSubscription_RemoveAll() {
LiteSubscription subscription = registry.getLiteSubscription(clientId);
assertNotNull(subscription);
- assertTrue(subscription.getLiteTopicSet().contains("lmq1"));
- assertTrue(subscription.getLiteTopicSet().contains("lmq2"));
+ assertTrue(subscription.getLmqSet().contains("lmq1"));
+ assertTrue(subscription.getLmqSet().contains("lmq2"));
assertEquals(2, registry.getActiveSubscriptionNum());
// Remove complete subscription
@@ -432,28 +432,35 @@ public void testAddListener_AddListener() {
* Test getAllSubscriber gets wildcard subscribers
*/
@Test
- public void testGetAllSubscriber_WildcardGroup() {
+ public void testGetAllSubscribers_WildcardGroup() {
String group = "testGroup";
String topic = "testTopic";
- String lmqName = topic + "@" + group;
+ String lmqName = LiteUtil.toLmqName(topic, "liteTopic");
+ String wildcardLmqName = topic + "@" + group;
- // Simulate wildcard group
+ // Simulate wildcard group with subscription data
SubscriptionGroupConfig groupConfig = new SubscriptionGroupConfig();
groupConfig.setGroupName(group);
groupConfig.setWildcardLiteGroup(true);
+ groupConfig.setLiteBindTopic(topic);
when(mockSubscriptionGroupManager.findSubscriptionGroupConfig(group)).thenReturn(groupConfig);
- SubscriberWrapper result = registry.getAllSubscriber(group, lmqName);
+ ClientGroup clientGroup = new ClientGroup("testClient", group);
+ Set clientSet = ConcurrentHashMap.newKeySet();
+ clientSet.add(clientGroup);
+ registry.liteTopic2ClientGroup.put(wildcardLmqName, clientSet);
+
+ Map> result = registry.getAllSubscribers(group, lmqName);
assertNotNull(result);
- assertInstanceOf(SubscriberWrapper.ListWrapper.class, result);
+ assertTrue(result.containsKey(group));
}
/**
* Test getAllSubscriber gets subscribers for specific group
*/
@Test
- public void testGetAllSubscriber_SpecificGroup() {
+ public void testGetAllSubscribers_SpecificGroup() {
String clientId = "testClient";
String group = "testGroup";
String lmqName = "lmq1";
@@ -462,16 +469,16 @@ public void testGetAllSubscriber_SpecificGroup() {
ClientGroup clientGroup = new ClientGroup(clientId, group);
Set clientSet = ConcurrentHashMap.newKeySet();
clientSet.add(clientGroup);
- registry.liteTopic2Group.put(lmqName, clientSet);
+ registry.liteTopic2ClientGroup.put(lmqName, clientSet);
- SubscriberWrapper result = registry.getAllSubscriber(group, lmqName);
+ Map> result = registry.getAllSubscribers(group, lmqName);
assertNotNull(result);
- assertInstanceOf(SubscriberWrapper.ListWrapper.class, result);
- SubscriberWrapper.ListWrapper listWrapper = (SubscriberWrapper.ListWrapper) result;
- assertEquals(1, listWrapper.getClients().size());
- assertEquals(clientId, listWrapper.getClients().get(0).clientId);
- assertEquals(group, listWrapper.getClients().get(0).group);
+ assertTrue(result.containsKey(group));
+ List clients = result.get(group);
+ assertEquals(1, clients.size());
+ assertEquals(clientId, clients.get(0).clientId);
+ assertEquals(group, clients.get(0).group);
}
/**
@@ -492,18 +499,16 @@ public void testGetAllSubscriber_AllGroups() {
Set clientSet = ConcurrentHashMap.newKeySet();
clientSet.add(clientGroup1);
clientSet.add(clientGroup2);
- registry.liteTopic2Group.put(lmqName, clientSet);
+ registry.liteTopic2ClientGroup.put(lmqName, clientSet);
- SubscriberWrapper result = registry.getAllSubscriber(null, lmqName);
+ Map> result = registry.getAllSubscribers(null, lmqName);
assertNotNull(result);
- assertInstanceOf(SubscriberWrapper.MapWrapper.class, result);
- SubscriberWrapper.MapWrapper mapWrapper = (SubscriberWrapper.MapWrapper) result;
- assertEquals(2, mapWrapper.getGroupMap().size());
- assertTrue(mapWrapper.getGroupMap().containsKey(group1));
- assertTrue(mapWrapper.getGroupMap().containsKey(group2));
- assertEquals(1, mapWrapper.getGroupMap().get(group1).size());
- assertEquals(1, mapWrapper.getGroupMap().get(group2).size());
+ assertEquals(2, result.size());
+ assertTrue(result.containsKey(group1));
+ assertTrue(result.containsKey(group2));
+ assertEquals(1, result.get(group1).size());
+ assertEquals(1, result.get(group2).size());
}
/**
@@ -519,18 +524,18 @@ public void testCleanSubscription_CleanSubscription() {
ClientGroup clientGroup = new ClientGroup(clientId, group);
Set clientSet = ConcurrentHashMap.newKeySet();
clientSet.add(clientGroup);
- registry.liteTopic2Group.put(lmqName, clientSet);
+ registry.liteTopic2ClientGroup.put(lmqName, clientSet);
LiteSubscription subscription = new LiteSubscription();
subscription.setGroup(group);
- subscription.addLiteTopic(lmqName);
+ subscription.addLmq(lmqName);
registry.client2Subscription.put(clientId, subscription);
registry.activeNum.set(1);
registry.cleanSubscription(lmqName, false);
- assertFalse(registry.liteTopic2Group.containsKey(lmqName));
- assertFalse(subscription.getLiteTopicSet().contains(lmqName));
+ assertFalse(registry.liteTopic2ClientGroup.containsKey(lmqName));
+ assertFalse(subscription.getLmqSet().contains(lmqName));
assertEquals(0, registry.getActiveSubscriptionNum());
}
@@ -924,4 +929,285 @@ public void testExclusiveEviction_CompleteSyncReNotifiesForTombstonedLmq() {
verify(mockBroker2Client, org.mockito.Mockito.atLeast(2))
.notifyUnsubscribeLite(eq(clientAChannel), captor.capture());
}
+
+ // ==================== resetOffset Edge Cases ====================
+
+ /**
+ * Test: resetOffset with null option is a no-op
+ */
+ @Test
+ public void testResetOffset_NullOption() {
+ registry.resetOffset("lmq1", "group", "client", null);
+ // No interaction with offset manager
+ org.mockito.Mockito.verifyNoInteractions(mockConsumerOffsetManager);
+ }
+
+ /**
+ * Test: resetOffset with TAIL_N computes target correctly
+ */
+ @Test
+ public void testResetOffset_TailN() {
+ String lmqName = "lmq1";
+ String group = "testGroup";
+ when(mockConsumerOffsetManager.queryOffset(group, lmqName, 0)).thenReturn(100L);
+
+ OffsetOption option = new OffsetOption(OffsetOption.Type.TAIL_N, 30);
+ registry.resetOffset(lmqName, group, "client", option);
+
+ // targetOffset = max(0, 100 - 30) = 70
+ verify(mockConsumerOffsetManager).assignResetOffset(lmqName, group, 0, 70L);
+ }
+
+ /**
+ * Test: resetOffset with TAIL_N when no existing offset (currentOffset < 0)
+ */
+ @Test
+ public void testResetOffset_TailN_NoExistingOffset() {
+ String lmqName = "lmq1";
+ String group = "testGroup";
+ when(mockConsumerOffsetManager.queryOffset(group, lmqName, 0)).thenReturn(-1L);
+
+ OffsetOption option = new OffsetOption(OffsetOption.Type.TAIL_N, 30);
+ registry.resetOffset(lmqName, group, "client", option);
+
+ // currentOffset < 0 → targetOffset stays null → no reset
+ org.mockito.Mockito.verify(mockConsumerOffsetManager, org.mockito.Mockito.never())
+ .assignResetOffset(anyString(), anyString(), eq(0), eq(0L));
+ }
+
+ /**
+ * Test: resetOffset with TIMESTAMP is silently disabled
+ */
+ @Test
+ public void testResetOffset_Timestamp() {
+ String lmqName = "lmq1";
+ String group = "testGroup";
+ when(mockConsumerOffsetManager.queryOffset(group, lmqName, 0)).thenReturn(100L);
+
+ OffsetOption option = new OffsetOption(OffsetOption.Type.TIMESTAMP, System.currentTimeMillis());
+ registry.resetOffset(lmqName, group, "client", option);
+
+ // TIMESTAMP is disabled → no reset
+ org.mockito.Mockito.verify(mockConsumerOffsetManager, org.mockito.Mockito.never())
+ .assignResetOffset(anyString(), anyString(), eq(0), eq(0L));
+ }
+
+ /**
+ * Test: resetOffset skips when target equals current
+ */
+ @Test
+ public void testResetOffset_SameOffset_NoReset() {
+ String lmqName = "lmq1";
+ String group = "testGroup";
+ when(mockConsumerOffsetManager.queryOffset(group, lmqName, 0)).thenReturn(250L);
+
+ OffsetOption option = new OffsetOption(OffsetOption.Type.OFFSET, 250L);
+ registry.resetOffset(lmqName, group, "client", option);
+
+ org.mockito.Mockito.verify(mockConsumerOffsetManager, org.mockito.Mockito.never())
+ .assignResetOffset(anyString(), anyString(), eq(0), eq(0L));
+ }
+
+ // ==================== removePartialSubscription Supplements ====================
+
+ /**
+ * Test: removePartialSubscription triggers resetOffset when group has resetOffsetOnUnsubscribe
+ */
+ @Test
+ public void testRemovePartialSubscription_ResetOffsetOnUnsubscribe() {
+ String clientId = "testClient";
+ String group = "testGroup";
+ String topic = "testTopic";
+ String lmqName = "lmq1";
+
+ SubscriptionGroupConfig groupConfig = new SubscriptionGroupConfig();
+ groupConfig.setGroupName(group);
+ groupConfig.getAttributes().put("lite.sub.reset.offset.unsubscribe", "true");
+ when(mockSubscriptionGroupManager.findSubscriptionGroupConfig(group)).thenReturn(groupConfig);
+ when(mockLifecycleManager.isSubscriptionActive(topic, lmqName)).thenReturn(true);
+
+ registry.addPartialSubscription(clientId, group, topic, Collections.singleton(lmqName), null);
+
+ when(mockConsumerOffsetManager.queryOffset(group, lmqName, 0)).thenReturn(100L);
+
+ registry.removePartialSubscription(clientId, group, topic, Collections.singleton(lmqName));
+
+ // resetOffset should be called with POLICY MIN
+ verify(mockConsumerOffsetManager).assignResetOffset(eq(lmqName), eq(group), eq(0), eq(0L));
+ }
+
+
+ // ==================== cleanSubscription Supplements ====================
+
+ /**
+ * Test: cleanSubscription with notifyClient=true sends notification
+ */
+ @Test
+ public void testCleanSubscription_NotifyClient() {
+ String clientId = "testClient";
+ String group = "testGroup";
+ String topic = "testTopic";
+ String lmqName = LiteUtil.toLmqName(topic, "liteTopic");
+ Channel mockChannel = mock(Channel.class);
+
+ registry.clientChannels.put(clientId, mockChannel);
+ ClientGroup clientGroup = new ClientGroup(clientId, group);
+ Set clientSet = ConcurrentHashMap.newKeySet();
+ clientSet.add(clientGroup);
+ registry.liteTopic2ClientGroup.put(lmqName, clientSet);
+
+ LiteSubscription subscription = new LiteSubscription();
+ subscription.setGroup(group);
+ subscription.addLmq(lmqName);
+ registry.client2Subscription.put(clientId, subscription);
+ registry.activeNum.set(1);
+
+ registry.cleanSubscription(lmqName, true);
+
+ verify(mockBroker2Client).notifyUnsubscribeLite(eq(mockChannel),
+ org.mockito.Mockito.any(NotifyUnsubscribeLiteRequestHeader.class));
+ }
+
+ /**
+ * Test: cleanSubscription with empty/nonexistent lmq is a no-op
+ */
+ @Test
+ public void testCleanSubscription_EmptyClientSet() {
+ int beforeActive = registry.getActiveSubscriptionNum();
+ registry.cleanSubscription("nonexistent_lmq", true);
+ assertEquals(beforeActive, registry.getActiveSubscriptionNum());
+ }
+
+ /**
+ * Test: cleanSubscription skips clientGroup when client2Subscription has no entry
+ */
+ @Test
+ public void testCleanSubscription_NullSubscription() {
+ String lmqName = "lmq1";
+ ClientGroup orphanCg = new ClientGroup("orphanClient", "orphanGroup");
+ Set clientSet = ConcurrentHashMap.newKeySet();
+ clientSet.add(orphanCg);
+ registry.liteTopic2ClientGroup.put(lmqName, clientSet);
+ registry.activeNum.set(1);
+
+ // client2Subscription has no entry for "orphanClient"
+ registry.cleanSubscription(lmqName, false);
+
+ // lmqName removed from liteTopic2ClientGroup, activeNum unchanged (removeLmq returned false)
+ assertFalse(registry.liteTopic2ClientGroup.containsKey(lmqName));
+ }
+
+ // ==================== getWildcardGroupClients Direct Tests ====================
+
+ /**
+ * Test: getWildcardGroupClients returns clients when data exists
+ */
+ @Test
+ public void testGetWildcardGroupClients_HasClients() {
+ String group = "wildcardGroup";
+ String topic = "testTopic";
+
+ SubscriptionGroupConfig groupConfig = new SubscriptionGroupConfig();
+ groupConfig.setGroupName(group);
+ groupConfig.setWildcardLiteGroup(true);
+ groupConfig.setLiteBindTopic(topic);
+ when(mockSubscriptionGroupManager.findSubscriptionGroupConfig(group)).thenReturn(groupConfig);
+
+ ClientGroup cg = new ClientGroup("client1", group);
+ Set clientSet = ConcurrentHashMap.newKeySet();
+ clientSet.add(cg);
+ registry.liteTopic2ClientGroup.put(topic + "@" + group, clientSet);
+
+ List result = registry.getWildcardGroupClients(group);
+ assertNotNull(result);
+ assertEquals(1, result.size());
+ assertEquals("client1", result.get(0).clientId);
+ }
+
+ /**
+ * Test: getWildcardGroupClients returns empty list when bindTopic is null
+ */
+ @Test
+ public void testGetWildcardGroupClients_NoBindTopic() {
+ String group = "wildcardGroup";
+
+ SubscriptionGroupConfig groupConfig = new SubscriptionGroupConfig();
+ groupConfig.setGroupName(group);
+ groupConfig.setWildcardLiteGroup(true);
+ // No liteBindTopic set → getLiteBindTopic returns null
+ when(mockSubscriptionGroupManager.findSubscriptionGroupConfig(group)).thenReturn(groupConfig);
+
+ List result = registry.getWildcardGroupClients(group);
+ assertNotNull(result);
+ assertTrue(result.isEmpty());
+ }
+
+ // ==================== Boundary / Defensive Tests ====================
+
+ /**
+ * Test: removeCompleteSubscription with nonexistent clientId is a no-op
+ */
+ @Test
+ public void testRemoveCompleteSubscription_NullSubscription() {
+ // Should not throw
+ registry.removeCompleteSubscription("nonexistent_client");
+ }
+
+ /**
+ * Test: removeCompleteSubscription for non-exclusive group does not clear tombstones
+ */
+ @Test
+ public void testRemoveCompleteSubscription_NonExclusiveGroup() {
+ String clientId = "testClient";
+ String group = "normalGroup";
+ String topic = "testTopic";
+
+ SubscriptionGroupConfig groupConfig = new SubscriptionGroupConfig();
+ groupConfig.setGroupName(group);
+ when(mockSubscriptionGroupManager.findSubscriptionGroupConfig(group)).thenReturn(groupConfig);
+ when(mockLifecycleManager.isSubscriptionActive(eq(topic), anyString())).thenReturn(true);
+
+ registry.addPartialSubscription(clientId, group, topic, Collections.singleton("lmq1"), null);
+
+ // Manually add a tombstone to verify it's NOT cleaned for non-exclusive
+ // (non-exclusive groups shouldn't have tombstones, but verify the guard logic)
+ registry.removeCompleteSubscription(clientId);
+ assertNull(registry.getLiteSubscription(clientId));
+ }
+
+ /**
+ * Test: notifyUnsubscribeLite with null channel does not throw
+ */
+ @Test
+ public void testNotifyUnsubscribeLite_ChannelNull() {
+ String lmqName = LiteUtil.toLmqName("testTopic", "liteTopic");
+ // No channel registered for this client
+ registry.notifyUnsubscribeLite("unknownClient", "group", lmqName);
+
+ // broker2Client should not be called
+ org.mockito.Mockito.verifyNoInteractions(mockBroker2Client);
+ }
+
+ /**
+ * Test: excludeClientByLmqName with empty client set is a no-op
+ */
+ @Test
+ public void testExcludeClientByLmqName_EmptyClientSet() {
+ // No subscribers for lmq1
+ int activeBefore = registry.getActiveSubscriptionNum();
+ // excludeClientByLmqName is protected, test through addPartialSubscription in exclusive mode
+ // But we can verify indirectly: adding a new client to an empty lmq should not trigger exclusion logic
+ SubscriptionGroupConfig groupConfig = new SubscriptionGroupConfig();
+ groupConfig.setGroupName("exclusiveGroup");
+ groupConfig.setLiteSubExclusive(true);
+ when(mockSubscriptionGroupManager.findSubscriptionGroupConfig("exclusiveGroup")).thenReturn(groupConfig);
+ when(mockLifecycleManager.isSubscriptionActive("testTopic", "lmq1")).thenReturn(true);
+
+ registry.addPartialSubscription("newClient", "exclusiveGroup", "testTopic",
+ Collections.singleton("lmq1"), null);
+
+ assertEquals(activeBefore + 1, registry.getActiveSubscriptionNum());
+ assertFalse(registry.hasExclusiveEvictionTombstone("newClient", "lmq1"));
+ }
+
}
diff --git a/broker/src/test/java/org/apache/rocketmq/broker/lite/LmqPrefixIndexTest.java b/broker/src/test/java/org/apache/rocketmq/broker/lite/LmqPrefixIndexTest.java
new file mode 100644
index 00000000000..e3bc6903cb0
--- /dev/null
+++ b/broker/src/test/java/org/apache/rocketmq/broker/lite/LmqPrefixIndexTest.java
@@ -0,0 +1,229 @@
+/*
+ * 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.ArrayList;
+import java.util.List;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import org.apache.rocketmq.common.lite.LiteUtil;
+import org.junit.Before;
+import org.junit.Test;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
+
+public class LmqPrefixIndexTest {
+
+ private LmqPrefixIndex index;
+
+ @Before
+ public void setUp() {
+ index = new LmqPrefixIndex();
+ }
+
+ // --- add ---
+
+ @Test
+ public void addBasic() {
+ String lmqName = LiteUtil.toLmqName("topicA", "lite1");
+ assertTrue(index.add(lmqName));
+ assertEquals(1, index.size());
+ }
+
+ @Test
+ public void addNull() {
+ assertFalse(index.add(null));
+ assertEquals(0, index.size());
+ }
+
+ @Test
+ public void addDuplicate() {
+ String lmqName = LiteUtil.toLmqName("topicA", "lite1");
+ assertTrue(index.add(lmqName));
+ assertFalse(index.add(lmqName));
+ assertEquals(1, index.size());
+ }
+
+ // --- remove ---
+
+ @Test
+ public void removeExisting() {
+ String lmqName = LiteUtil.toLmqName("topicA", "lite1");
+ index.add(lmqName);
+ assertTrue(index.remove(lmqName));
+ assertEquals(0, index.size());
+ }
+
+ @Test
+ public void removeNonExistent() {
+ assertFalse(index.remove(LiteUtil.toLmqName("topicA", "nonexistent")));
+ assertEquals(0, index.size());
+ }
+
+ // --- forEachLmqByPrefix ---
+
+ @Test
+ public void forEachByPrefixMatchesMultiple() {
+ String lmq1 = LiteUtil.toLmqName("topicA", "lite1");
+ String lmq2 = LiteUtil.toLmqName("topicA", "lite2");
+ String lmq3 = LiteUtil.toLmqName("topicB", "lite1");
+ index.add(lmq1);
+ index.add(lmq2);
+ index.add(lmq3);
+
+ String prefix = LiteUtil.LITE_TOPIC_PREFIX + "topicA";
+ List collected = new ArrayList<>();
+ boolean completed = index.forEachLmqByPrefix(prefix, name -> {
+ collected.add(name);
+ return true;
+ });
+
+ assertTrue(completed);
+ assertEquals(2, collected.size());
+ assertTrue(collected.contains(lmq1));
+ assertTrue(collected.contains(lmq2));
+ assertFalse(collected.contains(lmq3));
+ }
+
+ @Test
+ public void forEachByPrefixExactMatch() {
+ String lmq1 = LiteUtil.toLmqName("topicA", "lite1");
+ String lmq2 = LiteUtil.toLmqName("topicA", "lite2");
+ index.add(lmq1);
+ index.add(lmq2);
+
+ List collected = new ArrayList<>();
+ boolean completed = index.forEachLmqByPrefix(lmq1, name -> {
+ collected.add(name);
+ return true;
+ });
+
+ assertTrue(completed);
+ assertEquals(1, collected.size());
+ assertEquals(lmq1, collected.get(0));
+ }
+
+ @Test
+ public void forEachByPrefixNoMatch() {
+ index.add(LiteUtil.toLmqName("topicA", "lite1"));
+
+ AtomicInteger visitCount = new AtomicInteger(0);
+ boolean completed = index.forEachLmqByPrefix(
+ LiteUtil.LITE_TOPIC_PREFIX + "topicX", name -> {
+ visitCount.incrementAndGet();
+ return true;
+ });
+
+ assertTrue(completed);
+ assertEquals(0, visitCount.get());
+ }
+
+ @Test
+ public void forEachByPrefixEarlyBreak() {
+ index.add(LiteUtil.toLmqName("topicA", "lite1"));
+ index.add(LiteUtil.toLmqName("topicA", "lite2"));
+ index.add(LiteUtil.toLmqName("topicA", "lite3"));
+
+ List collected = new ArrayList<>();
+ boolean completed = index.forEachLmqByPrefix(
+ LiteUtil.LITE_TOPIC_PREFIX + "topicA", name -> {
+ collected.add(name);
+ return collected.size() < 2;
+ });
+
+ assertFalse(completed);
+ assertEquals(2, collected.size());
+ }
+
+ @Test
+ public void forEachByPrefixEmptyPrefix() {
+ index.add(LiteUtil.toLmqName("topicA", "lite1"));
+
+ assertFalse(index.forEachLmqByPrefix("", name -> true));
+ assertFalse(index.forEachLmqByPrefix(null, name -> true));
+ }
+
+ @Test
+ public void forEachByPrefixNullVisitor() {
+ index.add(LiteUtil.toLmqName("topicA", "lite1"));
+ assertFalse(index.forEachLmqByPrefix(LiteUtil.LITE_TOPIC_PREFIX + "topicA", null));
+ }
+
+ // --- isEmpty / size ---
+
+ @Test
+ public void isEmptyAndSize() {
+ assertTrue(index.isEmpty());
+ assertEquals(0, index.size());
+
+ String lmq1 = LiteUtil.toLmqName("topicA", "lite1");
+ String lmq2 = LiteUtil.toLmqName("topicA", "lite2");
+
+ index.add(lmq1);
+ assertFalse(index.isEmpty());
+ assertEquals(1, index.size());
+
+ index.add(lmq2);
+ assertEquals(2, index.size());
+
+ index.remove(lmq1);
+ assertFalse(index.isEmpty());
+ assertEquals(1, index.size());
+
+ index.remove(lmq2);
+ assertTrue(index.isEmpty());
+ assertEquals(0, index.size());
+ }
+
+ // --- concurrency ---
+
+ @Test
+ public void concurrentAddAndForEach() throws Exception {
+ int threads = 4;
+ int entriesPerThread = 500;
+ ExecutorService executor = Executors.newFixedThreadPool(threads);
+ CountDownLatch latch = new CountDownLatch(threads);
+
+ for (int t = 0; t < threads; t++) {
+ final int threadIdx = t;
+ executor.submit(() -> {
+ try {
+ for (int i = 0; i < entriesPerThread; i++) {
+ String lmqName = LiteUtil.toLmqName("topic" + threadIdx, "lite" + i);
+ index.add(lmqName);
+ }
+ // concurrent prefix scan while other threads may still be writing
+ String prefix = LiteUtil.LITE_TOPIC_PREFIX + "topic" + threadIdx;
+ index.forEachLmqByPrefix(prefix, name -> true);
+ } finally {
+ latch.countDown();
+ }
+ });
+ }
+
+ assertTrue(latch.await(30, TimeUnit.SECONDS));
+ executor.shutdown();
+ assertEquals(threads * entriesPerThread, index.size());
+ }
+}
diff --git a/broker/src/test/java/org/apache/rocketmq/broker/lite/RocksDBLiteLifecycleManagerTest.java b/broker/src/test/java/org/apache/rocketmq/broker/lite/RocksDBLiteLifecycleManagerTest.java
index 47db902ebce..dd34840535e 100644
--- a/broker/src/test/java/org/apache/rocketmq/broker/lite/RocksDBLiteLifecycleManagerTest.java
+++ b/broker/src/test/java/org/apache/rocketmq/broker/lite/RocksDBLiteLifecycleManagerTest.java
@@ -21,11 +21,8 @@
import org.apache.rocketmq.broker.subscription.SubscriptionGroupManager;
import org.apache.rocketmq.broker.topic.TopicConfigManager;
import org.apache.rocketmq.common.BrokerConfig;
-import org.apache.rocketmq.common.Pair;
-import org.apache.rocketmq.common.TopicAttributes;
import org.apache.rocketmq.common.TopicConfig;
import org.apache.rocketmq.common.UtilAll;
-import org.apache.rocketmq.common.attribute.TopicMessageType;
import org.apache.rocketmq.common.lite.LiteUtil;
import org.apache.rocketmq.store.MessageStore;
import org.apache.rocketmq.store.config.MessageStoreConfig;
@@ -66,7 +63,6 @@ public class RocksDBLiteLifecycleManagerTest {
private static String storePathRootDir;
private static MessageStore messageStore;
private static RocksDBLiteLifecycleManager liteLifecycleManager;
- private static TopicConfig mockTopicConfig = new TopicConfig();
@BeforeClass
public static void setUp() throws Exception {
@@ -87,7 +83,6 @@ public static void setUp() throws Exception {
when(brokerController.getTopicConfigManager()).thenReturn(topicConfigManager);
when(brokerController.getSubscriptionGroupManager()).thenReturn(subscriptionGroupManager);
when(topicConfigManager.getTopicConfigTable()).thenReturn(TOPIC_CONFIG_TABLE);
- when(topicConfigManager.selectTopicConfig(anyString())).thenReturn(mockTopicConfig);
when(subscriptionGroupManager.getSubscriptionGroupTable()).thenReturn(new ConcurrentHashMap<>());
RocksDBLiteLifecycleManager testObject = new RocksDBLiteLifecycleManager(brokerController, liteSharding);
@@ -100,7 +95,6 @@ public static void reset() {
messageStore.shutdown();
messageStore.destroy();
UtilAll.deleteFile(new File(storePathRootDir));
- mockTopicConfig = new TopicConfig();
}
@Ignore
@@ -147,48 +141,6 @@ public void testGetMaxOffsetInQueue() {
Assert.assertEquals(0, liteLifecycleManager.getMaxOffsetInQueue(UUID.randomUUID().toString()));
}
- @Test
- public void testCollectByParentTopic() {
- int num = 3;
- String parentTopic = UUID.randomUUID().toString();
- for (int i = 0; i < num; i++) {
- messageStore.putMessage(LiteTestUtil.buildMessage(parentTopic, UUID.randomUUID().toString()));
- messageStore.putMessage(LiteTestUtil.buildMessage(UUID.randomUUID().toString(), UUID.randomUUID().toString()));
- }
- await().atMost(5, SECONDS).pollInterval(200, MILLISECONDS).until(() -> messageStore.dispatchBehindBytes() <= 0);
- List result = liteLifecycleManager.collectByParentTopic(parentTopic);
- Assert.assertEquals(num, result.size());
- for (String lmqName : result) {
- Assert.assertTrue(LiteUtil.belongsTo(lmqName, parentTopic));
- }
-
- result = liteLifecycleManager.collectByParentTopic(UUID.randomUUID().toString());
- Assert.assertEquals(0, result.size());
- }
-
- @Test
- public void testCollectExpiredLiteTopic() {
- int num = 3;
- String parentTopic = UUID.randomUUID().toString();
- for (int i = 0; i < num; i++) {
- messageStore.putMessage(LiteTestUtil.buildMessage(parentTopic, UUID.randomUUID().toString()));
- messageStore.putMessage(LiteTestUtil.buildMessage(UUID.randomUUID().toString(), null));
- }
- await().atMost(5, SECONDS).pollInterval(200, MILLISECONDS).until(() -> messageStore.dispatchBehindBytes() <= 0);
-
- when(liteLifecycleManager.isLiteTopicExpired(anyString(), anyString(), anyLong())).thenReturn(false);
- List> result = liteLifecycleManager.collectExpiredLiteTopic();
- Assert.assertEquals(0, result.size());
-
- when(liteLifecycleManager.isLiteTopicExpired(eq(parentTopic), anyString(), anyLong())).thenReturn(true);
- result = liteLifecycleManager.collectExpiredLiteTopic();
- Assert.assertEquals(num, result.size());
- for (Pair pair : result) {
- Assert.assertEquals(parentTopic, pair.getObject1());
- Assert.assertTrue(LiteUtil.belongsTo(pair.getObject2(), parentTopic));
- }
- }
-
@Test
public void testCleanExpiredLiteTopic() throws Exception {
int num = 3;
@@ -216,34 +168,6 @@ public void testCleanExpiredLiteTopic() throws Exception {
}
}
- @Test
- public void testCleanByParentTopic() throws Exception {
- int num = 3;
- String parentTopic = UUID.randomUUID().toString();
- mockTopicConfig.getAttributes().put(
- TopicAttributes.TOPIC_MESSAGE_TYPE_ATTRIBUTE.getName(), TopicMessageType.LITE.getValue());
- List liteTopics =
- IntStream.range(0, 3).mapToObj(i -> UUID.randomUUID().toString()).collect(Collectors.toList());
- for (int i = 0; i < num; i++) {
- messageStore.putMessage(LiteTestUtil.buildMessage(parentTopic, liteTopics.get(i)));
- }
- await().atMost(5, SECONDS).pollInterval(200, MILLISECONDS).until(() -> messageStore.dispatchBehindBytes() <= 0);
-
- for (int i = 0; i < num; i++) {
- String lmqName = LiteUtil.toLmqName(parentTopic, liteTopics.get(i));
- Assert.assertEquals(1, (long) messageStore.getQueueStore().getMaxOffset(lmqName, 0));
- Assert.assertEquals(1, liteLifecycleManager.getMaxOffsetInQueue(lmqName));
- }
-
- liteLifecycleManager.cleanByParentTopic(parentTopic);
-
- for (int i = 0; i < num; i++) {
- String lmqName = LiteUtil.toLmqName(parentTopic, liteTopics.get(i));
- Assert.assertEquals(0, (long) messageStore.getQueueStore().getMaxOffset(lmqName, 0));
- Assert.assertEquals(0, liteLifecycleManager.getMaxOffsetInQueue(lmqName));
- }
- }
-
@Test
public void testInit_combineConsumeQueueStore() throws Exception {
MessageStoreConfig storeConfig = new MessageStoreConfig();
diff --git a/broker/src/test/java/org/apache/rocketmq/broker/processor/LiteManagerProcessorTest.java b/broker/src/test/java/org/apache/rocketmq/broker/processor/LiteManagerProcessorTest.java
index 24fe1b9f7b8..b7a23390986 100644
--- a/broker/src/test/java/org/apache/rocketmq/broker/processor/LiteManagerProcessorTest.java
+++ b/broker/src/test/java/org/apache/rocketmq/broker/processor/LiteManagerProcessorTest.java
@@ -20,8 +20,10 @@
import io.netty.channel.ChannelHandlerContext;
import java.util.ArrayList;
import java.util.Collections;
+import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
+import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
@@ -31,7 +33,6 @@
import org.apache.rocketmq.broker.lite.LiteEventDispatcher;
import org.apache.rocketmq.broker.lite.LiteSharding;
import org.apache.rocketmq.broker.lite.LiteSubscriptionRegistry;
-import org.apache.rocketmq.broker.lite.SubscriberWrapper;
import org.apache.rocketmq.broker.metrics.BrokerMetricsManager;
import org.apache.rocketmq.broker.metrics.LiteConsumerLagCalculator;
import org.apache.rocketmq.broker.offset.ConsumerOffsetManager;
@@ -351,9 +352,9 @@ public void testGetLiteTopicInfo_Success() throws RemotingCommandException {
when(messageStore.getMinOffsetInQueue(lmqName, 0)).thenReturn(minOffset);
when(messageStore.getMessageStoreTimeStamp(lmqName, 0, maxOffset - 1)).thenReturn(lastUpdateTimestamp);
- SubscriberWrapper.MapWrapper wrapper = new SubscriberWrapper.MapWrapper();
- wrapper.getGroupMap().put("group", Collections.singletonList(new ClientGroup("clientId", "group")));
- when(liteSubscriptionRegistry.getAllSubscriber(null, lmqName)).thenReturn(wrapper);
+ Map> subscriberMap = new HashMap<>();
+ subscriberMap.put("group", Collections.singletonList(new ClientGroup("clientId", "group")));
+ when(liteSubscriptionRegistry.getAllSubscribers(null, lmqName)).thenReturn(subscriberMap);
when(brokerController.getBrokerConfig()).thenReturn(mock(BrokerConfig.class));
when(brokerController.getBrokerConfig().getBrokerName()).thenReturn("broker1");
when(liteSharding.shardingByLmqName("parent_topic", lmqName)).thenReturn("broker1");
@@ -478,7 +479,7 @@ public void testGetLiteClientInfo_WithSubscription() throws RemotingCommandExcep
liteTopicSet.add("lite_topic2");
LiteSubscription liteSubscription = new LiteSubscription();
- liteSubscription.setLiteTopicSet(liteTopicSet);
+ liteSubscription.setLmqSet(liteTopicSet);
when(topicConfigManager.selectTopicConfig("parent_topic")).thenReturn(topicConfig);
when(subscriptionGroupManager.findSubscriptionGroupConfig("group1")).thenReturn(groupConfig);
@@ -747,7 +748,8 @@ public void testTriggerLiteDispatch() throws Exception {
@Test
public void testGetSubscriber_null() {
String lmqName = "lmqName";
- when(liteSubscriptionRegistry.getAllSubscriber(null, lmqName)).thenReturn(new SubscriberWrapper.ListWrapper());
+ Map> emptyMap = Collections.emptyMap();
+ when(liteSubscriptionRegistry.getAllSubscribers(null, lmqName)).thenReturn(emptyMap);
Set result = processor.getSubscriber(lmqName);
assertEquals(0, result.size());
@@ -756,9 +758,9 @@ public void testGetSubscriber_null() {
@Test
public void testGetSubscriber_without_wildcard() {
String lmqName = "lmqName";
- SubscriberWrapper.MapWrapper wrapper = new SubscriberWrapper.MapWrapper();
- wrapper.getGroupMap().put("group", Collections.singletonList(new ClientGroup("clientId", "group")));
- when(liteSubscriptionRegistry.getAllSubscriber(null, lmqName)).thenReturn(wrapper);
+ Map> subscriberMap = new HashMap<>();
+ subscriberMap.put("group", Collections.singletonList(new ClientGroup("clientId", "group")));
+ when(liteSubscriptionRegistry.getAllSubscribers(null, lmqName)).thenReturn(subscriberMap);
Set result = processor.getSubscriber(lmqName);
assertEquals(1, result.size());
@@ -768,13 +770,13 @@ public void testGetSubscriber_without_wildcard() {
@Test
public void testGetSubscriber_with_wildcard() {
String lmqName = "lmqName";
- SubscriberWrapper.MapWrapper wrapper = new SubscriberWrapper.MapWrapper();
- wrapper.getGroupMap().put("group", Collections.singletonList(new ClientGroup("clientId", "group")));
- wrapper.getGroupMap().put("wildcardGroup", Collections.singletonList(new ClientGroup("clientId", "wildcardGroup")));
+ Map> subscriberMap = new HashMap<>();
+ subscriberMap.put("group", Collections.singletonList(new ClientGroup("clientId", "group")));
+ subscriberMap.put("wildcardGroup", Collections.singletonList(new ClientGroup("clientId", "wildcardGroup")));
SubscriptionGroupConfig groupConfig = new SubscriptionGroupConfig();
groupConfig.getAttributes().put(LITE_SUB_WILDCARD_ATTRIBUTE.getName(), "xxx");
- when(liteSubscriptionRegistry.getAllSubscriber(null, lmqName)).thenReturn(wrapper);
+ when(liteSubscriptionRegistry.getAllSubscribers(null, lmqName)).thenReturn(subscriberMap);
when(subscriptionGroupManager.findSubscriptionGroupConfig("wildcardGroup")).thenReturn(groupConfig);
Set result = processor.getSubscriber(lmqName);
diff --git a/common/src/main/java/org/apache/rocketmq/common/lite/LiteSubscription.java b/common/src/main/java/org/apache/rocketmq/common/lite/LiteSubscription.java
index abf7c9ee3af..e9d312afcc1 100644
--- a/common/src/main/java/org/apache/rocketmq/common/lite/LiteSubscription.java
+++ b/common/src/main/java/org/apache/rocketmq/common/lite/LiteSubscription.java
@@ -17,34 +17,28 @@
package org.apache.rocketmq.common.lite;
-import java.util.Collection;
+import java.util.Collections;
+import java.util.HashSet;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
public class LiteSubscription {
private String group;
private String topic;
- private final Set liteTopicSet = ConcurrentHashMap.newKeySet();
+ private final Set lmqSet = ConcurrentHashMap.newKeySet();
private volatile long updateTime = System.currentTimeMillis();
- public boolean addLiteTopic(String liteTopic) {
- updateTime();
- return this.liteTopicSet.add(liteTopic);
- }
-
- public void addLiteTopic(Collection set) {
- updateTime();
- this.liteTopicSet.addAll(set);
+ public LiteSubscription touch() {
+ this.updateTime = System.currentTimeMillis();
+ return this;
}
- public boolean removeLiteTopic(String liteTopic) {
- updateTime();
- return this.liteTopicSet.remove(liteTopic);
+ public boolean addLmq(String lmqName) {
+ return this.lmqSet.add(lmqName);
}
- public void removeLiteTopic(Collection set) {
- updateTime();
- this.liteTopicSet.removeAll(set);
+ public boolean removeLmq(String lmqName) {
+ return this.lmqSet.remove(lmqName);
}
public String getGroup() {
@@ -65,12 +59,13 @@ public LiteSubscription setTopic(String topic) {
return this;
}
- public Set getLiteTopicSet() {
- return liteTopicSet;
+ public Set getLmqSet() {
+ return lmqSet;
}
- public LiteSubscription setLiteTopicSet(Set liteTopicSet) {
- this.liteTopicSet.addAll(liteTopicSet);
+ public LiteSubscription setLmqSet(Set lmqSet) {
+ this.lmqSet.clear();
+ this.lmqSet.addAll(lmqSet);
return this;
}
@@ -82,8 +77,15 @@ public void setUpdateTime(long updateTime) {
this.updateTime = updateTime;
}
- private void updateTime() {
- this.updateTime = System.currentTimeMillis();
+ public static Set removals(Set current, Set target) {
+ Set safeTarget = target == null ? Collections.emptySet() : target;
+ Set result = new HashSet<>();
+ for (String item : current) {
+ if (!safeTarget.contains(item)) {
+ result.add(item);
+ }
+ }
+ return result;
}
@Override
@@ -91,8 +93,9 @@ public String toString() {
return "LiteSubscription{" +
"group='" + group + '\'' +
", topic='" + topic + '\'' +
- ", liteTopicSet=" + liteTopicSet +
+ ", lmqSet=" + lmqSet +
", updateTime=" + updateTime +
'}';
}
+
}
diff --git a/common/src/test/java/org/apache/rocketmq/common/lite/LiteSubscriptionTest.java b/common/src/test/java/org/apache/rocketmq/common/lite/LiteSubscriptionTest.java
new file mode 100644
index 00000000000..4f88f6c62fe
--- /dev/null
+++ b/common/src/test/java/org/apache/rocketmq/common/lite/LiteSubscriptionTest.java
@@ -0,0 +1,172 @@
+/*
+ * 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.lite;
+
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.Set;
+import org.junit.Before;
+import org.junit.Test;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertSame;
+import static org.junit.Assert.assertTrue;
+
+public class LiteSubscriptionTest {
+
+ private LiteSubscription subscription;
+
+ @Before
+ public void setUp() {
+ subscription = new LiteSubscription();
+ }
+
+ // ========== Group 1: chainable setters ==========
+
+ @Test
+ public void setGroup_returnsThis() {
+ LiteSubscription result = subscription.setGroup("testGroup");
+ assertSame(subscription, result);
+ assertEquals("testGroup", subscription.getGroup());
+ }
+
+ @Test
+ public void setTopic_returnsThis() {
+ LiteSubscription result = subscription.setTopic("testTopic");
+ assertSame(subscription, result);
+ assertEquals("testTopic", subscription.getTopic());
+ }
+
+ @Test
+ public void setLmqSet_returnsThis() {
+ Set newSet = new HashSet<>();
+ newSet.add("lmq1");
+ LiteSubscription result = subscription.setLmqSet(newSet);
+ assertSame(subscription, result);
+ assertTrue(subscription.getLmqSet().contains("lmq1"));
+ }
+
+ // ========== Group 2: touch ==========
+
+ @Test
+ public void touch_updatesTimeAndReturnsThis() throws InterruptedException {
+ long before = subscription.getUpdateTime();
+ Thread.sleep(10);
+ LiteSubscription result = subscription.touch();
+ assertSame(subscription, result);
+ assertTrue(subscription.getUpdateTime() > before);
+ }
+
+ // ========== Group 3: lmqSet add/remove ==========
+
+ @Test
+ public void addLmq_newElement_returnsTrue() {
+ assertTrue(subscription.addLmq("lmq1"));
+ assertEquals(1, subscription.getLmqSet().size());
+ }
+
+ @Test
+ public void addLmq_duplicate_returnsFalse() {
+ subscription.addLmq("lmq1");
+ assertFalse(subscription.addLmq("lmq1"));
+ assertEquals(1, subscription.getLmqSet().size());
+ }
+
+ @Test
+ public void removeLmq_existing_returnsTrue() {
+ subscription.addLmq("lmq1");
+ assertTrue(subscription.removeLmq("lmq1"));
+ assertTrue(subscription.getLmqSet().isEmpty());
+ }
+
+ @Test
+ public void removeLmq_absent_returnsFalse() {
+ assertFalse(subscription.removeLmq("nonexistent"));
+ }
+
+ // ========== Group 4: setLmqSet replacement semantics ==========
+
+ @Test
+ public void setLmqSet_clearsOldAndAddsNew() {
+ subscription.addLmq("old1");
+ subscription.addLmq("old2");
+
+ Set newSet = new HashSet<>();
+ newSet.add("new1");
+ subscription.setLmqSet(newSet);
+
+ assertEquals(1, subscription.getLmqSet().size());
+ assertTrue(subscription.getLmqSet().contains("new1"));
+ assertFalse(subscription.getLmqSet().contains("old1"));
+ }
+
+ // ========== Group 5: removals static utility ==========
+
+ @Test
+ public void removals_normalDiff() {
+ Set current = new HashSet<>();
+ current.add("a");
+ current.add("b");
+ current.add("c");
+ Set target = new HashSet<>();
+ target.add("b");
+
+ Set result = LiteSubscription.removals(current, target);
+ assertEquals(2, result.size());
+ assertTrue(result.contains("a"));
+ assertTrue(result.contains("c"));
+ }
+
+ @Test
+ public void removals_targetNull() {
+ Set current = new HashSet<>();
+ current.add("a");
+ current.add("b");
+
+ Set result = LiteSubscription.removals(current, null);
+ assertEquals(2, result.size());
+ assertTrue(result.contains("a"));
+ assertTrue(result.contains("b"));
+ }
+
+ @Test
+ public void removals_noDiff() {
+ Set current = new HashSet<>();
+ current.add("a");
+ Set target = new HashSet<>();
+ target.add("a");
+ target.add("b");
+
+ Set result = LiteSubscription.removals(current, target);
+ assertTrue(result.isEmpty());
+ }
+
+ @Test
+ public void removals_emptyCurrent() {
+ Set result = LiteSubscription.removals(Collections.emptySet(), Collections.singleton("a"));
+ assertTrue(result.isEmpty());
+ }
+
+ // ========== Group 6: thread safety type check ==========
+
+ @Test
+ public void lmqSet_isConcurrentSafe() {
+ assertTrue(subscription.getLmqSet().getClass().getName().contains("ConcurrentHashMap"));
+ }
+}
diff --git a/pom.xml b/pom.xml
index 645ad51225a..f4f2ed23e74 100644
--- a/pom.xml
+++ b/pom.xml
@@ -119,6 +119,7 @@
2.0.3
1.0.0
1.10.0
+ 4.5.0
1.5.2-2
1.10.3
0.33.0
@@ -784,6 +785,11 @@
commons-validator
${commons-validator.version}