opensource/rocketmq
3
0
Fork 0
mirror of https://github.com/apache/rocketmq.git synced 2026-10-24 02:15:19 +08:00
Code Issues Packages Projects Releases Wiki Activity

[ISSUE #10929] Add LmqPrefixIndex to accelerate wildcard dispatch and refine lite subscription model (#10930)

Browse Source 
* [ISSUE #10929] Add LmqPrefixIndex to accelerate wildcard dispatch and refine lite subscription model

- Add LmqPrefixIndex, a prefix-ordered in-memory index over lmq names, bootstrapped at startup and maintained via onLmqCreate/onLmqDelete hooks
- Rewrite wildcard full dispatch, getLiteTopicCount, collectByParentTopic and cleanByParentTopic on index-backed forEachLiteTopicByPrefix/ByParent, turning O(total lmqs) scans into O(matched lmqs)
- Refine lite subscription model: replace SubscriberWrapper with Map-based getAllSubscribers, simplify getWildcardGroupClients, rename getLiteTopicSet to getLmqSet, make LiteCtlListener callbacks default methods

* [ISSUE #10929] Retrigger CI
This commit is contained in:
Quan
2026-09-08 15:38:33 +08:00
committed by GitHub
parent 66362b9c9f
commit 1e29112853
24 changed files with 1348 additions and 596 deletions
Download Patch File Download Diff File Expand all files Collapse all files
WORKSPACE
+1
View File
@@ -60,6 +60,7 @@ maven_install(
"io.opentracing:opentracing-api:0.33.0",
"io.opentracing:opentracing-mock:0.33.0",
"commons-collections:commons-collections:3.2.2",
"org.apache.commons:commons-collections4:4.5.0",
"org.awaitility:awaitility:4.1.0",
"commons-cli:commons-cli:1.5.0",
"com.google.guava:guava:32.0.1-jre",
broker/BUILD.bazel
+2
View File
@@ -37,6 +37,7 @@ java_library(
"@maven//:com_googlecode_concurrentlinkedhashmap_concurrentlinkedhashmap_lru",
"@maven//:commons_cli_commons_cli",
"@maven//:commons_collections_commons_collections",
"@maven//:org_apache_commons_commons_collections4",
"@maven//:commons_io_commons_io",
"@maven//:commons_validator_commons_validator",
"@maven//:io_netty_netty_all",
@@ -95,6 +96,7 @@ java_library(
"@maven//:com_googlecode_concurrentlinkedhashmap_concurrentlinkedhashmap_lru",
"@maven//:org_rocksdb_rocksdbjni",
"@maven//:commons_collections_commons_collections",
"@maven//:org_apache_commons_commons_collections4",
"@maven//:org_junit_jupiter_junit_jupiter_api",
"@maven//:com_github_ben_manes_caffeine_caffeine",
],
broker/pom.xml
+4
View File
@@ -66,6 +66,10 @@
<groupId>commons-io</groupId>
<artifactId>commons-io</artifactId>
</dependency>
<dependency>
<groupId>org.apache.commons</groupId>
<artifactId>commons-collections4</artifactId>
</dependency>
<dependency>
<groupId>org.javassist</groupId>
<artifactId>javassist</artifactId>
broker/src/main/java/org/apache/rocketmq/broker/BrokerController.java
+5 -1
View File
@@ -1155,7 +1155,11 @@ public class BrokerController {
private boolean initLiteService() {
this.liteEventDispatcher.init();
return this.liteLifecycleManager.init();
if (!this.liteLifecycleManager.init()) {
return false;
}
this.liteLifecycleManager.bootstrapLmqPrefixIndex();
return true;
}
public void registerProcessor() {
broker/src/main/java/org/apache/rocketmq/broker/lite/AbstractLiteLifecycleManager.java
+108 -16
View File
@@ -18,10 +18,10 @@
package org.apache.rocketmq.broker.lite;
import com.google.common.collect.Sets;
import org.apache.commons.lang3.StringUtils;
import org.apache.commons.lang3.tuple.Triple;
import org.apache.rocketmq.broker.BrokerController;
import org.apache.rocketmq.common.MixAll;
import org.apache.rocketmq.common.Pair;
import org.apache.rocketmq.common.ServiceThread;
import org.apache.rocketmq.common.constant.LoggerName;
import org.apache.rocketmq.common.lite.LiteUtil;
@@ -29,6 +29,7 @@ import org.apache.rocketmq.logging.org.slf4j.Logger;
import org.apache.rocketmq.logging.org.slf4j.LoggerFactory;
import org.apache.rocketmq.store.MessageStore;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.Map;
@@ -56,6 +57,12 @@ public abstract class AbstractLiteLifecycleManager extends ServiceThread {
protected Map<String, Integer> offsetInvalidScanCountMap = new ConcurrentHashMap<>();
protected Map<String, Integer> storeTimeInvalidScanCountMap = new ConcurrentHashMap<>();
/**
* Global prefix index over lmqName, maintained on the lmq lifecycle hot path and
* consumed by {@link LiteEventDispatcher} wildcard full dispatch and lifecycle queries.
*/
protected final LmqPrefixIndex lmqPrefixIndex = new LmqPrefixIndex();
public AbstractLiteLifecycleManager(BrokerController brokerController, LiteSharding liteSharding) {
this.brokerController = brokerController;
this.brokerName = brokerController.getBrokerConfig().getBrokerName();
@@ -68,22 +75,53 @@ public abstract class AbstractLiteLifecycleManager extends ServiceThread {
return true;
}
/**
* Populate the prefix index once at startup. Must be called after {@link #init()}.
*/
public void bootstrapLmqPrefixIndex() {
long start = System.currentTimeMillis();
forEachLiteTopic(triple -> {
lmqPrefixIndex.add(triple.getLeft());
return true;
});
LOGGER.info("bootstrap lmq prefix index finish, indexed:{}, costMs:{}",
lmqPrefixIndex.size(), System.currentTimeMillis() - start);
}
/**
* Hook fired on the first message of a freshly created lmq.
*/
public void onLmqCreate(String lmqName) {
lmqPrefixIndex.add(lmqName);
}
/**
* Hook fired when an lmq is deleted.
*/
public void onLmqDelete(String lmqName) {
lmqPrefixIndex.remove(lmqName);
}
/**
* This method actually returns NEXT slot index to use, starting from 0
*/
public abstract long getMaxOffsetInQueue(String lmqName);
/**
* Collect expired LMQ of lite topic, and also attach its parent topic name
* return Pair of parent topic and lmq name, not null
*/
public abstract List<Pair<String, String>> collectExpiredLiteTopic();
/**
* Collect LMQ by parent topic
* return lmq name list, not null
*/
public abstract List<String> collectByParentTopic(String parentTopic);
public List<String> collectByParentTopic(String parentTopic) {
if (StringUtils.isEmpty(parentTopic)) {
return Collections.emptyList();
}
List<String> resultList = new ArrayList<>();
forEachLiteTopicByParent(parentTopic, triple -> {
resultList.add(triple.getLeft());
return true;
});
return resultList;
}
/**
* Iterator of lite topic, for high frequency iteration
@@ -94,6 +132,36 @@ public abstract class AbstractLiteLifecycleManager extends ServiceThread {
*/
public abstract void forEachLiteTopic(Function<Triple<String, Long, Long>, Boolean> function);
/**
* Delegate to {@link #forEachLiteTopicByPrefix} with prefix = LITE_TOPIC_PREFIX + parentTopic + SEPARATOR.
*
* @param parentTopic parent topic to filter by
* @param function consumer func; caller must NOT add/remove lmqPrefixIndex inside the callback
*/
public void forEachLiteTopicByParent(String parentTopic, Function<Triple<String, Long, Long>, Boolean> function) {
forEachLiteTopicByPrefix(LiteUtil.LITE_TOPIC_PREFIX + parentTopic + LiteUtil.SEPARATOR, function);
}
/**
* Iterator of lite topic filtered by lmqName prefix.
* Triple<lmqName, maxOffsetInQueue, lastStoreTimestamp>, lastStoreTimestamp is null for now.
* Entries with maxOffset <= 0 (no messages ever written) are skipped and will NOT be applied.
* Return true to continue, false to break.
*
* @param prefix lmqName prefix to filter by
* @param function consumer func; caller must NOT add/remove lmqPrefixIndex inside the callback
*/
public void forEachLiteTopicByPrefix(String prefix, Function<Triple<String, Long, Long>, Boolean> function) {
lmqPrefixIndex.forEachLmqByPrefix(prefix, lmqName -> {
long maxOffset = getMaxOffsetInQueue(lmqName);
if (maxOffset <= 0) {
return true;
}
Triple<String, Long, Long> triple = Triple.of(lmqName, maxOffset, null);
return function.apply(triple);
});
}
/**
* Check if the subscription for the given LMQ is active.
* A subscription is considered active if either:
@@ -108,7 +176,12 @@ public abstract class AbstractLiteLifecycleManager extends ServiceThread {
if (!LiteMetadataUtil.isLiteMessageType(parentTopic, brokerController)) {
return 0;
}
return collectByParentTopic(parentTopic).size();
int[] count = {0};
forEachLiteTopicByParent(parentTopic, triple -> {
count[0]++;
return true;
});
return count[0];
}
public boolean isLmqExist(String lmqName) {
@@ -117,11 +190,23 @@ public abstract class AbstractLiteLifecycleManager extends ServiceThread {
public void cleanExpiredLiteTopic() {
try {
long startMs = System.currentTimeMillis();
updateMetadata(); // necessary
List<Pair<String, String>> lmqToDelete = collectExpiredLiteTopic();
LOGGER.info("collect expired topic, size:{}", lmqToDelete.size());
lmqToDelete.forEach(pair -> deleteLmq(pair.getObject1(), pair.getObject2()));
if (!lmqToDelete.isEmpty()) {
int[] count = {0};
forEachLiteTopic(triple -> {
String lmqName = triple.getLeft();
String parentTopic = LiteUtil.getParentTopic(lmqName);
if (parentTopic == null) {
return true;
}
if (isLiteTopicExpired(parentTopic, lmqName, triple.getMiddle())) {
deleteLmq(parentTopic, lmqName);
count[0]++;
}
return true;
});
LOGGER.info("clean expired topic, size:{}, cost:{}ms", count[0], System.currentTimeMillis() - startMs);
if (count[0] > 0) {
brokerController.getMessageStore().getQueueStore().flush();
}
} catch (Exception e) {
@@ -134,10 +219,16 @@ public abstract class AbstractLiteLifecycleManager extends ServiceThread {
if (!LiteMetadataUtil.isLiteMessageType(parentTopic, brokerController)) {
return;
}
long startMs = System.currentTimeMillis();
updateMetadata(); // necessary
List<String> lmqToDelete = collectByParentTopic(parentTopic);
LOGGER.info("clean by parent topic, {}, size:{}", parentTopic, lmqToDelete.size());
lmqToDelete.forEach(lmqName -> deleteLmq(parentTopic, lmqName));
// collect-then-delete: forEachLiteTopicByParent and deleteLmq each hold a lock, nesting causes deadlock
List<String> toDelete = new ArrayList<>();
forEachLiteTopicByParent(parentTopic, triple -> {
toDelete.add(triple.getLeft());
return true;
});
toDelete.forEach(liteTopic -> deleteLmq(parentTopic, liteTopic));
LOGGER.info("clean by parent topic:{}, size:{}, cost:{}ms", parentTopic, toDelete.size(), System.currentTimeMillis() - startMs);
} catch (Exception e) {
LOGGER.error("cleanByParentTopic error", e);
}
@@ -241,6 +332,7 @@ public abstract class AbstractLiteLifecycleManager extends ServiceThread {
brokerController.getConsumerOffsetManager().getPullOffsetTable().remove(
lmqName + TOPIC_GROUP_SEPARATOR + MixAll.TOOLS_CONSUMER_GROUP);
removeInvalidCount(lmqName);
onLmqDelete(lmqName);
LOGGER.info("delete lmq finish. {}, sharding:{}", lmqName, sharding);
} catch (Exception e) {
LOGGER.error("delete lmq error. {}", lmqName, e);
broker/src/main/java/org/apache/rocketmq/broker/lite/LiteCtlListener.java
+6 -3
View File
@@ -19,10 +19,13 @@ package org.apache.rocketmq.broker.lite;
public interface LiteCtlListener {
void onRegister(String clientId, String group, String lmqName);
default void onRegister(String clientId, String group, String lmqName) {
}
void onUnregister(String clientId, String group, String lmqName);
default void onUnregister(String clientId, String group, String lmqName) {
}
void onRemoveAll(String clientId, String group);
default void onRemoveAll(String clientId, String group) {
}
}
broker/src/main/java/org/apache/rocketmq/broker/lite/LiteEventDispatcher.java
+31 -44
View File
@@ -21,7 +21,6 @@ import com.google.common.annotations.VisibleForTesting;
import com.google.common.cache.Cache;
import com.google.common.cache.CacheBuilder;
import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.lang3.tuple.Triple;
import org.apache.rocketmq.broker.BrokerController;
import org.apache.rocketmq.common.BrokerConfig;
import org.apache.rocketmq.common.ServiceThread;
@@ -45,8 +44,6 @@ import java.util.concurrent.ConcurrentSkipListSet;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadLocalRandom;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.Function;
public class LiteEventDispatcher extends ServiceThread {
@@ -93,21 +90,17 @@ public class LiteEventDispatcher extends ServiceThread {
if (queueId != 0 || !LiteUtil.isLiteTopicQueue(lmqName)) {
return;
}
// Maintain prefix index only on the lmq's first message; pre-existing lmqs are
// populated once at startup during init().
if (offset == 0) {
liteLifecycleManager.onLmqCreate(lmqName);
}
doDispatch(group, lmqName, null);
}
protected void doDispatch(String group, String lmqName, String excludeClientId) {
SubscriberWrapper wrapper = liteSubscriptionRegistry.getAllSubscriber(group, lmqName);
if (null == wrapper) {
return;
}
if (wrapper instanceof SubscriberWrapper.ListWrapper) {
selectAndDispatch(lmqName, wrapper.asListWrapper().getClients(), excludeClientId);
}
if (wrapper instanceof SubscriberWrapper.MapWrapper) {
Map<String, List<ClientGroup>> map = wrapper.asMapWrapper().getGroupMap();
map.forEach((key, value) -> selectAndDispatch(lmqName, value, excludeClientId));
}
Map<String, List<ClientGroup>> subscriberMap = liteSubscriptionRegistry.getAllSubscribers(group, lmqName);
subscriberMap.values().forEach(clients -> selectAndDispatch(lmqName, clients, excludeClientId));
}
/**
@@ -201,7 +194,7 @@ public class LiteEventDispatcher extends ServiceThread {
*/
public void doFullDispatchForClient(String clientId, String group) {
LiteSubscription subscription = liteSubscriptionRegistry.getLiteSubscription(clientId);
if (null == subscription || CollectionUtils.isEmpty(subscription.getLiteTopicSet())) {
if (null == subscription || CollectionUtils.isEmpty(subscription.getLmqSet())) {
LOGGER.info("client full dispatch, but no subscription. {}", clientId);
return;
}
@@ -219,15 +212,11 @@ public class LiteEventDispatcher extends ServiceThread {
+ (isActiveConsuming ? 0 : random.nextInt(10 * 1000)));
return;
}
LOGGER.info("client full dispatch, {}, total:{}", clientId, subscription.getLiteTopicSet().size());
LOGGER.info("client full dispatch, {}, total:{}", clientId, subscription.getLmqSet().size());
int count = 0;
for (String lmqName : subscription.getLiteTopicSet()) {
for (String lmqName : subscription.getLmqSet()) {
long maxOffset = liteLifecycleManager.getMaxOffsetInQueue(lmqName);
if (maxOffset <= 0) {
continue;
}
long consumerOffset = brokerController.getConsumerOffsetManager().queryOffset(group, lmqName, 0);
if (consumerOffset >= maxOffset) {
if (isFullyConsumed(group, lmqName, maxOffset)) {
continue;
}
if (eventSet.offer(lmqName)) {
@@ -274,34 +263,36 @@ public class LiteEventDispatcher extends ServiceThread {
if (null == parentTopic || !LiteMetadataUtil.isWildcardGroup(group, brokerController)) {
return;
}
List<ClientGroup> clients = liteSubscriptionRegistry.getWildcardSubscriber(group, parentTopic).getClients();
List<ClientGroup> clients = liteSubscriptionRegistry.getWildcardGroupClients(group);
if (CollectionUtils.isEmpty(clients)) {
return;
}
AtomicInteger count = new AtomicInteger();
Function<Triple<String, Long, Long>, Boolean> function = triple -> {
int[] count = {0};
liteLifecycleManager.forEachLiteTopicByParent(parentTopic, triple -> {
String lmqName = triple.getLeft();
long maxOffset = triple.getMiddle();
if (!LiteUtil.belongsTo(lmqName, parentTopic)) {
return true;
}
if (maxOffset <= 0) {
return true;
}
long consumerOffset = brokerController.getConsumerOffsetManager().queryOffset(group, lmqName, 0);
if (consumerOffset >= maxOffset) {
if (isFullyConsumed(group, lmqName, maxOffset)) {
return true;
}
if (selectAndDispatch(lmqName, clients, null)) {
count.incrementAndGet();
} else {
LOGGER.warn("doFullDispatchForWildcardGroup, wait another period. {}", group);
return false;
count[0]++;
return true;
}
LOGGER.warn("doFullDispatchForWildcardGroup, wait another period. {}", group);
return false;
});
LOGGER.info("doFullDispatchForWildcardGroup finish. {}, dispatch:{}", group, count[0]);
}
/**
* Returns true if all messages of the lmq have been consumed for the given group.
*/
private boolean isFullyConsumed(String group, String lmqName, long maxOffset) {
if (maxOffset <= 0) {
return true;
};
liteLifecycleManager.forEachLiteTopic(function);
LOGGER.info("doFullDispatchForWildcardGroup finish. {}, dispatch:{}", group, count);
}
long consumerOffset = brokerController.getConsumerOffsetManager().queryOffset(group, lmqName, 0);
return consumerOffset >= maxOffset;
}
/**
@@ -522,10 +513,6 @@ public class LiteEventDispatcher extends ServiceThread {
}
}
@Override
public void onUnregister(String clientId, String group, String lmqName) {
}
/**
* Mostly triggered when client channel closed, ensure that lite subscriptions is cleared before.
*/
broker/src/main/java/org/apache/rocketmq/broker/lite/LiteLifecycleManager.java
-46
View File
@@ -17,20 +17,15 @@
package org.apache.rocketmq.broker.lite;
import org.apache.commons.lang3.StringUtils;
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;
import org.apache.rocketmq.logging.org.slf4j.LoggerFactory;
import org.apache.rocketmq.store.queue.ConsumeQueueInterface;
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;
@@ -48,47 +43,6 @@ public class LiteLifecycleManager extends AbstractLiteLifecycleManager {
return consumeQueue != null ? consumeQueue.getMaxOffsetInQueue() : 0L;
}
@Override
public List<String> collectByParentTopic(String parentTopic) {
if (StringUtils.isEmpty(parentTopic)) {
return Collections.emptyList();
}
List<String> resultList = new ArrayList<>();
Iterator<Map.Entry<String, ConcurrentMap<Integer, ConsumeQueueInterface>>> iterator =
messageStore.getQueueStore().getConsumeQueueTable().entrySet().iterator();
while (iterator.hasNext()) {
Map.Entry<String, ConcurrentMap<Integer, ConsumeQueueInterface>> entry = iterator.next();
if (LiteUtil.belongsTo(entry.getKey(), parentTopic)) {
resultList.add(entry.getKey());
}
}
return resultList;
}
@Override
public List<Pair<String, String>> collectExpiredLiteTopic() {
List<Pair<String, String>> lmqToDelete = new ArrayList<>();
Iterator<Map.Entry<String, ConcurrentMap<Integer, ConsumeQueueInterface>>> iterator =
messageStore.getQueueStore().getConsumeQueueTable().entrySet().iterator();
while (iterator.hasNext()) {
Map.Entry<String, ConcurrentMap<Integer, ConsumeQueueInterface>> entry = iterator.next();
String lmqName = entry.getKey();
String parentTopic = LiteUtil.getParentTopic(lmqName);
if (null == parentTopic) {
continue;
}
Map<Integer, ConsumeQueueInterface> map = entry.getValue();
if (map.size() != 1 || null == map.get(0)) {
LOGGER.warn("unexpected lmq count. {}", lmqName);
continue;
}
if (isLiteTopicExpired(parentTopic, entry.getKey(), map.get(0).getMaxOffsetInQueue())) {
lmqToDelete.add(new Pair<>(parentTopic, lmqName));
}
}
return lmqToDelete;
}
@Override
public void forEachLiteTopic(Function<Triple<String, Long, Long>, Boolean> function) {
Iterator<Map.Entry<String, ConcurrentMap<Integer, ConsumeQueueInterface>>> iterator =
broker/src/main/java/org/apache/rocketmq/broker/lite/LiteSubscriptionRegistry.java
+4 -2
View File
@@ -20,7 +20,9 @@ package org.apache.rocketmq.broker.lite;
import io.netty.channel.Channel;
import java.util.List;
import java.util.Map;
import java.util.Set;
import org.apache.rocketmq.common.entity.ClientGroup;
import org.apache.rocketmq.common.lite.LiteSubscription;
import org.apache.rocketmq.common.lite.OffsetOption;
@@ -42,9 +44,9 @@ public interface LiteSubscriptionRegistry {
void addListener(LiteCtlListener listener);
SubscriberWrapper getAllSubscriber(String group, String lmqName);
Map<String, List<ClientGroup>> getAllSubscribers(String group, String lmqName);
SubscriberWrapper.ListWrapper getWildcardSubscriber(String group, String parentTopic);
List<ClientGroup> getWildcardGroupClients(String group);
List<String> getAllClientIdByGroup(String group);
broker/src/main/java/org/apache/rocketmq/broker/lite/LiteSubscriptionRegistryImpl.java
+111 -102
View File
@@ -23,6 +23,7 @@ import com.google.common.cache.CacheBuilder;
import io.netty.channel.Channel;
import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Objects;
@@ -52,7 +53,7 @@ public class LiteSubscriptionRegistryImpl extends ServiceThread implements LiteS
protected final ConcurrentMap<String/*clientId*/, Channel> clientChannels = new ConcurrentHashMap<>();
protected final ConcurrentMap<String/*clientId*/, LiteSubscription> client2Subscription = new ConcurrentHashMap<>();
protected final ConcurrentMap<String/*lmqName*/, Set<ClientGroup>> liteTopic2Group = new ConcurrentHashMap<>();
protected final ConcurrentMap<String/*lmqName*/, Set<ClientGroup>> liteTopic2ClientGroup = new ConcurrentHashMap<>();
protected final ConcurrentMap<String/*topic*/, Set<String/*group*/>> wildcardGroupMap = new ConcurrentHashMap<>();
private final Cache<String/*group*/, List<ClientGroup>> wildcardClientCache =
CacheBuilder.newBuilder().maximumSize(2000).expireAfterWrite(30, TimeUnit.SECONDS).build();
@@ -90,14 +91,15 @@ public class LiteSubscriptionRegistryImpl extends ServiceThread implements LiteS
throw new IllegalStateException("subscribe lite operation is not supported for this group");
}
LiteSubscription thisSub = getOrCreateLiteSubscription(clientId, group, topic);
// 1) Normalize input: keep active lmqs only
LiteSubscription thisSub = getOrCreateLiteSubscription(clientId, group, topic).touch();
// Utilize existing string object
final ClientGroup clientGroup = new ClientGroup(clientId, thisSub.getGroup());
for (String lmqName : lmqNameSet) {
if (!liteLifecycleManager.isSubscriptionActive(topic, lmqName)) {
continue;
}
thisSub.addLiteTopic(lmqName);
// First remove the old subscription
if (LiteMetadataUtil.isSubLiteExclusive(group, brokerController)) {
excludeClientByLmqName(clientId, group, lmqName);
@@ -107,63 +109,61 @@ public class LiteSubscriptionRegistryImpl extends ServiceThread implements LiteS
exclusiveEvictionTombstones.remove(clientId, lmqName);
}
resetOffset(lmqName, group, clientId, offsetOption);
addTopicGroup(clientGroup, lmqName);
addClientGroup(thisSub, clientGroup, lmqName);
}
}
@Override
public void removePartialSubscription(String clientId, String group, String topic, Set<String> lmqNameSet) {
LiteSubscription thisSub = getOrCreateLiteSubscription(clientId, group, topic);
LiteSubscription thisSub = getOrCreateLiteSubscription(clientId, group, topic).touch();
ClientGroup clientGroup = new ClientGroup(clientId, thisSub.getGroup());
boolean isResetOffsetOnUnsubscribe = LiteMetadataUtil.isResetOffsetOnUnsubscribe(group, brokerController);
for (String lmqName : lmqNameSet) {
thisSub.removeLiteTopic(lmqName);
removeTopicGroup(clientGroup, lmqName, isResetOffsetOnUnsubscribe);
thisSub.removeLmq(lmqName);
removeClientGroup(clientGroup, lmqName, isResetOffsetOnUnsubscribe);
}
}
@Override
public void addCompleteSubscription(String clientId, String group, String topic, Set<String> lmqNameAll, long version) {
Set<String> lmqNameNew;
if (LiteMetadataUtil.isWildcardGroup(group, brokerController)) {
lmqNameNew = Collections.singleton(mockLmqNameForWildcardGroup(topic, group));
markWildcardGroup(topic, group);
} else {
lmqNameNew = lmqNameAll.stream()
.filter(lmqName -> liteLifecycleManager.isSubscriptionActive(topic, lmqName))
.collect(Collectors.toSet());
String wildcardLmq = mockLmqNameForWildcardGroup(topic, group);
LiteSubscription wildcardSub = getOrCreateLiteSubscription(clientId, group, topic).touch();
addClientGroup(wildcardSub, new ClientGroup(clientId, group), wildcardLmq);
return;
}
LiteSubscription thisSub = getOrCreateLiteSubscription(clientId, group, topic);
Set<String> lmqNamePrev = thisSub.getLiteTopicSet();
// Find topics to remove (in current set but not in new set)
Set<String> lmqNameRemove = lmqNamePrev.stream()
.filter(lmqName -> !lmqNameNew.contains(lmqName))
// 1) Normalize input: keep active lmqs only
Set<String> targetLmqs = lmqNameAll.stream()
.filter(lmqName -> liteLifecycleManager.isSubscriptionActive(topic, lmqName))
.collect(Collectors.toSet());
// 2) Compute removal delta
LiteSubscription thisSub = getOrCreateLiteSubscription(clientId, group, topic).touch();
Set<String> lmqsToRemove = LiteSubscription.removals(thisSub.getLmqSet(), targetLmqs);
// 3) Apply removals, then idempotent adds over the full target set
ClientGroup clientGroup = new ClientGroup(clientId, thisSub.getGroup());
lmqNameRemove.forEach(lmqName -> {
thisSub.removeLiteTopic(lmqName);
removeTopicGroup(clientGroup, lmqName, false);
});
lmqNameNew.forEach(lmqName -> {
thisSub.addLiteTopic(lmqName);
addTopicGroup(clientGroup, lmqName);
lmqsToRemove.forEach(lmqName -> {
thisSub.removeLmq(lmqName);
removeClientGroup(clientGroup, lmqName, false);
});
targetLmqs.forEach(lmqName -> addClientGroup(thisSub, clientGroup, lmqName));
// Tombstone operations only apply to exclusive groups.
if (LiteMetadataUtil.isSubLiteExclusive(group, brokerController)) {
// Boundary case: if any lmqName in the client's reported full subscription still has
// a tombstone, the previous notifyUnsubscribeLite was likely lost. Re-send the
// unsubscribe notification to drive the client's local state to converge.
lmqNameNew.stream()
// If any lmqName in the client's full subscription still has a tombstone,
// the previous notifyUnsubscribeLite was likely lost. Re-send to drive convergence.
targetLmqs.stream()
.filter(lmqName -> exclusiveEvictionTombstones.contains(clientId, lmqName))
.forEach(lmqName -> {
LOGGER.info("re-notify unsubscribe for tombstoned lmqName, clientId:{}, group:{}, lmqName:{}",
clientId, group, lmqName);
notifyUnsubscribeLite(clientId, group, lmqName);
});
// Clean exclusive-eviction tombstones for liteTopics no longer in the client's full subscription set
exclusiveEvictionTombstones.removeStale(clientId, lmqNameNew);
// Clean tombstones for lmqNames no longer in the client's full subscription set
exclusiveEvictionTombstones.removeStale(clientId, targetLmqs);
}
}
@@ -180,9 +180,7 @@ public class LiteSubscriptionRegistryImpl extends ServiceThread implements LiteS
}
LOGGER.info("removeCompleteSubscription, topic:{}, group:{}, clientId:{}", thisSub.getTopic(), thisSub.getGroup(), clientId);
ClientGroup clientGroup = new ClientGroup(clientId, thisSub.getGroup());
thisSub.getLiteTopicSet().forEach(lmqName -> {
removeTopicGroup(clientGroup, lmqName, false);
});
thisSub.getLmqSet().forEach(lmqName -> removeClientGroup(clientGroup, lmqName, false));
for (LiteCtlListener listener : listeners) {
listener.onRemoveAll(clientId, thisSub.getGroup());
}
@@ -195,52 +193,75 @@ public class LiteSubscriptionRegistryImpl extends ServiceThread implements LiteS
/**
* Get all subscribers for a specific LMQ, with optional group filtering.
* This method returns different types based on the subscription scenario:
* 1. When there's only one subscriber, return List<ClientGroup>
* 2. When group is specified, return List<ClientGroup> containing subscribers of that group
* 3. When group is null and multiple groups exist, return Map<String, List<ClientGroup>>
* mapping each group to its subscribers
* This method merges results from two sources:
* 1. Exact subscriptions from liteTopic2ClientGroup
* 2. Wildcard subscriptions from wildcardGroupMap
* <p>
* 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<ClientGroup> 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<String, List<ClientGroup>> getAllSubscribers(String group, String lmqName) {
Map<String, List<ClientGroup>> result = new HashMap<>();
if (LiteMetadataUtil.isWildcardGroup(group, brokerController)) {
collectWildcardSubscribers(result, group, lmqName);
} else {
SubscriberWrapper.MapWrapper wrapper = new SubscriberWrapper.MapWrapper();
Set<ClientGroup> 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<String> wildcardGroups = wildcardGroupMap.get(topic);
if (wildcardGroups != null) {
for (String wildcardGroup : wildcardGroups) {
List<ClientGroup> wildcardClients = getWildcardGroupClients(topic, wildcardGroup);
if (CollectionUtils.isNotEmpty(wildcardClients)) {
wrapper.getGroupMap().putIfAbsent(wildcardGroup, wildcardClients);
}
}
}
return wrapper;
}
return result;
}
@Override
public SubscriberWrapper.ListWrapper getWildcardSubscriber(String group, String topic) {
return new SubscriberWrapper.ListWrapper(getWildcardGroupClients(topic, group));
public List<ClientGroup> getWildcardGroupClients(String group) {
List<ClientGroup> list = null;
try {
list = wildcardClientCache.get(group, () -> {
String topic = LiteMetadataUtil.getLiteBindTopic(group, brokerController);
if (topic == null) {
return Collections.emptyList();
}
Set<ClientGroup> clientSet = liteTopic2ClientGroup.get(mockLmqNameForWildcardGroup(topic, group));
return clientSet != null ? new ArrayList<>(clientSet) : Collections.emptyList();
});
} catch (ExecutionException ignored) {
}
return list;
}
private void collectExactSubscribers(Map<String, List<ClientGroup>> result, String group, String lmqName) {
Set<ClientGroup> 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);
}
}
}
private void collectWildcardSubscribers(Map<String, List<ClientGroup>> result, String group, String lmqName) {
if (group == null) {
String topic = LiteUtil.getParentTopic(lmqName);
Set<String> wildcardGroups = wildcardGroupMap.get(topic);
if (wildcardGroups == null) {
return;
}
for (String wildcardGroup : wildcardGroups) {
List<ClientGroup> wildcardClients = getWildcardGroupClients(wildcardGroup);
if (CollectionUtils.isNotEmpty(wildcardClients)) {
result.putIfAbsent(wildcardGroup, wildcardClients);
}
}
} else if (LiteMetadataUtil.isWildcardGroup(group, brokerController)) {
List<ClientGroup> wildcardClients = getWildcardGroupClients(group);
if (CollectionUtils.isNotEmpty(wildcardClients)) {
result.put(group, wildcardClients);
}
}
}
/**
@@ -251,7 +272,7 @@ public class LiteSubscriptionRegistryImpl extends ServiceThread implements LiteS
*/
@Override
public void cleanSubscription(String lmqName, boolean notifyClient) {
Set<ClientGroup> topicGroupSet = liteTopic2Group.remove(lmqName);
Set<ClientGroup> topicGroupSet = liteTopic2ClientGroup.remove(lmqName);
if (CollectionUtils.isEmpty(topicGroupSet)) {
return;
}
@@ -260,7 +281,7 @@ public class LiteSubscriptionRegistryImpl extends ServiceThread implements LiteS
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 class LiteSubscriptionRegistryImpl extends ServiceThread implements LiteS
}
}
protected void addTopicGroup(ClientGroup clientGroup, String lmqName) {
Set<ClientGroup> 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<ClientGroup> topicGroupSet = liteTopic2Group.get(lmqName);
protected void removeClientGroup(ClientGroup clientGroup, String lmqName, boolean resetOffset) {
Set<ClientGroup> topicGroupSet = liteTopic2ClientGroup.get(lmqName);
if (topicGroupSet == null) {
return;
}
@@ -298,7 +319,7 @@ public class LiteSubscriptionRegistryImpl extends ServiceThread implements LiteS
}
}
if (topicGroupSet.isEmpty()) {
liteTopic2Group.remove(lmqName);
liteTopic2ClientGroup.remove(lmqName);
unmarkWildcardGroupIfNecessary(lmqName);
}
}
@@ -307,7 +328,7 @@ public class LiteSubscriptionRegistryImpl extends ServiceThread implements LiteS
* Remove clients that subscribe to the same liteTopic under the same group
*/
protected void excludeClientByLmqName(String newClientId, String group, String lmqName) {
Set<ClientGroup> clientSet = liteTopic2Group.get(lmqName);
Set<ClientGroup> clientSet = liteTopic2ClientGroup.get(lmqName);
if (CollectionUtils.isEmpty(clientSet)) {
return;
}
@@ -318,9 +339,9 @@ public class LiteSubscriptionRegistryImpl extends ServiceThread implements LiteS
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 @@ public class LiteSubscriptionRegistryImpl extends ServiceThread implements LiteS
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 @@ public class LiteSubscriptionRegistryImpl extends ServiceThread implements LiteS
return topic + "@" + group;
}
private List<ClientGroup> getWildcardGroupClients(String topic, String group) {
List<ClientGroup> list = null;
try {
list = wildcardClientCache.get(group, () -> {
Set<ClientGroup> 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.");
broker/src/main/java/org/apache/rocketmq/broker/lite/LmqPrefixIndex.java
+109
View File
@@ -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}.
*
* <p>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.
*
* <p>Used to accelerate prefix-subscription full dispatch in
* {@link LiteEventDispatcher#doFullDispatchForClient(String, String)}.
*
* <p>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<Boolean> 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<String, Boolean> visitor) {
if (StringUtils.isEmpty(lmqPrefix) || visitor == null) {
return false;
}
rwLock.readLock().lock();
try {
SortedMap<String, Boolean> 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();
}
}
broker/src/main/java/org/apache/rocketmq/broker/lite/RocksDBLiteLifecycleManager.java
-42
View File
@@ -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.RocksDBConsumeQueueOffsetTable;
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 class RocksDBLiteLifecycleManager extends AbstractLiteLifecycleManager {
return maxCqOffsetTable.getOrDefault(lmqName + "-0", -1L) + 1;
}
@Override
public List<String> collectByParentTopic(String parentTopic) {
if (StringUtils.isEmpty(parentTopic)) {
return Collections.emptyList();
}
List<String> resultList = new ArrayList<>();
Iterator<Map.Entry<String, Long>> iterator = maxCqOffsetTable.entrySet().iterator();
while (iterator.hasNext()) {
Map.Entry<String, Long> 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<Pair<String, String>> collectExpiredLiteTopic() {
List<Pair<String, String>> lmqToDelete = new ArrayList<>();
Iterator<Map.Entry<String, Long>> iterator = maxCqOffsetTable.entrySet().iterator();
while (iterator.hasNext()) {
Map.Entry<String, Long> 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();
broker/src/main/java/org/apache/rocketmq/broker/lite/SubscriberWrapper.java
-64
View File
@@ -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<ClientGroup> clients;
public ListWrapper() {
this.clients = new ArrayList<>();
}
public ListWrapper(List<ClientGroup> clients) {
this.clients = clients;
}
public List<ClientGroup> getClients() {
return this.clients;
}
}
public static class MapWrapper extends SubscriberWrapper {
private final Map<String, List<ClientGroup>> groupMap = new HashMap<>();
public MapWrapper() {
}
public Map<String, List<ClientGroup>> getGroupMap() {
return groupMap;
}
}
public ListWrapper asListWrapper() {
return this instanceof ListWrapper ? (ListWrapper) this : null;
}
public MapWrapper asMapWrapper() {
return this instanceof MapWrapper ? (MapWrapper) this : null;
}
}
broker/src/main/java/org/apache/rocketmq/broker/processor/LiteManagerProcessor.java
+6 -11
View File
@@ -20,15 +20,14 @@ package org.apache.rocketmq.broker.processor;
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.CombineConsumeQueueStore;
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 @@ public class LiteManagerProcessor implements NettyRequestProcessor {
Set<String> returnSet = null;
int liteTopicCount = 0;
LiteSubscription liteSubscription = brokerController.getLiteSubscriptionRegistry().getLiteSubscription(clientId);
if (liteSubscription != null && liteSubscription.getLiteTopicSet() != null) {
Set<String> liteTopicSet = liteSubscription.getLiteTopicSet();
if (liteSubscription != null && liteSubscription.getLmqSet() != null) {
Set<String> liteTopicSet = liteSubscription.getLmqSet();
liteTopicCount = liteTopicSet.size();
if (maxCount >= liteTopicCount) {
returnSet = liteTopicSet;
@@ -393,12 +391,9 @@ public class LiteManagerProcessor implements NettyRequestProcessor {
@VisibleForTesting
public Set<ClientGroup> getSubscriber(String lmqName) {
SubscriberWrapper.MapWrapper wrapper =
brokerController.getLiteSubscriptionRegistry().getAllSubscriber(null, lmqName).asMapWrapper();
if (null == wrapper) {
return Collections.emptySet();
}
return wrapper.getGroupMap().entrySet().stream()
Map<String, List<ClientGroup>> subscriberMap =
brokerController.getLiteSubscriptionRegistry().getAllSubscribers(null, lmqName);
return subscriberMap.entrySet().stream()
.flatMap(entry -> {
String group = entry.getKey();
if (LiteMetadataUtil.isWildcardGroup(group, brokerController)) {
broker/src/test/java/org/apache/rocketmq/broker/lite/AbstractLiteLifecycleManagerTest.java
+120 -24
View File
@@ -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.processor.PopLiteMessageProcessor;
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.Before;
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<String, ConcurrentMap<Integer, Long>> offsetTable = new ConcurrentHashMap<>();
private final ConcurrentMap<String, TopicConfig> topicConfigTable = new ConcurrentHashMap<>();
@Before
public void setUp() {
@@ -104,7 +109,7 @@ public class AbstractLiteLifecycleManagerTest {
topicConfig.getAttributes().put(
TopicAttributes.TOPIC_MESSAGE_TYPE_ATTRIBUTE.getName(), TopicMessageType.LITE.getValue());
ConcurrentMap<String, TopicConfig> 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 class AbstractLiteLifecycleManagerTest {
TestLiteLifecycleManager testObject = new TestLiteLifecycleManager(brokerController, liteSharding);
lifecycleManager = Mockito.spy(testObject);
lifecycleManager.init();
lifecycleManager.lmqPrefixIndex.add(EXIST_LMQ_NAME);
}
@After
@@ -128,6 +134,7 @@ public class AbstractLiteLifecycleManagerTest {
topicConfig.getAttributes().clear();
groupConfig.getAttributes().clear();
offsetTable.clear();
topicConfigTable.clear();
}
@Test
@@ -154,10 +161,23 @@ public class AbstractLiteLifecycleManagerTest {
@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 class AbstractLiteLifecycleManagerTest {
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 class AbstractLiteLifecycleManagerTest {
@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<String> parentCaptor = ArgumentCaptor.forClass(String.class);
ArgumentCaptor<String> lmqCaptor = ArgumentCaptor.forClass(String.class);
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, 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()));
lifecycleManager.cleanByParentTopic("whatever");
verify(lifecycleManager, never()).collectByParentTopic("whatever");
// other parent's LMQs remain untouched
List<String> 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(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());
}
@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<String> 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 class AbstractLiteLifecycleManagerTest {
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<Pair<String, String>> collectExpiredLiteTopic() {
return Collections.singletonList(new Pair<>(PARENT_TOPIC, EXIST_LMQ_NAME));
}
@Override
public List<String> 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<Triple<String, Long, Long>, Boolean> function) {
List<Triple<String, Long, Long>> 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<String, Long, Long> triple : triples) {
if (!function.apply(triple)) {
break;
}
}
}
}
}
broker/src/test/java/org/apache/rocketmq/broker/lite/LiteEventDispatcherTest.java
+63 -13
View File
@@ -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.List;
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.anyBoolean;
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 class LiteEventDispatcherTest {
@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 class LiteEventDispatcherTest {
@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 class LiteEventDispatcherTest {
fail("Exception should not be thrown");
}
verify(liteSubscriptionRegistry).getAllSubscriber("group", "lmqName");
verify(liteSubscriptionRegistry).getAllSubscribers("group", "lmqName");
}
@Test
@@ -144,11 +148,10 @@ public class LiteEventDispatcherTest {
subscriptionGroupConfig.setWildcardLiteGroup(false);
when(subscriptionGroupManager.findSubscriptionGroupConfig("group")).thenReturn(subscriptionGroupConfig);
SubscriberWrapper.ListWrapper listWrapper = mock(SubscriberWrapper.ListWrapper.class);
List<ClientGroup> 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<String, List<ClientGroup>> 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 class LiteEventDispatcherTest {
@Test
public void testDoDispatchWithMapWrapper() {
SubscriberWrapper.MapWrapper mapWrapper = mock(SubscriberWrapper.MapWrapper.class);
Map<String, List<ClientGroup>> 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 class LiteEventDispatcherTest {
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 class LiteEventDispatcherTest {
LiteSubscription subscription = new LiteSubscription();
Set<String> 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 class LiteEventDispatcherTest {
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<ClientGroup> clients = Collections.singletonList(new ClientGroup("clientId", group));
when(liteSubscriptionRegistry.getWildcardGroupClients(group)).thenReturn(clients);
doAnswer(invocation -> {
Function<Triple<String, Long, Long>, 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<String> 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);
}
}
broker/src/test/java/org/apache/rocketmq/broker/lite/LiteLifecycleManagerTest.java
+1 -73
View File
@@ -22,11 +22,8 @@ import org.apache.rocketmq.broker.offset.ConsumerOffsetManager;
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 class LiteLifecycleManagerTest {
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<String> 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<Pair<String, String>> 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<String, String> pair : result) {
Assert.assertEquals(parentTopic, pair.getObject1());
Assert.assertTrue(LiteUtil.belongsTo(pair.getObject2(), parentTopic));
}
}
@Ignore
@Test
public void testCleanExpiredLiteTopic() {
@@ -182,31 +137,4 @@ public class LiteLifecycleManagerTest {
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<String> 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));
}
}
}
}
broker/src/test/java/org/apache/rocketmq/broker/lite/LiteSubscriptionRegistryImplTest.java
+329 -43
View File
@@ -21,6 +21,7 @@ import io.netty.channel.Channel;
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.assertNotNull;
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 class LiteSubscriptionRegistryImplTest {
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 class LiteSubscriptionRegistryImplTest {
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 class LiteSubscriptionRegistryImplTest {
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 class LiteSubscriptionRegistryImplTest {
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 class LiteSubscriptionRegistryImplTest {
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 class LiteSubscriptionRegistryImplTest {
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 class LiteSubscriptionRegistryImplTest {
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 class LiteSubscriptionRegistryImplTest {
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 class LiteSubscriptionRegistryImplTest {
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 class LiteSubscriptionRegistryImplTest {
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 class LiteSubscriptionRegistryImplTest {
* 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<ClientGroup> clientSet = ConcurrentHashMap.newKeySet();
clientSet.add(clientGroup);
registry.liteTopic2ClientGroup.put(wildcardLmqName, clientSet);
Map<String, List<ClientGroup>> 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 class LiteSubscriptionRegistryImplTest {
ClientGroup clientGroup = new ClientGroup(clientId, group);
Set<ClientGroup> clientSet = ConcurrentHashMap.newKeySet();
clientSet.add(clientGroup);
registry.liteTopic2Group.put(lmqName, clientSet);
registry.liteTopic2ClientGroup.put(lmqName, clientSet);
SubscriberWrapper result = registry.getAllSubscriber(group, lmqName);
Map<String, List<ClientGroup>> 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<ClientGroup> 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 class LiteSubscriptionRegistryImplTest {
Set<ClientGroup> 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<String, List<ClientGroup>> 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 class LiteSubscriptionRegistryImplTest {
ClientGroup clientGroup = new ClientGroup(clientId, group);
Set<ClientGroup> 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 class LiteSubscriptionRegistryImplTest {
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<ClientGroup> 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<ClientGroup> 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<ClientGroup> clientSet = ConcurrentHashMap.newKeySet();
clientSet.add(cg);
registry.liteTopic2ClientGroup.put(topic + "@" + group, clientSet);
List<ClientGroup> 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<ClientGroup> 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"));
}
}
broker/src/test/java/org/apache/rocketmq/broker/lite/LmqPrefixIndexTest.java
+229
View File
@@ -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<String> 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<String> 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<String> 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());
}
}
broker/src/test/java/org/apache/rocketmq/broker/lite/RocksDBLiteLifecycleManagerTest.java
-76
View File
@@ -21,11 +21,8 @@ import org.apache.rocketmq.broker.BrokerController;
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 class RocksDBLiteLifecycleManagerTest {
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 class RocksDBLiteLifecycleManagerTest {
messageStore.shutdown();
messageStore.destroy();
UtilAll.deleteFile(new File(storePathRootDir));
mockTopicConfig = new TopicConfig();
}
@Ignore
@@ -147,48 +141,6 @@ public class RocksDBLiteLifecycleManagerTest {
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<String> 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<Pair<String, String>> 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<String, String> 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 class RocksDBLiteLifecycleManagerTest {
}
}
@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<String> 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();
broker/src/test/java/org/apache/rocketmq/broker/processor/LiteManagerProcessorTest.java
+15 -13
View File
@@ -20,8 +20,10 @@ package org.apache.rocketmq.broker.processor;
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.AbstractLiteLifecycleManager;
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 class LiteManagerProcessorTest {
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<String, List<ClientGroup>> 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 class LiteManagerProcessorTest {
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 class LiteManagerProcessorTest {
@Test
public void testGetSubscriber_null() {
String lmqName = "lmqName";
when(liteSubscriptionRegistry.getAllSubscriber(null, lmqName)).thenReturn(new SubscriberWrapper.ListWrapper());
Map<String, List<ClientGroup>> emptyMap = Collections.emptyMap();
when(liteSubscriptionRegistry.getAllSubscribers(null, lmqName)).thenReturn(emptyMap);
Set<ClientGroup> result = processor.getSubscriber(lmqName);
assertEquals(0, result.size());
@@ -756,9 +758,9 @@ public class LiteManagerProcessorTest {
@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<String, List<ClientGroup>> subscriberMap = new HashMap<>();
subscriberMap.put("group", Collections.singletonList(new ClientGroup("clientId", "group")));
when(liteSubscriptionRegistry.getAllSubscribers(null, lmqName)).thenReturn(subscriberMap);
Set<ClientGroup> result = processor.getSubscriber(lmqName);
assertEquals(1, result.size());
@@ -768,13 +770,13 @@ public class LiteManagerProcessorTest {
@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<String, List<ClientGroup>> 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<ClientGroup> result = processor.getSubscriber(lmqName);
common/src/main/java/org/apache/rocketmq/common/lite/LiteSubscription.java
+26 -23
View File
@@ -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<String> liteTopicSet = ConcurrentHashMap.newKeySet();
private final Set<String> lmqSet = ConcurrentHashMap.newKeySet();
private volatile long updateTime = System.currentTimeMillis();
public boolean addLiteTopic(String liteTopic) {
updateTime();
return this.liteTopicSet.add(liteTopic);
public LiteSubscription touch() {
this.updateTime = System.currentTimeMillis();
return this;
}
public void addLiteTopic(Collection<String> set) {
updateTime();
this.liteTopicSet.addAll(set);
public boolean addLmq(String lmqName) {
return this.lmqSet.add(lmqName);
}
public boolean removeLiteTopic(String liteTopic) {
updateTime();
return this.liteTopicSet.remove(liteTopic);
}
public void removeLiteTopic(Collection<String> set) {
updateTime();
this.liteTopicSet.removeAll(set);
public boolean removeLmq(String lmqName) {
return this.lmqSet.remove(lmqName);
}
public String getGroup() {
@@ -65,12 +59,13 @@ public class LiteSubscription {
return this;
}
public Set<String> getLiteTopicSet() {
return liteTopicSet;
public Set<String> getLmqSet() {
return lmqSet;
}
public LiteSubscription setLiteTopicSet(Set<String> liteTopicSet) {
this.liteTopicSet.addAll(liteTopicSet);
public LiteSubscription setLmqSet(Set<String> lmqSet) {
this.lmqSet.clear();
this.lmqSet.addAll(lmqSet);
return this;
}
@@ -82,8 +77,15 @@ public class LiteSubscription {
this.updateTime = updateTime;
}
private void updateTime() {
this.updateTime = System.currentTimeMillis();
public static Set<String> removals(Set<String> current, Set<String> target) {
Set<String> safeTarget = target == null ? Collections.emptySet() : target;
Set<String> result = new HashSet<>();
for (String item : current) {
if (!safeTarget.contains(item)) {
result.add(item);
}
}
return result;
}
@Override
@@ -91,8 +93,9 @@ public class LiteSubscription {
return "LiteSubscription{" +
"group='" + group + '\'' +
", topic='" + topic + '\'' +
", liteTopicSet=" + liteTopicSet +
", lmqSet=" + lmqSet +
", updateTime=" + updateTime +
'}';
}
}
common/src/test/java/org/apache/rocketmq/common/lite/LiteSubscriptionTest.java
+172
View File
@@ -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<String> 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<String> 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<String> current = new HashSet<>();
current.add("a");
current.add("b");
current.add("c");
Set<String> target = new HashSet<>();
target.add("b");
Set<String> result = LiteSubscription.removals(current, target);
assertEquals(2, result.size());
assertTrue(result.contains("a"));
assertTrue(result.contains("c"));
}
@Test
public void removals_targetNull() {
Set<String> current = new HashSet<>();
current.add("a");
current.add("b");
Set<String> result = LiteSubscription.removals(current, null);
assertEquals(2, result.size());
assertTrue(result.contains("a"));
assertTrue(result.contains("b"));
}
@Test
public void removals_noDiff() {
Set<String> current = new HashSet<>();
current.add("a");
Set<String> target = new HashSet<>();
target.add("a");
target.add("b");
Set<String> result = LiteSubscription.removals(current, target);
assertTrue(result.isEmpty());
}
@Test
public void removals_emptyCurrent() {
Set<String> 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"));
}
}
pom.xml
+6
View File
@@ -118,6 +118,7 @@
<slf4j-api.version>2.0.3</slf4j-api.version>
<rocketmq-shaded-slf4j-api-bridge.version>1.0.0</rocketmq-shaded-slf4j-api-bridge.version>
<commons-validator.version>1.10.0</commons-validator.version>
<commons-collections4.version>4.5.0</commons-collections4.version>
<zstd-jni.version>1.5.2-2</zstd-jni.version>
<lz4-java.version>1.10.3</lz4-java.version>
<opentracing.version>0.33.0</opentracing.version>
@@ -778,6 +779,11 @@
<artifactId>commons-validator</artifactId>
<version>${commons-validator.version}</version>
</dependency>
<dependency>
<groupId>org.apache.commons</groupId>
<artifactId>commons-collections4</artifactId>
<version>${commons-collections4.version}</version>
</dependency>
<dependency>
<groupId>com.github.luben</groupId>
<artifactId>zstd-jni</artifactId>