From 614b81693b8175bb59cff5de5cdc228f3d8b0735 Mon Sep 17 00:00:00 2001 From: Quan Date: Fri, 3 Apr 2026 11:23:08 +0800 Subject: [PATCH] [ISSUE #10203] Support wildcard subscription and and consumer suspend for LiteTopic (#10204) - Add wildcard (*) subscription support for liteTopic - Implement consume suspend mechanism with invalid scan count threshold - Refactor subscriber query interface with SubscriberWrapper for flexible retrieval - Add wildcard client cache with 30s TTL for performance optimization - Update related components and enhance test coverage Change-Id: I4ecaceec7daa2f4364d911437007df98dc49d542 --- WORKSPACE | 2 +- .../lite/AbstractLiteLifecycleManager.java | 24 +- .../broker/lite/LiteEventDispatcher.java | 248 ++-- .../broker/lite/LiteLifecycleManager.java | 27 + .../broker/lite/LiteMetadataUtil.java | 9 + .../broker/lite/LiteSubscriptionRegistry.java | 5 +- .../lite/LiteSubscriptionRegistryImpl.java | 123 +- .../lite/RocksDBLiteLifecycleManager.java | 22 + .../broker/lite/SubscriberWrapper.java | 64 + .../MemoryConsumerOrderInfoManager.java | 24 + .../orderly/QueueLevelConsumerManager.java | 2 +- .../ChangeInvisibleTimeProcessor.java | 14 +- .../processor/LiteManagerProcessor.java | 29 +- .../AbstractLiteLifecycleManagerTest.java | 7 + .../broker/lite/LiteEventDispatcherTest.java | 1008 +++++++------- .../LiteSubscriptionRegistryImplTest.java | 1200 +++++++---------- .../processor/LiteManagerProcessorTest.java | 58 +- .../apache/rocketmq/common/BrokerConfig.java | 10 + .../common/SubscriptionGroupAttributes.java | 12 +- pom.xml | 2 +- .../ChangeInvisibleDurationActivity.java | 8 +- .../processor/ReceiptHandleProcessor.java | 6 +- .../ChangeInvisibleDurationActivityTest.java | 33 +- .../subscription/SubscriptionGroupConfig.java | 24 +- 24 files changed, 1616 insertions(+), 1345 deletions(-) create mode 100644 broker/src/main/java/org/apache/rocketmq/broker/lite/SubscriberWrapper.java diff --git a/WORKSPACE b/WORKSPACE index 328c43995c..775bdc0a25 100644 --- a/WORKSPACE +++ b/WORKSPACE @@ -71,7 +71,7 @@ maven_install( "org.bouncycastle:bcpkix-jdk15on:1.69", "com.google.code.gson:gson:2.8.9", "com.googlecode.concurrentlinkedhashmap:concurrentlinkedhashmap-lru:1.4.2", - "org.apache.rocketmq:rocketmq-proto:2.1.1", + "org.apache.rocketmq:rocketmq-proto:2.1.2", "com.google.protobuf:protobuf-java:3.20.1", "com.google.protobuf:protobuf-java-util:3.20.1", "com.conversantmedia:disruptor:1.2.10", diff --git a/broker/src/main/java/org/apache/rocketmq/broker/lite/AbstractLiteLifecycleManager.java b/broker/src/main/java/org/apache/rocketmq/broker/lite/AbstractLiteLifecycleManager.java index e8fb2bde4d..eaf6288c5c 100644 --- a/broker/src/main/java/org/apache/rocketmq/broker/lite/AbstractLiteLifecycleManager.java +++ b/broker/src/main/java/org/apache/rocketmq/broker/lite/AbstractLiteLifecycleManager.java @@ -18,6 +18,7 @@ package org.apache.rocketmq.broker.lite; import com.google.common.collect.Sets; +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; @@ -32,6 +33,8 @@ import java.util.Collections; import java.util.List; import java.util.Map; import java.util.Set; +import java.util.concurrent.ConcurrentHashMap; +import java.util.function.Function; import static org.apache.rocketmq.broker.offset.ConsumerOffsetManager.TOPIC_GROUP_SEPARATOR; @@ -41,6 +44,7 @@ import static org.apache.rocketmq.broker.offset.ConsumerOffsetManager.TOPIC_GROU */ public abstract class AbstractLiteLifecycleManager extends ServiceThread { private static final Logger LOGGER = LoggerFactory.getLogger(LoggerName.ROCKETMQ_POP_LITE_LOGGER_NAME); + private static final int MAX_INVALID_SCAN_COUNT = 5; protected final BrokerController brokerController; protected final String brokerName; @@ -48,6 +52,7 @@ public abstract class AbstractLiteLifecycleManager extends ServiceThread { protected MessageStore messageStore; protected Map ttlMap = Collections.emptyMap(); protected Map> subscriberGroupMap = Collections.emptyMap(); + protected Map invalidScanCountMap = new ConcurrentHashMap<>(); public AbstractLiteLifecycleManager(BrokerController brokerController, LiteSharding liteSharding) { this.brokerController = brokerController; @@ -77,6 +82,15 @@ public abstract class AbstractLiteLifecycleManager extends ServiceThread { */ public abstract List collectByParentTopic(String parentTopic); + /** + * Iterator of lite topic, for high frequency iteration + * Triple, lastStoreTimestamp is null for now + * return true to continue, false to break. + * + * @param function consumer func + */ + public abstract void forEachLiteTopic(Function, Boolean> function); + /** * Check if the subscription for the given LMQ is active. * A subscription is considered active if either: @@ -153,8 +167,16 @@ public abstract class AbstractLiteLifecycleManager extends ServiceThread { return false; } if (maxOffset <= 0) { - LOGGER.warn("unexpected condition, max offset <= 0, {}, {}", lmqName, maxOffset); + int invalidCount = invalidScanCountMap.getOrDefault(lmqName, 0) + 1; + LOGGER.warn("unexpected condition, max offset <= 0, {}, {}, scanCount:{}", lmqName, maxOffset, invalidCount); + if (invalidCount > MAX_INVALID_SCAN_COUNT) { // check more times in case of concurrent issue + invalidScanCountMap.remove(lmqName); + return true; + } + invalidScanCountMap.put(lmqName, invalidCount); return false; + } else { + invalidScanCountMap.remove(lmqName); } long latestStoreTime = this.brokerController.getMessageStore().getMessageStoreTimeStamp(lmqName, 0, maxOffset - 1); diff --git a/broker/src/main/java/org/apache/rocketmq/broker/lite/LiteEventDispatcher.java b/broker/src/main/java/org/apache/rocketmq/broker/lite/LiteEventDispatcher.java index e2b82906a3..8bdb2879df 100644 --- a/broker/src/main/java/org/apache/rocketmq/broker/lite/LiteEventDispatcher.java +++ b/broker/src/main/java/org/apache/rocketmq/broker/lite/LiteEventDispatcher.java @@ -21,9 +21,9 @@ 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.broker.offset.ConsumerOffsetManager; -import org.apache.rocketmq.broker.pop.orderly.ConsumerOrderInfoManager; import org.apache.rocketmq.common.ServiceThread; import org.apache.rocketmq.common.constant.LoggerName; import org.apache.rocketmq.common.entity.ClientGroup; @@ -35,12 +35,10 @@ import org.apache.rocketmq.logging.org.slf4j.LoggerFactory; import java.util.ArrayList; import java.util.Collections; import java.util.Comparator; -import java.util.HashMap; import java.util.Iterator; import java.util.List; import java.util.Map; import java.util.Random; -import java.util.Set; import java.util.concurrent.BlockingQueue; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ConcurrentMap; @@ -48,15 +46,17 @@ 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 { private static final Logger LOGGER = LoggerFactory.getLogger(LoggerName.ROCKETMQ_POP_LITE_LOGGER_NAME); private static final Object PRESENT = new Object(); private static final long CLIENT_INACTIVE_INTERVAL = 10 * 1000; // inactive time when it has unprocessed events - private static final long CLIENT_LONG_POLLING_INTERVAL = 30 * 1000 + 5000; // at least a period of long polling as 30s - private static final long ACTIVE_CONSUMING_WINDOW = 5000; - private static final double LOW_WATER_MARK = 0.2; + protected static final long CLIENT_LONG_POLLING_INTERVAL = 30 * 1000 + 5000; // at least a period of long polling as 30s + protected static final long ACTIVE_CONSUMING_WINDOW = 5000; + protected static final double LOW_WATER_MARK = 0.2; private static final int BLACKLIST_EXPIRE_SECONDS = 10; private static final int SCAN_LOG_INTERVAL = 10000; @@ -64,11 +64,10 @@ public class LiteEventDispatcher extends ServiceThread { private final LiteSubscriptionRegistry liteSubscriptionRegistry; private final AbstractLiteLifecycleManager liteLifecycleManager; private final ConsumerOffsetManager consumerOffsetManager; - private ConsumerOrderInfoManager consumerOrderInfoManager; - private final ConcurrentMap clientEventMap = new ConcurrentHashMap<>(); - private final ConcurrentSkipListSet fullDispatchSet = new ConcurrentSkipListSet<>(COMPARATOR); - private final ConcurrentMap fullDispatchMap = new ConcurrentHashMap<>(); // deduplication + protected final ConcurrentMap clientEventMap = new ConcurrentHashMap<>(); + protected final ConcurrentSkipListSet fullDispatchSet = new ConcurrentSkipListSet<>(COMPARATOR); + protected final ConcurrentMap fullDispatchMap = new ConcurrentHashMap<>(); // deduplication private final Cache blacklist = CacheBuilder.newBuilder().expireAfterWrite(BLACKLIST_EXPIRE_SECONDS, TimeUnit.SECONDS).build(); private final Random random = ThreadLocalRandom.current(); @@ -83,7 +82,6 @@ public class LiteEventDispatcher extends ServiceThread { } public void init() { - this.consumerOrderInfoManager = brokerController.getPopLiteMessageProcessor().getConsumerOrderInfoManager(); this.liteSubscriptionRegistry.addListener(new LiteCtlListenerImpl()); } @@ -106,20 +104,19 @@ public class LiteEventDispatcher extends ServiceThread { doDispatch(group, lmqName, null); } - @SuppressWarnings("unchecked") - private void doDispatch(String group, String lmqName, String excludeClientId) { + protected void doDispatch(String group, String lmqName, String excludeClientId) { if (!this.brokerController.getBrokerConfig().isEnableLiteEventMode()) { return; } - Object subscribers = getAllSubscriber(group, lmqName); - if (null == subscribers) { + SubscriberWrapper wrapper = liteSubscriptionRegistry.getAllSubscriber(group, lmqName); + if (null == wrapper) { return; } - if (subscribers instanceof List) { - selectAndDispatch(lmqName, (List) subscribers, excludeClientId); + if (wrapper instanceof SubscriberWrapper.ListWrapper) { + selectAndDispatch(lmqName, wrapper.asListWrapper().getClients(), excludeClientId); } - if (subscribers instanceof Map) { - Map> map = (Map>) subscribers; + if (wrapper instanceof SubscriberWrapper.MapWrapper) { + Map> map = wrapper.asMapWrapper().getGroupMap(); map.forEach((key, value) -> selectAndDispatch(lmqName, value, excludeClientId)); } } @@ -132,72 +129,73 @@ public class LiteEventDispatcher extends ServiceThread { * * @param clients all clients of one group * @param excludeClientId the client ID to exclude from selection, probably consuming blocked. + * @return true if dispatched to one client */ @VisibleForTesting - public void selectAndDispatch(String lmqName, List clients, String excludeClientId) { + public boolean selectAndDispatch(String lmqName, List clients, String excludeClientId) { if (!this.brokerController.getBrokerConfig().isEnableLiteEventMode()) { - return; + return true; } if (CollectionUtils.isEmpty(clients)) { - return; + return true; } - String clientId = null; // the selected one - if (clients.size() == 1) { - clientId = clients.get(0).clientId; - if (brokerController.getBrokerConfig().isEnableLitePopLog() && clientId.equals(excludeClientId)) { - LOGGER.info("no others, still dispatch to {}, {}", clientId, lmqName); + String group = clients.get(0).group; + boolean isWildcardGroup = LiteMetadataUtil.isWildcardGroup(group, brokerController); + String selectedClient = null; // the selected one + int start = random.nextInt(clients.size()); + List fallbackList = new ArrayList<>(clients.size()); + for (int i = 0; i < clients.size(); i++) { + int index = (start + i) % clients.size(); + if (clients.get(index).clientId.equals(excludeClientId)) { + fallbackList.add(clients.get(index)); + continue; } - if (!tryDispatchToClient(lmqName, clientId, clients.get(0).group)) { - clientId = null; + if (blacklist.getIfPresent(clients.get(index).clientId) != null) { + if (!isWildcardGroup) { // prevent iterating twice for large client set + fallbackList.add(clients.get(index)); + } + continue; } - } else { - int start = random.nextInt(clients.size()); - boolean dispatched = false; - List fallbackList = new ArrayList<>(clients.size()); - for (int i = 0; i < clients.size(); i++) { - int index = (start + i) % clients.size(); - clientId = clients.get(index).clientId; - if (clientId.equals(excludeClientId)) { - fallbackList.add(clients.get(index)); - continue; - } - if (blacklist.getIfPresent(clientId) != null) { - fallbackList.add(clients.get(index)); - continue; - } - if (tryDispatchToClient(lmqName, clientId, clients.get(index).group)) { - dispatched = true; + if (tryDispatchToClient(lmqName, clients.get(index).clientId, group, !isWildcardGroup)) { + selectedClient = clients.get(index).clientId; + break; + } + } + if (null == selectedClient) { + for (ClientGroup clientGroup : fallbackList) { + if (tryDispatchToClient(lmqName, clientGroup.clientId, group, !isWildcardGroup)) { + selectedClient = clientGroup.clientId; break; } } - if (!dispatched) { - clientId = null; - for (ClientGroup clientGroup : fallbackList) { - if (tryDispatchToClient(lmqName, clientGroup.clientId, clientGroup.group)) { - clientId = clientGroup.clientId; - break; - } - } - } } - if (clientId != null) { + if (selectedClient != null) { this.brokerController.getPopLiteMessageProcessor().getPopLiteLongPollingService() - .notifyMessageArriving(clientId, true, 0, clients.get(0).group); + .notifyMessageArriving(selectedClient, true, 0, group); + } else if (isWildcardGroup) { // no one available in this group, so schedule a full dispatch once + scheduleFullDispatchForWildcardGroup(group, + brokerController.getBrokerConfig().getLiteEventFullDispatchDelayTimeForWildcardGroup()); } + return selectedClient != null; } /** * Try to dispatch an event to a selected client by adding it to the client's event queue. * If the event queue is full, mark a full dispatch for retry later. + * @param scheduleFullDispatchIfFull schedule full dispatch if full, only false if it's a wildcard group. */ @VisibleForTesting - public boolean tryDispatchToClient(String lmqName, String clientId, String group) { + public boolean tryDispatchToClient(String lmqName, String clientId, String group, boolean scheduleFullDispatchIfFull) { ClientEventSet eventSet = clientEventMap.computeIfAbsent(clientId, key -> new ClientEventSet(group)); if (eventSet.offer(lmqName)) { return true; } - scheduleFullDispatch(clientId, group, blacklist.getIfPresent(clientId) != null); + if (scheduleFullDispatchIfFull) { + long delayTime = brokerController.getBrokerConfig().getLiteEventFullDispatchDelayTime() + + (blacklist.getIfPresent(clientId) != null ? random.nextInt(15 * 1000) : 0); + scheduleFullDispatchForClient(clientId, group, delayTime); + } LOGGER.warn("client event set is full. {}", clientId); return false; } @@ -224,7 +222,7 @@ public class LiteEventDispatcher extends ServiceThread { * It iterates through all LMQ topics subscribed by the client and dispatches events for those * with available messages. */ - public void doFullDispatch(String clientId, String group) { + public void doFullDispatchForClient(String clientId, String group) { if (!this.brokerController.getBrokerConfig().isEnableLiteEventMode()) { return; } @@ -236,13 +234,15 @@ public class LiteEventDispatcher extends ServiceThread { ClientEventSet eventSet = clientEventMap.computeIfAbsent(clientId, key -> new ClientEventSet(group)); if (eventSet.maybeBlock()) { LOGGER.warn("client may block for a while, wait another period. {}", clientId); - scheduleFullDispatch(clientId, group, true); + scheduleFullDispatchForClient(clientId, group, + brokerController.getBrokerConfig().getLiteEventFullDispatchDelayTime() + random.nextInt(15 * 1000)); return; } boolean isActiveConsuming = eventSet.isActiveConsuming(); if (!eventSet.isLowWaterMark()) { LOGGER.warn("client event set high water mark, wait another period. {}, {}", clientId, isActiveConsuming); - scheduleFullDispatch(clientId, group, !isActiveConsuming); + scheduleFullDispatchForClient(clientId, group, brokerController.getBrokerConfig().getLiteEventFullDispatchDelayTime() + + (isActiveConsuming ? 0 : random.nextInt(10 * 1000))); return; } LOGGER.info("client full dispatch, {}, total:{}", clientId, subscription.getLiteTopicSet().size()); @@ -263,7 +263,8 @@ public class LiteEventDispatcher extends ServiceThread { } } else { LOGGER.warn("client event set full again, wait another period. {}, {}", clientId, isActiveConsuming); - scheduleFullDispatch(clientId, group, !isActiveConsuming); + scheduleFullDispatchForClient(clientId, group, brokerController.getBrokerConfig().getLiteEventFullDispatchDelayTime() + + (isActiveConsuming ? 0 : random.nextInt(15 * 1000))); break; } } @@ -273,63 +274,74 @@ public class LiteEventDispatcher extends ServiceThread { } /** - * Perform a full dispatch for all clients under a specific group, only invoked by admin for now. + * Perform a full dispatch for wildcard group which was previously marked for a delayed full dispatch. + * It iterates through all LMQ topics in CQ table, so it may be a heavy work. */ - public void doFullDispatchByGroup(String group) { - List clientIds = liteSubscriptionRegistry.getAllClientIdByGroup(group); - LOGGER.info("do full dispatch by group, {}, size:{}", group, clientIds.size()); - for (String clientId : clientIds) { - doFullDispatch(clientId, group); - } - } - - public void scheduleFullDispatch(String clientId, String group, boolean reentry) { - if (fullDispatchMap.putIfAbsent(clientId, PRESENT) != null) { + public void doFullDispatchForWildcardGroup(String group) { + if (!this.brokerController.getBrokerConfig().isEnableLiteEventMode()) { return; } - int randomDelay = reentry ? random.nextInt(25 * 1000) : 0; - fullDispatchSet.add(new FullDispatchRequest(clientId, group, - brokerController.getBrokerConfig().getLiteEventFullDispatchDelayTime() + randomDelay)); + String parentTopic = LiteMetadataUtil.getLiteBindTopic(group, brokerController); + if (null == parentTopic || !LiteMetadataUtil.isWildcardGroup(group, brokerController)) { + return; + } + List clients = liteSubscriptionRegistry.getWildcardSubscriber(group, parentTopic).getClients(); + if (CollectionUtils.isEmpty(clients)) { + return; + } + AtomicInteger count = new AtomicInteger(); + Function, Boolean> function = triple -> { + String lmqName = triple.getLeft(); + long maxOffset = triple.getMiddle(); + if (!LiteUtil.belongsTo(lmqName, parentTopic)) { + return true; + } + if (maxOffset <= 0) { + return true; + } + long consumerOffset = consumerOffsetManager.queryOffset(group, lmqName, 0); + if (consumerOffset >= maxOffset) { + return true; + } + if (selectAndDispatch(lmqName, clients, null)) { + count.incrementAndGet(); + } else { + LOGGER.warn("doFullDispatchForWildcardGroup, wait another period. {}", group); + return false; + } + return true; + }; + liteLifecycleManager.forEachLiteTopic(function); + LOGGER.info("doFullDispatchForWildcardGroup finish. {}, dispatch:{}", group, count); } /** - * Get all subscribers for a specific LMQ, with optional group filtering. - * To avoid unnecessary comparisons and wrapping, Object is used as the return type here. - * This method returns different types based on the subscription scenario: - * 1. When there's only one subscriber, return List - * 2. When group is specified, return List containing subscribers of that group - * 3. When group is null and multiple groups exist, return Map> - * mapping each group to its subscribers - * - * @return Object that can be either List or Map> or null if not found + * Perform a full dispatch for all clients under a specific group, only invoked by admin for now. */ - @VisibleForTesting - public Object getAllSubscriber(String group, String lmqName) { - Set observers = liteSubscriptionRegistry.getSubscriber(lmqName); - if (null == observers || observers.isEmpty()) { - return null; - } - if (observers.size() == 1) { - if (null == group || group.equals(observers.iterator().next().group)) { - return new ArrayList<>(observers); + public void doFullDispatchByGroup(String group) { + if (LiteMetadataUtil.isWildcardGroup(group, brokerController)) { + doFullDispatchForWildcardGroup(group); + } else { + List clientIds = liteSubscriptionRegistry.getAllClientIdByGroup(group); + LOGGER.info("do full dispatch by group, {}, size:{}", group, clientIds.size()); + for (String clientId : clientIds) { + doFullDispatchForClient(clientId, group); } - return null; - } - if (group != null) { - List result = new ArrayList<>(4); - for (ClientGroup ele : observers) { - if (group.equals(ele.group)) { - result.add(ele); - } - } - return !result.isEmpty() ? result : null; } + } - Map> group2Clients = new HashMap<>(4); - for (ClientGroup ele : observers) { - group2Clients.computeIfAbsent(ele.group, k -> new ArrayList<>(2)).add(ele); + public void scheduleFullDispatchForClient(String clientId, String group, long delayTime) { + if (fullDispatchMap.putIfAbsent(clientId, PRESENT) != null) { + return; } - return group2Clients; + if (delayTime < 50) { + delayTime = 50; + } + fullDispatchSet.add(new FullDispatchRequest(clientId, group, delayTime)); + } + + public void scheduleFullDispatchForWildcardGroup(String group, long delayTime) { + scheduleFullDispatchForClient("$" + group + "$", group, delayTime); // $group$ as clientId, no conflict expected } /** @@ -410,7 +422,11 @@ public class LiteEventDispatcher extends ServiceThread { break; } fullDispatchMap.remove(request.clientId); - doFullDispatch(request.clientId, request.group); + if (LiteMetadataUtil.isWildcardGroup(request.group, brokerController)) { + doFullDispatchForWildcardGroup(request.group); + } else { + doFullDispatchForClient(request.clientId, request.group); + } } } @@ -423,7 +439,7 @@ public class LiteEventDispatcher extends ServiceThread { * and ensure event deduplication to avoid duplicate events, although it * has a bit more memory usage than a single concurrent set. */ - class ClientEventSet { + protected class ClientEventSet { private final BlockingQueue events; private final ConcurrentMap map = new ConcurrentHashMap<>(); private final String group; @@ -486,8 +502,12 @@ public class LiteEventDispatcher extends ServiceThread { @Override public void onRegister(String clientId, String group, String lmqName) { - if (liteLifecycleManager.isLmqExist(lmqName)) { - doDispatch(group, lmqName, null); + if (LiteMetadataUtil.isWildcardGroup(group, brokerController)) { + scheduleFullDispatchForWildcardGroup(group, 5000); + } else { + if (liteLifecycleManager.isLmqExist(lmqName)) { + doDispatch(group, lmqName, null); + } } } @@ -548,7 +568,7 @@ public class LiteEventDispatcher extends ServiceThread { } } - static class FullDispatchRequest { + protected static class FullDispatchRequest { private final String clientId; private final String group; private final long timestamp; diff --git a/broker/src/main/java/org/apache/rocketmq/broker/lite/LiteLifecycleManager.java b/broker/src/main/java/org/apache/rocketmq/broker/lite/LiteLifecycleManager.java index 8cbf9c48e5..55af9e9215 100644 --- a/broker/src/main/java/org/apache/rocketmq/broker/lite/LiteLifecycleManager.java +++ b/broker/src/main/java/org/apache/rocketmq/broker/lite/LiteLifecycleManager.java @@ -18,6 +18,7 @@ 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; @@ -32,6 +33,7 @@ import java.util.Iterator; import java.util.List; import java.util.Map; import java.util.concurrent.ConcurrentMap; +import java.util.function.Function; public class LiteLifecycleManager extends AbstractLiteLifecycleManager { private static final Logger LOGGER = LoggerFactory.getLogger(LoggerName.ROCKETMQ_POP_LITE_LOGGER_NAME); @@ -86,4 +88,29 @@ public class LiteLifecycleManager extends AbstractLiteLifecycleManager { } return lmqToDelete; } + + @Override + public void forEachLiteTopic(Function, Boolean> function) { + Iterator>> iterator = + messageStore.getQueueStore().getConsumeQueueTable().entrySet().iterator(); + while (iterator.hasNext()) { + Map.Entry> entry = iterator.next(); + if (!LiteUtil.isLiteTopicQueue(entry.getKey())) { + continue; + } + ConsumeQueueInterface consumeQueueInterface = entry.getValue().get(0); + if (null == consumeQueueInterface) { + continue; + } + Triple triple = Triple.of(entry.getKey(), consumeQueueInterface.getMaxOffsetInQueue(), null); + try { + if (!function.apply(triple)) { + break; + } + } catch (Throwable e) { + LOGGER.error("forEachLiteTopic error. {}", entry.getKey(), e); + break; + } + } + } } diff --git a/broker/src/main/java/org/apache/rocketmq/broker/lite/LiteMetadataUtil.java b/broker/src/main/java/org/apache/rocketmq/broker/lite/LiteMetadataUtil.java index aa78f384a9..92aadfb6f0 100644 --- a/broker/src/main/java/org/apache/rocketmq/broker/lite/LiteMetadataUtil.java +++ b/broker/src/main/java/org/apache/rocketmq/broker/lite/LiteMetadataUtil.java @@ -103,6 +103,15 @@ public class LiteMetadataUtil { return groupConfig.getMaxClientEventCount(); } + public static boolean isWildcardGroup(String group, BrokerController brokerController) { + if (null == group || null == brokerController) { + return false; + } + SubscriptionGroupConfig groupConfig = + brokerController.getSubscriptionGroupManager().findSubscriptionGroupConfig(group); + return groupConfig != null && groupConfig.isWildcardLiteGroup(); + } + public static Map getTopicTtlMap(BrokerController brokerController) { if (null == brokerController) { return Collections.emptyMap(); diff --git a/broker/src/main/java/org/apache/rocketmq/broker/lite/LiteSubscriptionRegistry.java b/broker/src/main/java/org/apache/rocketmq/broker/lite/LiteSubscriptionRegistry.java index 92d6b4ea7c..50fbc373f2 100644 --- a/broker/src/main/java/org/apache/rocketmq/broker/lite/LiteSubscriptionRegistry.java +++ b/broker/src/main/java/org/apache/rocketmq/broker/lite/LiteSubscriptionRegistry.java @@ -21,7 +21,6 @@ import io.netty.channel.Channel; import java.util.List; 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; @@ -43,7 +42,9 @@ public interface LiteSubscriptionRegistry { void addListener(LiteCtlListener listener); - Set getSubscriber(String lmqName); + SubscriberWrapper getAllSubscriber(String group, String lmqName); + + SubscriberWrapper.ListWrapper getWildcardSubscriber(String group, String parentTopic); List getAllClientIdByGroup(String group); diff --git a/broker/src/main/java/org/apache/rocketmq/broker/lite/LiteSubscriptionRegistryImpl.java b/broker/src/main/java/org/apache/rocketmq/broker/lite/LiteSubscriptionRegistryImpl.java index dc02e6393a..878ff13788 100644 --- a/broker/src/main/java/org/apache/rocketmq/broker/lite/LiteSubscriptionRegistryImpl.java +++ b/broker/src/main/java/org/apache/rocketmq/broker/lite/LiteSubscriptionRegistryImpl.java @@ -18,17 +18,23 @@ package org.apache.rocketmq.broker.lite; import com.google.common.annotations.VisibleForTesting; +import com.google.common.cache.Cache; +import com.google.common.cache.CacheBuilder; import io.netty.channel.Channel; import java.util.ArrayList; +import java.util.Collections; import java.util.List; import java.util.Map; import java.util.Objects; import java.util.Set; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ConcurrentMap; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicInteger; import java.util.stream.Collectors; import org.apache.commons.collections.CollectionUtils; +import org.apache.commons.lang3.StringUtils; import org.apache.rocketmq.broker.BrokerController; import org.apache.rocketmq.common.ServiceThread; import org.apache.rocketmq.common.constant.LoggerName; @@ -47,8 +53,11 @@ public class LiteSubscriptionRegistryImpl extends ServiceThread implements LiteS protected final ConcurrentMap clientChannels = new ConcurrentHashMap<>(); protected final ConcurrentMap client2Subscription = new ConcurrentHashMap<>(); protected final ConcurrentMap> liteTopic2Group = new ConcurrentHashMap<>(); + protected final ConcurrentMap> wildcardGroupMap = new ConcurrentHashMap<>(); + private final Cache> wildcardClientCache = + CacheBuilder.newBuilder().maximumSize(2000).expireAfterWrite(30, TimeUnit.SECONDS).build(); - private final List listeners = new ArrayList<>(); + protected final List listeners = new ArrayList<>(); private final BrokerController brokerController; private final AbstractLiteLifecycleManager liteLifecycleManager; @@ -75,6 +84,9 @@ public class LiteSubscriptionRegistryImpl extends ServiceThread implements LiteS // No need to check existence, if reach here, it must be new. throw new LiteQuotaException("lite subscription quota exceeded " + maxCount); } + if (LiteMetadataUtil.isWildcardGroup(group, brokerController)) { + throw new IllegalStateException("subscribe lite operation is not supported for this group"); + } LiteSubscription thisSub = getOrCreateLiteSubscription(clientId, group, topic); // Utilize existing string object @@ -106,9 +118,15 @@ public class LiteSubscriptionRegistryImpl extends ServiceThread implements LiteS @Override public void addCompleteSubscription(String clientId, String group, String topic, Set lmqNameAll, long version) { - Set lmqNameNew = lmqNameAll.stream() - .filter(lmqName -> liteLifecycleManager.isSubscriptionActive(topic, lmqName)) - .collect(Collectors.toSet()); + Set 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()); + } LiteSubscription thisSub = getOrCreateLiteSubscription(clientId, group, topic); Set lmqNamePrev = thisSub.getLiteTopicSet(); @@ -150,9 +168,54 @@ public class LiteSubscriptionRegistryImpl extends ServiceThread implements LiteS listeners.add(listener); } + /** + * 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 + * 2. When group is specified, return List containing subscribers of that group + * 3. When group is null and multiple groups exist, return Map> + * mapping each group to its subscribers + */ @Override - public Set getSubscriber(String lmqName) { - return liteTopic2Group.get(lmqName); + public SubscriberWrapper getAllSubscriber(String group, String lmqName) { + String topic = LiteUtil.getParentTopic(lmqName); + + if (group != null) { + if (LiteMetadataUtil.isWildcardGroup(group, brokerController)) { + return getWildcardSubscriber(group, topic); + } + SubscriberWrapper.ListWrapper wrapper = new SubscriberWrapper.ListWrapper(); + Set subscribers = liteTopic2Group.get(lmqName); + if (subscribers != null) { + wrapper.getClients().addAll(subscribers.stream() + .filter(clientGroup -> group.equals(clientGroup.group)) + .collect(Collectors.toSet())); + } + return wrapper; + } else { + SubscriberWrapper.MapWrapper wrapper = new SubscriberWrapper.MapWrapper(); + Set subscribers = liteTopic2Group.get(lmqName); + if (subscribers != null) { + for (ClientGroup clientGroup : subscribers) { + wrapper.getGroupMap().computeIfAbsent(clientGroup.group, k -> new ArrayList<>()).add(clientGroup); + } + } + Set wildcardGroups = wildcardGroupMap.get(topic); + if (wildcardGroups != null) { + for (String wildcardGroup : wildcardGroups) { + List wildcardClients = getWildcardGroupClients(topic, wildcardGroup); + if (CollectionUtils.isNotEmpty(wildcardClients)) { + wrapper.getGroupMap().putIfAbsent(wildcardGroup, wildcardClients); + } + } + } + return wrapper; + } + } + + @Override + public SubscriberWrapper.ListWrapper getWildcardSubscriber(String group, String topic) { + return new SubscriberWrapper.ListWrapper(getWildcardGroupClients(topic, group)); } /** @@ -186,6 +249,7 @@ public class LiteSubscriptionRegistryImpl extends ServiceThread implements LiteS .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); } @@ -199,6 +263,7 @@ public class LiteSubscriptionRegistryImpl extends ServiceThread implements LiteS } if (topicGroupSet.remove(clientGroup)) { activeNum.decrementAndGet(); + invalidateWildcardCacheIfNecessary(clientGroup.group); for (LiteCtlListener listener : listeners) { listener.onUnregister(clientGroup.clientId, clientGroup.group, lmqName); } @@ -209,6 +274,7 @@ public class LiteSubscriptionRegistryImpl extends ServiceThread implements LiteS } if (topicGroupSet.isEmpty()) { liteTopic2Group.remove(lmqName); + unmarkWildcardGroupIfNecessary(lmqName); } } @@ -228,6 +294,10 @@ public class LiteSubscriptionRegistryImpl extends ServiceThread implements LiteS LiteSubscription liteSubscription = client2Subscription.get(clientGroup.clientId); if (liteSubscription != null) { liteSubscription.removeLiteTopic(lmqName); + // remove client if no more liteTopic + if (liteSubscription.getLiteTopicSet().isEmpty()) { + client2Subscription.remove(clientGroup.clientId); + } } notifyUnsubscribeLite(clientGroup.clientId, clientGroup.group, lmqName); boolean resetOffset = LiteMetadataUtil.isResetOffsetInExclusiveMode(group, brokerController); @@ -240,7 +310,7 @@ public class LiteSubscriptionRegistryImpl extends ServiceThread implements LiteS /** * Notify the client to remove the liteTopic subscription from its local memory */ - private void notifyUnsubscribeLite(String clientId, String group, String lmqName) { + protected void notifyUnsubscribeLite(String clientId, String group, String lmqName) { String topic = LiteUtil.getParentTopic(lmqName); String liteTopic = LiteUtil.getLiteTopic(lmqName); Channel channel = clientChannels.get(clientId); @@ -318,6 +388,45 @@ public class LiteSubscriptionRegistryImpl extends ServiceThread implements LiteS return curLiteSubscription; } + private void invalidateWildcardCacheIfNecessary(String group) { + if (LiteMetadataUtil.isWildcardGroup(group, brokerController)) { + wildcardClientCache.invalidate(group); + } + } + + private void markWildcardGroup(String topic, String group) { + wildcardGroupMap.computeIfAbsent(topic, k -> ConcurrentHashMap.newKeySet()).add(group); + } + + private void unmarkWildcardGroupIfNecessary(String lmqName) { + if (!LiteUtil.isLiteTopicQueue(lmqName)) { // must be topic@group + String[] topicAtGroup = StringUtils.split(lmqName); + if (null == topicAtGroup || topicAtGroup.length != 2) { + return; + } + wildcardGroupMap.computeIfPresent(topicAtGroup[0], (k, v) -> { + v.remove(topicAtGroup[1]); + return v.isEmpty() ? null : v; + }); + } + } + + private String mockLmqNameForWildcardGroup(String topic, String group) { + return topic + "@" + group; + } + + private List getWildcardGroupClients(String topic, String group) { + List list = null; + try { + list = wildcardClientCache.get(group, () -> { + Set clientSet = liteTopic2Group.get(mockLmqNameForWildcardGroup(topic, group)); + return clientSet != null ? new ArrayList<>(clientSet) : Collections.emptyList(); + }); + } catch (ExecutionException ignored) { + } + return list; + } + @Override public void run() { LOGGER.info("Start checking lite subscription."); diff --git a/broker/src/main/java/org/apache/rocketmq/broker/lite/RocksDBLiteLifecycleManager.java b/broker/src/main/java/org/apache/rocketmq/broker/lite/RocksDBLiteLifecycleManager.java index fb0eb51540..fb51a9afce 100644 --- a/broker/src/main/java/org/apache/rocketmq/broker/lite/RocksDBLiteLifecycleManager.java +++ b/broker/src/main/java/org/apache/rocketmq/broker/lite/RocksDBLiteLifecycleManager.java @@ -19,6 +19,7 @@ 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; @@ -36,6 +37,7 @@ import java.util.Iterator; import java.util.List; import java.util.Map; import java.util.concurrent.ConcurrentMap; +import java.util.function.Function; public class RocksDBLiteLifecycleManager extends AbstractLiteLifecycleManager { private static final Logger LOGGER = LoggerFactory.getLogger(LoggerName.ROCKETMQ_POP_LITE_LOGGER_NAME); @@ -110,4 +112,24 @@ public class RocksDBLiteLifecycleManager extends AbstractLiteLifecycleManager { LOGGER.error("LiteLifecycleManager-init error", e); } } + + @Override + public void forEachLiteTopic(Function, Boolean> function) { + for (Map.Entry entry : maxCqOffsetTable.entrySet()) { + String queueAndQid = entry.getKey(); + String queueName = queueAndQid.substring(0, queueAndQid.lastIndexOf("-")); + if (!LiteUtil.isLiteTopicQueue(queueName)) { + continue; + } + Triple triple = Triple.of(queueName, entry.getValue() + 1, null); + try { + if (!function.apply(triple)) { + break; + } + } catch (Throwable e) { + LOGGER.error("forEachLiteTopic error. {}", queueName, e); + break; + } + } + } } diff --git a/broker/src/main/java/org/apache/rocketmq/broker/lite/SubscriberWrapper.java b/broker/src/main/java/org/apache/rocketmq/broker/lite/SubscriberWrapper.java new file mode 100644 index 0000000000..97c02e5282 --- /dev/null +++ b/broker/src/main/java/org/apache/rocketmq/broker/lite/SubscriberWrapper.java @@ -0,0 +1,64 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.rocketmq.broker.lite; + +import org.apache.rocketmq.common.entity.ClientGroup; + +import java.util.ArrayList; +import java.util.HashMap; +import java.util.List; +import java.util.Map; + +public abstract class SubscriberWrapper { + + public static class ListWrapper extends SubscriberWrapper { + private final List clients; + + public ListWrapper() { + this.clients = new ArrayList<>(); + } + + public ListWrapper(List clients) { + this.clients = clients; + } + + public List getClients() { + return this.clients; + } + } + + public static class MapWrapper extends SubscriberWrapper { + private final Map> groupMap = new HashMap<>(); + + public MapWrapper() { + } + + public Map> getGroupMap() { + return groupMap; + } + } + + public ListWrapper asListWrapper() { + return this instanceof ListWrapper ? (ListWrapper) this : null; + } + + public MapWrapper asMapWrapper() { + return this instanceof MapWrapper ? (MapWrapper) this : null; + } + +} diff --git a/broker/src/main/java/org/apache/rocketmq/broker/offset/MemoryConsumerOrderInfoManager.java b/broker/src/main/java/org/apache/rocketmq/broker/offset/MemoryConsumerOrderInfoManager.java index 94acc454fa..fad3a5b444 100644 --- a/broker/src/main/java/org/apache/rocketmq/broker/offset/MemoryConsumerOrderInfoManager.java +++ b/broker/src/main/java/org/apache/rocketmq/broker/offset/MemoryConsumerOrderInfoManager.java @@ -20,6 +20,8 @@ package org.apache.rocketmq.broker.offset; import org.apache.rocketmq.broker.BrokerController; import org.apache.rocketmq.broker.pop.orderly.QueueLevelConsumerManager; +import java.util.concurrent.ConcurrentHashMap; + /** * Memory-based Consumer Order Information Manager for Lite Topics * Trade-off considerations:: @@ -46,6 +48,28 @@ public class MemoryConsumerOrderInfoManager extends QueueLevelConsumerManager { } } + public void suspendQueue(String topic, String group, int queueId, long popTime, long visibilityTimeout) { + ConcurrentHashMap orderInfoMap = this.getTable().get(buildKey(topic, group)); + if (null == orderInfoMap) { + return; + } + OrderInfo orderInfo = orderInfoMap.get(queueId); + if (null == orderInfo) { + return; + } + if (popTime != orderInfo.getPopTime()) { + log.warn("suspendQueue, popTime not match. {}, {}, {}, popTime:{}", topic, group, orderInfo, popTime); + return; + } + + if (orderInfo.getOffsetConsumedCount() != null) { + orderInfo.getOffsetConsumedCount().replaceAll((key, value) -> value > 0 ? value - 1 : value); + } + orderInfo.setOffsetNextVisibleTime(null); + orderInfo.setInvisibleTime(visibilityTimeout - orderInfo.getPopTime()); + updateLockFreeTimestamp(topic, group, queueId, orderInfo); + } + @Override public void persist() { // MemoryConsumerOrderInfoManager persist, do nothing. diff --git a/broker/src/main/java/org/apache/rocketmq/broker/pop/orderly/QueueLevelConsumerManager.java b/broker/src/main/java/org/apache/rocketmq/broker/pop/orderly/QueueLevelConsumerManager.java index 6cf5aabe44..6c57dd7ab4 100644 --- a/broker/src/main/java/org/apache/rocketmq/broker/pop/orderly/QueueLevelConsumerManager.java +++ b/broker/src/main/java/org/apache/rocketmq/broker/pop/orderly/QueueLevelConsumerManager.java @@ -42,7 +42,7 @@ import org.apache.rocketmq.store.GetMessageResult; public class QueueLevelConsumerManager extends ConfigManager implements ConsumerOrderInfoManager { - private static final Logger log = LoggerFactory.getLogger(LoggerName.BROKER_LOGGER_NAME); + protected static final Logger log = LoggerFactory.getLogger(LoggerName.BROKER_LOGGER_NAME); private static final String TOPIC_GROUP_SEPARATOR = "@"; private static final long CLEAN_SPAN_FROM_LAST = 24 * 3600 * 1000; diff --git a/broker/src/main/java/org/apache/rocketmq/broker/processor/ChangeInvisibleTimeProcessor.java b/broker/src/main/java/org/apache/rocketmq/broker/processor/ChangeInvisibleTimeProcessor.java index a8b01ceed2..5ff132ca23 100644 --- a/broker/src/main/java/org/apache/rocketmq/broker/processor/ChangeInvisibleTimeProcessor.java +++ b/broker/src/main/java/org/apache/rocketmq/broker/processor/ChangeInvisibleTimeProcessor.java @@ -25,6 +25,7 @@ import java.util.concurrent.TimeUnit; import org.apache.commons.lang3.StringUtils; import org.apache.rocketmq.broker.BrokerController; import org.apache.rocketmq.broker.offset.ConsumerOffsetManager; +import org.apache.rocketmq.broker.offset.MemoryConsumerOrderInfoManager; import org.apache.rocketmq.broker.pop.PopConsumerLockService; import org.apache.rocketmq.broker.pop.orderly.ConsumerOrderInfoManager; import org.apache.rocketmq.common.PopAckConstants; @@ -382,8 +383,8 @@ public class ChangeInvisibleTimeProcessor implements NettyRequestProcessor { long popTime = ExtraInfoUtil.getPopTime(extraInfo); ConsumerOffsetManager consumerOffsetManager = this.brokerController.getConsumerOffsetManager(); - ConsumerOrderInfoManager consumerOrderInfoManager = - brokerController.getPopLiteMessageProcessor().getConsumerOrderInfoManager(); + MemoryConsumerOrderInfoManager consumerOrderInfoManager = + (MemoryConsumerOrderInfoManager) brokerController.getPopLiteMessageProcessor().getConsumerOrderInfoManager(); PopConsumerLockService consumerLockService = this.brokerController.getPopLiteMessageProcessor().getLockService(); long oldOffset = consumerOffsetManager.queryOffset(group, lmqName, 0); @@ -400,9 +401,12 @@ public class ChangeInvisibleTimeProcessor implements NettyRequestProcessor { return CompletableFuture.completedFuture(response); } long visibilityTimeout = System.currentTimeMillis() + requestHeader.getInvisibleTime(); - consumerOrderInfoManager.updateNextVisibleTime( - lmqName, group, 0, requestHeader.getOffset(), popTime, visibilityTimeout); - + if (requestHeader.isSuspend()) { + consumerOrderInfoManager.suspendQueue(lmqName, group, 0, popTime, visibilityTimeout); + } else { + consumerOrderInfoManager.updateNextVisibleTime( + lmqName, group, 0, requestHeader.getOffset(), popTime, visibilityTimeout); + } responseHeader.setInvisibleTime(visibilityTimeout - popTime); responseHeader.setPopTime(popTime); responseHeader.setReviveQid(ExtraInfoUtil.getReviveQid(extraInfo)); diff --git a/broker/src/main/java/org/apache/rocketmq/broker/processor/LiteManagerProcessor.java b/broker/src/main/java/org/apache/rocketmq/broker/processor/LiteManagerProcessor.java index ac12983d61..cc3db586dd 100644 --- a/broker/src/main/java/org/apache/rocketmq/broker/processor/LiteManagerProcessor.java +++ b/broker/src/main/java/org/apache/rocketmq/broker/processor/LiteManagerProcessor.java @@ -19,6 +19,8 @@ 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 org.apache.commons.lang3.StringUtils; @@ -26,10 +28,12 @@ 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; import org.apache.rocketmq.common.constant.LoggerName; +import org.apache.rocketmq.common.entity.ClientGroup; import org.apache.rocketmq.common.lite.LiteLagInfo; import org.apache.rocketmq.common.lite.LiteSubscription; import org.apache.rocketmq.common.lite.LiteUtil; @@ -58,6 +62,8 @@ import org.apache.rocketmq.remoting.protocol.subscription.SubscriptionGroupConfi import java.util.HashSet; import java.util.Map; import java.util.Set; +import java.util.stream.Collectors; +import java.util.stream.Stream; public class LiteManagerProcessor implements NettyRequestProcessor { private static final Logger LOGGER = LoggerFactory.getLogger(LoggerName.ROCKETMQ_POP_LITE_LOGGER_NAME); @@ -190,7 +196,7 @@ public class LiteManagerProcessor implements NettyRequestProcessor { GetLiteTopicInfoResponseBody body = new GetLiteTopicInfoResponseBody(); body.setParentTopic(parentTopic); body.setLiteTopic(liteTopic); - body.setSubscriber(brokerController.getLiteSubscriptionRegistry().getSubscriber(lmqName)); + body.setSubscriber(getSubscriber(lmqName)); body.setTopicOffset(topicOffset); body.setShardingToBroker(brokerController.getBrokerConfig().getBrokerName().equals( liteSharding.shardingByLmqName(parentTopic, lmqName))); @@ -366,7 +372,7 @@ public class LiteManagerProcessor implements NettyRequestProcessor { } if (StringUtils.isNotEmpty(clientId)) { - brokerController.getLiteEventDispatcher().doFullDispatch(clientId, group); + brokerController.getLiteEventDispatcher().doFullDispatchForClient(clientId, group); } else { brokerController.getLiteEventDispatcher().doFullDispatchByGroup(group); } @@ -376,6 +382,25 @@ public class LiteManagerProcessor implements NettyRequestProcessor { return response; } + @VisibleForTesting + public Set getSubscriber(String lmqName) { + SubscriberWrapper.MapWrapper wrapper = + brokerController.getLiteSubscriptionRegistry().getAllSubscriber(null, lmqName).asMapWrapper(); + if (null == wrapper) { + return Collections.emptySet(); + } + return wrapper.getGroupMap().entrySet().stream() + .flatMap(entry -> { + String group = entry.getKey(); + if (LiteMetadataUtil.isWildcardGroup(group, brokerController)) { + return Stream.of(new ClientGroup("*", group)); + } else { + return entry.getValue().stream(); + } + }) + .collect(Collectors.toSet()); + } + @Override public boolean rejectRequest() { return false; diff --git a/broker/src/test/java/org/apache/rocketmq/broker/lite/AbstractLiteLifecycleManagerTest.java b/broker/src/test/java/org/apache/rocketmq/broker/lite/AbstractLiteLifecycleManagerTest.java index d5742ea3ee..5c1ab35cd3 100644 --- a/broker/src/test/java/org/apache/rocketmq/broker/lite/AbstractLiteLifecycleManagerTest.java +++ b/broker/src/test/java/org/apache/rocketmq/broker/lite/AbstractLiteLifecycleManagerTest.java @@ -22,6 +22,8 @@ import java.util.List; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ConcurrentMap; +import java.util.function.Function; +import org.apache.commons.lang3.tuple.Triple; import org.apache.rocketmq.broker.BrokerController; import org.apache.rocketmq.broker.config.v1.RocksDBConsumerOffsetManager; import org.apache.rocketmq.broker.pop.orderly.ConsumerOrderInfoManager; @@ -278,5 +280,10 @@ public class AbstractLiteLifecycleManagerTest { public List collectByParentTopic(String parentTopic) { return PARENT_TOPIC.equals(parentTopic) ? Collections.singletonList(EXIST_LMQ_NAME) : Collections.emptyList(); } + + @Override + public void forEachLiteTopic(Function, Boolean> function) { + + } } } diff --git a/broker/src/test/java/org/apache/rocketmq/broker/lite/LiteEventDispatcherTest.java b/broker/src/test/java/org/apache/rocketmq/broker/lite/LiteEventDispatcherTest.java index 1360ec5676..31d5562f92 100644 --- a/broker/src/test/java/org/apache/rocketmq/broker/lite/LiteEventDispatcherTest.java +++ b/broker/src/test/java/org/apache/rocketmq/broker/lite/LiteEventDispatcherTest.java @@ -17,27 +17,22 @@ package org.apache.rocketmq.broker.lite; -import com.google.common.cache.Cache; -import org.apache.commons.lang3.reflect.FieldUtils; import org.apache.rocketmq.broker.BrokerController; -import org.apache.rocketmq.broker.longpolling.PopLiteLongPollingService; import org.apache.rocketmq.broker.offset.ConsumerOffsetManager; -import org.apache.rocketmq.broker.pop.orderly.ConsumerOrderInfoManager; import org.apache.rocketmq.broker.processor.PopLiteMessageProcessor; import org.apache.rocketmq.broker.subscription.SubscriptionGroupManager; import org.apache.rocketmq.common.BrokerConfig; import org.apache.rocketmq.common.entity.ClientGroup; import org.apache.rocketmq.common.lite.LiteSubscription; import org.apache.rocketmq.common.lite.LiteUtil; -import org.junit.After; -import org.junit.Assert; +import org.apache.rocketmq.remoting.protocol.subscription.SubscriptionGroupConfig; import org.junit.Before; import org.junit.Test; +import org.junit.jupiter.api.extension.ExtendWith; import org.junit.runner.RunWith; import org.mockito.Mock; -import org.mockito.Mockito; -import org.mockito.junit.MockitoJUnitRunner; +import java.util.ArrayList; import java.util.Arrays; import java.util.Collections; import java.util.HashSet; @@ -45,537 +40,566 @@ import java.util.Iterator; import java.util.List; import java.util.Map; import java.util.Set; -import java.util.UUID; -import java.util.concurrent.ConcurrentMap; -import java.util.concurrent.ConcurrentSkipListSet; +import java.util.HashMap; -import static org.apache.rocketmq.broker.lite.LiteEventDispatcher.COMPARATOR; +import org.mockito.Mockito; +import org.mockito.junit.MockitoJUnitRunner; +import org.mockito.junit.jupiter.MockitoExtension; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertNull; +import static org.junit.Assert.assertTrue; +import static org.junit.Assert.fail; +import static org.mockito.ArgumentMatchers.any; +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.doReturn; +import static org.mockito.Mockito.mock; import static org.mockito.Mockito.never; -import static org.mockito.Mockito.times; +import static org.mockito.Mockito.spy; import static org.mockito.Mockito.verify; import static org.mockito.Mockito.when; - @RunWith(MockitoJUnitRunner.class) +@ExtendWith(MockitoExtension.class) public class LiteEventDispatcherTest { + private LiteEventDispatcher liteEventDispatcher; + @Mock private BrokerController brokerController; + @Mock private LiteSubscriptionRegistry liteSubscriptionRegistry; + @Mock private AbstractLiteLifecycleManager liteLifecycleManager; + @Mock private ConsumerOffsetManager consumerOffsetManager; - @Mock - private PopLiteMessageProcessor popLiteMessageProcessor; - @Mock - private PopLiteLongPollingService popLiteLongPollingService; - @Mock - private ConsumerOrderInfoManager consumerOrderInfoManager; + @Mock private SubscriptionGroupManager subscriptionGroupManager; - private BrokerConfig brokerConfig; - private LiteEventDispatcher liteEventDispatcher; - private ConcurrentMap clientEventMap; - private Cache blacklist; + private final BrokerConfig brokerConfig = new BrokerConfig(); - @SuppressWarnings("unchecked") @Before - public void setUp() throws IllegalAccessException { - brokerConfig = new BrokerConfig(); - when(brokerController.getBrokerConfig()).thenReturn(brokerConfig); + public void setUp() { when(brokerController.getConsumerOffsetManager()).thenReturn(consumerOffsetManager); - when(brokerController.getPopLiteMessageProcessor()).thenReturn(popLiteMessageProcessor); + when(brokerController.getBrokerConfig()).thenReturn(brokerConfig); when(brokerController.getSubscriptionGroupManager()).thenReturn(subscriptionGroupManager); - when(popLiteMessageProcessor.getPopLiteLongPollingService()).thenReturn(popLiteLongPollingService); - when(popLiteMessageProcessor.getConsumerOrderInfoManager()).thenReturn(consumerOrderInfoManager); - LiteEventDispatcher testObject = new LiteEventDispatcher(brokerController, liteSubscriptionRegistry, liteLifecycleManager); - liteEventDispatcher = Mockito.spy(testObject); + liteEventDispatcher = new LiteEventDispatcher(brokerController, liteSubscriptionRegistry, liteLifecycleManager); + PopLiteMessageProcessor popLiteMessageProcessor = new PopLiteMessageProcessor(brokerController, liteEventDispatcher); + when(brokerController.getPopLiteMessageProcessor()).thenReturn(popLiteMessageProcessor); + } + + @Test + public void testInitAddsListener() { liteEventDispatcher.init(); - - clientEventMap = (ConcurrentMap) - FieldUtils.readDeclaredField(testObject, "clientEventMap", true); - blacklist = (Cache) FieldUtils.readDeclaredField(testObject, "blacklist", true); - } - - @After - public void reset() { - brokerConfig = new BrokerConfig(); - clientEventMap.clear(); - blacklist.invalidateAll(); + verify(liteSubscriptionRegistry).addListener(any(LiteEventDispatcher.LiteCtlListenerImpl.class)); } @Test - public void testFullDispatchRequestComparator() { - LiteEventDispatcher.FullDispatchRequest request1 = - new LiteEventDispatcher.FullDispatchRequest("client1", "whatever", 1000); - LiteEventDispatcher.FullDispatchRequest request2 = - new LiteEventDispatcher.FullDispatchRequest("client2", "whatever", 2000); - LiteEventDispatcher.FullDispatchRequest request3 = - new LiteEventDispatcher.FullDispatchRequest("client1", "whatever", 1000); - - Assert.assertTrue(COMPARATOR.compare(request1, request2) < 0); - Assert.assertTrue(COMPARATOR.compare(request2, request1) > 0); - Assert.assertEquals(0, COMPARATOR.compare(request1, request3)); + public void testDispatchWhenEventModeDisabled() { + brokerConfig.setEnableLiteEventMode(false); + liteEventDispatcher.dispatch("group", "lmqName", 0, 0L, 0L); + verify(liteSubscriptionRegistry, never()).getAllSubscriber(anyString(), anyString()); } @Test - public void testFullDispatchSet() { - ConcurrentSkipListSet set = - new ConcurrentSkipListSet<>(COMPARATOR); - - LiteEventDispatcher.FullDispatchRequest request1 = - new LiteEventDispatcher.FullDispatchRequest("client1", "whatever", 1000); - LiteEventDispatcher.FullDispatchRequest request2 = - new LiteEventDispatcher.FullDispatchRequest("client2", "whatever", 2000); - LiteEventDispatcher.FullDispatchRequest request3 = - new LiteEventDispatcher.FullDispatchRequest("client1", "whatever", 1000); - LiteEventDispatcher.FullDispatchRequest request4 = - new LiteEventDispatcher.FullDispatchRequest("client3", "whatever", 500); - LiteEventDispatcher.FullDispatchRequest request5 = - new LiteEventDispatcher.FullDispatchRequest("client4", "whatever", 1000); - LiteEventDispatcher.FullDispatchRequest request6 = - new LiteEventDispatcher.FullDispatchRequest(null, "whatever", 1000); - - set.add(request1); - set.add(request3); - set.add(request6); - Assert.assertEquals(1, set.size()); - Assert.assertEquals(request1, set.pollFirst()); - - set.clear(); - set.add(request1); - set.add(request2); - set.add(request3); - set.add(request4); - set.add(request5); - Assert.assertEquals(4, set.size()); - Assert.assertEquals(request4, set.pollFirst()); - Assert.assertEquals(request1, set.pollFirst()); - Assert.assertEquals(request5, set.pollFirst()); - Assert.assertEquals(request2, set.pollFirst()); + public void testDispatchWhenQueueIdNotZero() { + brokerConfig.setEnableLiteEventMode(true); + liteEventDispatcher.dispatch("group", "lmqName", 1, 0L, 0L); + verify(liteSubscriptionRegistry, never()).getAllSubscriber(anyString(), anyString()); } @Test - public void testEventSetIterator() { - LiteEventDispatcher.ClientEventSet clientEventSet = liteEventDispatcher.new ClientEventSet("group"); - clientEventSet.offer("event1"); - clientEventSet.offer("event2"); - - LiteEventDispatcher.EventSetIterator iterator = new LiteEventDispatcher.EventSetIterator(clientEventSet); - - Assert.assertTrue(iterator.hasNext()); - Assert.assertEquals("event1", iterator.next()); - Assert.assertTrue(iterator.hasNext()); - Assert.assertEquals("event2", iterator.next()); - Assert.assertFalse(iterator.hasNext()); + public void testDispatchCallsDoDispatch() { + brokerConfig.setEnableLiteEventMode(true); + String lmqName = LiteUtil.toLmqName("parentTopic", "lmqName"); + LiteEventDispatcher spyDispatcher = Mockito.spy(liteEventDispatcher); + spyDispatcher.dispatch("group", lmqName, 0, 0L, 0L); + verify(spyDispatcher).doDispatch("group", lmqName, null); } @Test - public void testLiteSubscriptionIterator() { - Iterator topicIterator = Arrays.asList("event1", "event2").iterator(); + public void testDoDispatchWhenWrapperIsNull() { + brokerConfig.setEnableLiteEventMode(true); + when(liteSubscriptionRegistry.getAllSubscriber("group", "lmqName")).thenReturn(null); + + // Use reflection to access private method + try { + java.lang.reflect.Method method = LiteEventDispatcher.class.getDeclaredMethod( + "doDispatch", String.class, String.class, String.class); + method.setAccessible(true); + method.invoke(liteEventDispatcher, "group", "lmqName", null); + } catch (Exception e) { + fail("Exception should not be thrown"); + } + + verify(liteSubscriptionRegistry).getAllSubscriber("group", "lmqName"); + } + + @Test + public void testDoDispatchWithListWrapper() { + brokerConfig.setEnableLiteEventMode(true); + + SubscriptionGroupConfig subscriptionGroupConfig = new SubscriptionGroupConfig(); + subscriptionGroupConfig.setWildcardLiteGroup(false); + when(subscriptionGroupManager.findSubscriptionGroupConfig("group")).thenReturn(subscriptionGroupConfig); + + SubscriberWrapper.ListWrapper listWrapper = mock(SubscriberWrapper.ListWrapper.class); + List clients = Collections.singletonList(new ClientGroup("clientId", "group")); + when(listWrapper.asListWrapper()).thenReturn(listWrapper); + when(listWrapper.getClients()).thenReturn(clients); + when(liteSubscriptionRegistry.getAllSubscriber("group", "lmqName")).thenReturn(listWrapper); + + LiteEventDispatcher spyDispatcher = Mockito.spy(liteEventDispatcher); + spyDispatcher.doDispatch("group", "lmqName", null); + verify(spyDispatcher).selectAndDispatch("lmqName", clients, null); + } + + @Test + public void testDoDispatchWithMapWrapper() { + brokerConfig.setEnableLiteEventMode(true); + + SubscriberWrapper.MapWrapper mapWrapper = mock(SubscriberWrapper.MapWrapper.class); + Map> groupMap = new HashMap<>(); + groupMap.put("key", Collections.singletonList(new ClientGroup("clientId", "group"))); + when(mapWrapper.getGroupMap()).thenReturn(groupMap); + when(mapWrapper.asMapWrapper()).thenReturn(mapWrapper); + when(liteSubscriptionRegistry.getAllSubscriber("group", "lmqName")).thenReturn(mapWrapper); + + LiteEventDispatcher spyDispatcher = Mockito.spy(liteEventDispatcher); + + spyDispatcher.doDispatch("group", "lmqName", null); + + verify(spyDispatcher).selectAndDispatch(eq("lmqName"), anyList(), eq(null)); + } + + @Test + public void testSelectAndDispatchWhenClientsEmpty() { + List clients = new ArrayList<>(); + boolean result = liteEventDispatcher.selectAndDispatch("lmqName", clients, null); + assertTrue(result); + } + + @Test + public void testSelectAndDispatchWhenEventModeDisabled() { + brokerConfig.setEnableLiteEventMode(false); + List clients = Collections.singletonList(new ClientGroup("clientId", "group")); + boolean result = liteEventDispatcher.selectAndDispatch("lmqName", clients, null); + assertTrue(result); + } + + @Test + public void testSelectAndDispatchSelectsClientAndDispatches() { + brokerConfig.setEnableLiteEventMode(true); + List clients = Collections.singletonList(new ClientGroup("clientId", "group")); + + LiteEventDispatcher spyDispatcher = Mockito.spy(liteEventDispatcher); + doReturn(true).when(spyDispatcher).tryDispatchToClient(anyString(), anyString(), anyString(), anyBoolean()); + + boolean result = spyDispatcher.selectAndDispatch("lmqName", clients, null); + assertTrue(result); + verify(spyDispatcher).tryDispatchToClient("lmqName", "clientId", "group", true); + } + + @Test + public void testSelectAndDispatchExcludesSpecifiedClient() { + brokerConfig.setEnableLiteEventMode(true); + List clients = Arrays.asList( + new ClientGroup("excludeId", "group"), + new ClientGroup("clientId", "group") + ); + + LiteEventDispatcher spyDispatcher = Mockito.spy(liteEventDispatcher); + doReturn(true).when(spyDispatcher).tryDispatchToClient(anyString(), anyString(), anyString(), anyBoolean()); + + boolean result = spyDispatcher.selectAndDispatch("lmqName", clients, "excludeId"); + assertTrue(result); + verify(spyDispatcher).tryDispatchToClient("lmqName", "clientId", "group", true); + verify(spyDispatcher, never()).tryDispatchToClient("lmqName", "excludeId", "group", true); + } + + @Test + public void testTryDispatchToClientWhenQueueHasSpace() { + String clientId = "clientId"; + String group = "group"; + String lmqName = "lmqName"; + + // Create a real ClientEventSet for testing + LiteEventDispatcher.ClientEventSet eventSet = liteEventDispatcher.new ClientEventSet(group); + + // Mock the clientEventMap to return our eventSet + liteEventDispatcher.clientEventMap.put(clientId, eventSet); + + boolean result = liteEventDispatcher.tryDispatchToClient(lmqName, clientId, group, true); + assertTrue(result); + assertEquals(1, eventSet.size()); + } + + @Test + public void testTryDispatchToClientWhenQueueIsFull() { + String clientId = "clientId"; + String group = "group"; + String lmqName = "lmqName"; + + // Create a ClientEventSet with capacity 1 + LiteEventDispatcher.ClientEventSet eventSet = mock(LiteEventDispatcher.ClientEventSet.class); + when(eventSet.offer(lmqName)).thenReturn(false); + + liteEventDispatcher.clientEventMap.put(clientId, eventSet); + + boolean result = liteEventDispatcher.tryDispatchToClient(lmqName, clientId, group, true); + assertFalse(result); + verify(eventSet).offer(lmqName); + } + + @Test + public void testGetEventIteratorInEventMode() { + brokerConfig.setEnableLiteEventMode(true); + String clientId = "clientId"; + String group = "group"; + + LiteEventDispatcher.ClientEventSet eventSet = liteEventDispatcher.new ClientEventSet(group); + liteEventDispatcher.clientEventMap.put(clientId, eventSet); + + Iterator iterator = liteEventDispatcher.getEventIterator(clientId); + assertNotNull(iterator); + assertFalse(iterator.hasNext()); + } + + @Test + public void testGetEventIteratorWhenNotInEventMode() { + brokerConfig.setEnableLiteEventMode(false); + String clientId = "clientId"; + LiteSubscription subscription = mock(LiteSubscription.class); + Set topicSet = new HashSet<>(); + topicSet.add("topic1"); + when(subscription.getLiteTopicSet()).thenReturn(topicSet); + when(liteSubscriptionRegistry.getLiteSubscription(clientId)).thenReturn(subscription); + + Iterator iterator = liteEventDispatcher.getEventIterator(clientId); + assertNotNull(iterator); + assertTrue(iterator.hasNext()); + assertEquals("topic1", iterator.next()); + } + + @Test + public void testDoFullDispatchForClientWhenSubscriptionIsNull() { + brokerConfig.setEnableLiteEventMode(true); + String clientId = "clientId"; + String group = "group"; + + when(liteSubscriptionRegistry.getLiteSubscription(clientId)).thenReturn(null); + + liteEventDispatcher.doFullDispatchForClient(clientId, group); + verify(liteSubscriptionRegistry).getLiteSubscription(clientId); + } + + @Test + public void testDoFullDispatchForClientWhenSubscriptionHasNoTopics() { + brokerConfig.setEnableLiteEventMode(true); + String clientId = "clientId"; + String group = "group"; + + LiteSubscription subscription = mock(LiteSubscription.class); + when(subscription.getLiteTopicSet()).thenReturn(Collections.emptySet()); + when(liteSubscriptionRegistry.getLiteSubscription(clientId)).thenReturn(subscription); + + liteEventDispatcher.doFullDispatchForClient(clientId, group); + verify(liteSubscriptionRegistry).getLiteSubscription(clientId); + } + + @Test + public void testScheduleFullDispatchForClientAddsRequestToSet() { + String clientId = "clientId"; + String group = "group"; + long delayTime = 1000L; + + liteEventDispatcher.scheduleFullDispatchForClient(clientId, group, delayTime); + + assertEquals(1, liteEventDispatcher.fullDispatchSet.size()); + assertEquals(1, liteEventDispatcher.fullDispatchMap.size()); + assertTrue(liteEventDispatcher.fullDispatchMap.containsKey(clientId)); + } + + @Test + public void testScheduleFullDispatchForClientDoesNotAddDuplicate() { + String clientId = "clientId"; + String group = "group"; + long delayTime = 1000L; + + liteEventDispatcher.scheduleFullDispatchForClient(clientId, group, delayTime); + liteEventDispatcher.scheduleFullDispatchForClient(clientId, group, delayTime); + + assertEquals(1, liteEventDispatcher.fullDispatchSet.size()); + assertEquals(1, liteEventDispatcher.fullDispatchMap.size()); + } + + @Test + public void testScheduleFullDispatchForWildcardGroup() { + String group = "group"; + long delayTime = 1000L; + + LiteEventDispatcher spyDispatcher = Mockito.spy(liteEventDispatcher); + spyDispatcher.scheduleFullDispatchForWildcardGroup(group, delayTime); + + verify(spyDispatcher).scheduleFullDispatchForClient("$group$", group, delayTime); + } + + @Test + public void testClientEventSetOffer() { + String group = "group"; + LiteEventDispatcher.ClientEventSet eventSet = liteEventDispatcher.new ClientEventSet(group); + + boolean result = eventSet.offer("event"); + assertTrue(result); + assertEquals(1, eventSet.size()); + } + + @Test + public void testClientEventSetPoll() { + String group = "group"; + LiteEventDispatcher.ClientEventSet eventSet = liteEventDispatcher.new ClientEventSet(group); + + eventSet.offer("event"); + String result = eventSet.poll(); + assertEquals("event", result); + assertEquals(0, eventSet.size()); + } + + @Test + public void testClientEventSetMaybeBlock() { + String group = "group"; + LiteEventDispatcher.ClientEventSet eventSet = liteEventDispatcher.new ClientEventSet(group); + + // Initially should not block + assertFalse(eventSet.maybeBlock()); + + // After adding an event and waiting, should block + eventSet.offer("event"); + // Simulate time passing by manipulating lastAccessTime + try { + // Use reflection to access private field + java.lang.reflect.Field lastAccessTimeField = + LiteEventDispatcher.ClientEventSet.class.getDeclaredField("lastAccessTime"); + lastAccessTimeField.setAccessible(true); + lastAccessTimeField.setLong(eventSet, System.currentTimeMillis() - + LiteEventDispatcher.CLIENT_LONG_POLLING_INTERVAL - 1000); + } catch (Exception e) { + fail("Failed to manipulate lastAccessTime"); + } + + assertTrue(eventSet.maybeBlock()); + } + + @Test + public void testClientEventSetIsLowWaterMark() { + String group = "group"; + LiteEventDispatcher.ClientEventSet eventSet = liteEventDispatcher.new ClientEventSet(group); + + // Empty queue should be low water mark + assertTrue(eventSet.isLowWaterMark()); + + // Add events to exceed low water mark + for (int i = 0; i < (int) (LiteEventDispatcher.LOW_WATER_MARK * 100) + 1; i++) { + eventSet.offer("event" + i); + } + + // Should no longer be low water mark + assertFalse(eventSet.isLowWaterMark()); + } + + @Test + public void testClientEventSetIsActiveConsuming() { + String group = "group"; + LiteEventDispatcher.ClientEventSet eventSet = liteEventDispatcher.new ClientEventSet(group); + + // Initially should be active consuming + assertTrue(eventSet.isActiveConsuming()); + + // Simulate time passing + try { + java.lang.reflect.Field lastAccessTimeField = + LiteEventDispatcher.ClientEventSet.class.getDeclaredField("lastAccessTime"); + lastAccessTimeField.setAccessible(true); + lastAccessTimeField.setLong(eventSet, System.currentTimeMillis() - + LiteEventDispatcher.ACTIVE_CONSUMING_WINDOW - 1000); + } catch (Exception e) { + fail("Failed to manipulate lastAccessTime"); + } + + // Should no longer be active consuming + assertFalse(eventSet.isActiveConsuming()); + } + + @Test + public void testEventSetIteratorHasNextAndNext() { + String group = "group"; + LiteEventDispatcher.ClientEventSet eventSet = liteEventDispatcher.new ClientEventSet(group); + eventSet.offer("event1"); + eventSet.offer("event2"); + + LiteEventDispatcher.EventSetIterator iterator = new LiteEventDispatcher.EventSetIterator(eventSet); + + assertTrue(iterator.hasNext()); + assertEquals("event1", iterator.next()); + assertTrue(iterator.hasNext()); + assertEquals("event2", iterator.next()); + assertFalse(iterator.hasNext()); + } + + @Test + public void testLiteSubscriptionIteratorHasNextAndNext() { + Set topics = new HashSet<>(); + topics.add("topic1"); + topics.add("topic2"); + Iterator topicIterator = topics.iterator(); LiteEventDispatcher.LiteSubscriptionIterator iterator = new LiteEventDispatcher.LiteSubscriptionIterator("parentTopic", topicIterator); - Assert.assertTrue(iterator.hasNext()); - Assert.assertEquals("event1", iterator.next()); - Assert.assertTrue(iterator.hasNext()); - Assert.assertEquals("event2", iterator.next()); - Assert.assertFalse(iterator.hasNext()); + assertTrue(iterator.hasNext()); + assertNotNull(iterator.next()); + assertTrue(iterator.hasNext()); + assertNotNull(iterator.next()); + assertFalse(iterator.hasNext()); } @Test - public void testClientEventSet_offerAndPoll() { - brokerConfig.setMaxClientEventCount(3); - LiteEventDispatcher.ClientEventSet clientEventSet = liteEventDispatcher.new ClientEventSet("group"); - - Assert.assertTrue(clientEventSet.offer("event1")); - Assert.assertTrue(clientEventSet.offer("event2")); - Assert.assertTrue(clientEventSet.offer("event1")); - Assert.assertTrue(clientEventSet.offer("event3")); - Assert.assertFalse(clientEventSet.offer("event4")); - - Assert.assertEquals(3, clientEventSet.size()); - Assert.assertEquals("event1", clientEventSet.poll()); - Assert.assertEquals("event2", clientEventSet.poll()); - Assert.assertEquals("event3", clientEventSet.poll()); - Assert.assertEquals(0, clientEventSet.size()); - Assert.assertNull(clientEventSet.poll()); - } - - @Test - public void testClientEventSet_isLowWaterMark() { - brokerConfig.setMaxClientEventCount(10); - LiteEventDispatcher.ClientEventSet clientEventSet = liteEventDispatcher.new ClientEventSet("group"); - Assert.assertTrue(clientEventSet.isLowWaterMark()); - - for (int i = 0; i < 4; i++) { - clientEventSet.offer("event" + i); - } - Assert.assertFalse(clientEventSet.isLowWaterMark()); - } - - @Test - public void testClientEventSetMaybeBlock() throws Exception { - LiteEventDispatcher.ClientEventSet clientEventSet = liteEventDispatcher.new ClientEventSet("group"); - Assert.assertFalse(clientEventSet.maybeBlock()); - - clientEventSet.offer("event"); - FieldUtils.writeDeclaredField(clientEventSet, "lastAccessTime", 0L, true); - Assert.assertTrue(clientEventSet.maybeBlock()); - clientEventSet.poll(); - Assert.assertFalse(clientEventSet.maybeBlock()); - } - - @Test - public void testGetAllSubscriber_noSubscribers() { - when(liteSubscriptionRegistry.getSubscriber("event")).thenReturn(null); - Object result = liteEventDispatcher.getAllSubscriber("group", "event"); - Assert.assertNull(result); - } - - @Test - @SuppressWarnings("unchecked") - public void testGetAllSubscriber_singleSubscriber() { - Set subscribers = new HashSet<>(); - subscribers.add(new ClientGroup("clientId", "group")); - when(liteSubscriptionRegistry.getSubscriber("event")).thenReturn(subscribers); - - Object result = liteEventDispatcher.getAllSubscriber("group", "event"); // specified - Assert.assertTrue(result instanceof List); - Assert.assertEquals(1, ((List) result).size()); - Assert.assertEquals("clientId", ((List) result).get(0).clientId); - - result = liteEventDispatcher.getAllSubscriber(null, "event"); // not specified - Assert.assertTrue(result instanceof List); - Assert.assertEquals(1, ((List) result).size()); - Assert.assertEquals("clientId", ((List) result).get(0).clientId); - - result = liteEventDispatcher.getAllSubscriber("otherGroup", "event"); // specified but not match - Assert.assertNull(result); - } - - @Test - @SuppressWarnings("unchecked") - public void testGetAllSubscriber_multipleSubscribers() { - Set subscribers = new HashSet<>(); - subscribers.add(new ClientGroup("clientId1", "group1")); - subscribers.add(new ClientGroup("clientId2", "group1")); - subscribers.add(new ClientGroup("clientId3", "group2")); - when(liteSubscriptionRegistry.getSubscriber("event")).thenReturn(subscribers); - - Object result = liteEventDispatcher.getAllSubscriber("group1", "event"); // specified - Assert.assertTrue(result instanceof List); - Assert.assertEquals(2, ((List) result).size()); - Assert.assertEquals("clientId1", ((List) result).get(0).clientId); - - result = liteEventDispatcher.getAllSubscriber("group2", "event"); // specified - Assert.assertTrue(result instanceof List); - Assert.assertEquals(1, ((List) result).size()); - Assert.assertEquals("clientId3", ((List) result).get(0).clientId); - - result = liteEventDispatcher.getAllSubscriber("otherGroup", "event"); // specified but not match - Assert.assertNull(result); - - result = liteEventDispatcher.getAllSubscriber(null, "event"); // not specified - Assert.assertTrue(result instanceof Map); - Assert.assertEquals(2, ((Map) result).size()); - Assert.assertEquals(2, ((Map>) result).get("group1").size()); - Assert.assertEquals(1, ((Map>) result).get("group2").size()); - } - - @Test - public void testTryDispatchToClient() { - brokerConfig.setMaxClientEventCount(1); - String clientId = "clientId"; - - boolean result = liteEventDispatcher.tryDispatchToClient("event1", clientId, "group"); - Assert.assertTrue(result); - - // not in blacklist - result = liteEventDispatcher.tryDispatchToClient("event2", clientId, "group"); - Assert.assertFalse(result); - verify(liteEventDispatcher).scheduleFullDispatch(clientId, "group", false); - - // in blacklist - blacklist.put(clientId, Boolean.TRUE); - result = liteEventDispatcher.tryDispatchToClient("event3", clientId, "group"); - Assert.assertFalse(result); - verify(liteEventDispatcher).scheduleFullDispatch(clientId, "group", true); - - blacklist.invalidate(clientId); - result = liteEventDispatcher.tryDispatchToClient("event3", clientId, "group"); - Assert.assertFalse(result); - verify(liteEventDispatcher, times(2)).scheduleFullDispatch(clientId, "group", false); - } - - @Test - public void testSelectAndDispatch_empty_or_singleClient() { - List clients = Collections.singletonList(new ClientGroup("client", "group")); - // disable event mode - brokerConfig.setEnableLiteEventMode(false); - liteEventDispatcher.selectAndDispatch("event", clients, null); - verify(liteEventDispatcher, never()).tryDispatchToClient(anyString(), anyString(), anyString()); - - // empty list - liteEventDispatcher.selectAndDispatch("event", Collections.emptyList(), null); - verify(liteEventDispatcher, never()).tryDispatchToClient(anyString(), anyString(), anyString()); - - // event mode - brokerConfig.setMaxClientEventCount(2); - brokerConfig.setEnableLiteEventMode(true); - - liteEventDispatcher.selectAndDispatch("event1", clients, null); - liteEventDispatcher.selectAndDispatch("event2", clients, "client"); // exclude - liteEventDispatcher.selectAndDispatch("event3", clients, null); - verify(popLiteLongPollingService, times(2)).notifyMessageArriving("client", true, 0, "group"); - } - - @Test - public void testSelectAndDispatch_multipleClients() { - brokerConfig.setMaxClientEventCount(2); - String client1 = UUID.randomUUID().toString(); - String client2 = UUID.randomUUID().toString(); - List clients = Arrays.asList( - new ClientGroup(client1, "group"), - new ClientGroup(client2, "group")); - - // no fallback - liteEventDispatcher.selectAndDispatch("event1", clients, client1); - verify(popLiteLongPollingService).notifyMessageArriving(client2, true, 0, "group"); - - // no fallback - liteEventDispatcher.selectAndDispatch("event2", clients, client2); - verify(popLiteLongPollingService).notifyMessageArriving(client1, true, 0, "group"); - - // fallback - blacklist.put(client1, Boolean.TRUE); - liteEventDispatcher.selectAndDispatch("event3", clients, null); - verify(popLiteLongPollingService, times(2)).notifyMessageArriving(client2, true, 0, "group"); - - // fallback - blacklist.invalidate(client1); - blacklist.put(client2, Boolean.TRUE); - liteEventDispatcher.selectAndDispatch("event4", clients, null); - verify(popLiteLongPollingService, times(2)).notifyMessageArriving(client1, true, 0, "group"); - - // queue all full - liteEventDispatcher.selectAndDispatch("event5", clients, null); - verify(popLiteLongPollingService, times(2)).notifyMessageArriving(client1, true, 0, "group"); - verify(popLiteLongPollingService, times(2)).notifyMessageArriving(client2, true, 0, "group"); - } - - @Test - public void testDispatch() { - // disable event mode - brokerConfig.setEnableLiteEventMode(false); - liteEventDispatcher.dispatch("group", "event", 0, 0, System.currentTimeMillis()); - verify(liteEventDispatcher, never()).getAllSubscriber(anyString(), anyString()); - - // event mode - brokerConfig.setEnableLiteEventMode(true); - liteEventDispatcher.dispatch("group", "event", 1, 0, System.currentTimeMillis()); // queue id not match - liteEventDispatcher.dispatch("group", "event", 0, 0, System.currentTimeMillis()); // queue name not match - verify(liteEventDispatcher, never()).getAllSubscriber(anyString(), anyString()); - - // do dispatch - liteEventDispatcher.dispatch("group", LiteUtil.toLmqName("p", "l"), 0, 0, System.currentTimeMillis()); - verify(liteEventDispatcher).getAllSubscriber(anyString(), anyString()); - } - - @Test - public void testDoFullDispatch_disable_or_emptySubscription() { - String clientId = "clientId"; - String group = "group"; - - // disable event mode - brokerConfig.setEnableLiteEventMode(false); - liteEventDispatcher.doFullDispatch(clientId, group); - verify(liteSubscriptionRegistry, never()).getLiteSubscription(clientId); - - // empty subscription - brokerConfig.setEnableLiteEventMode(true); - when(liteSubscriptionRegistry.getLiteSubscription("clientId")).thenReturn(null); - liteEventDispatcher.doFullDispatch(clientId, group); - verify(liteLifecycleManager, never()).getMaxOffsetInQueue(anyString()); - } - - @Test - public void testDoFullDispatch_maybeBlock() throws Exception { - int num = 10; - String clientId = "clientId"; - String group = "group"; - LiteSubscription subscription = new LiteSubscription(); - subscription.setTopic("parentTopic"); - for (int i = 0; i < num; i++) { - subscription.addLiteTopic(LiteUtil.toLmqName(subscription.getTopic(), "l" + i)); - } - when(liteSubscriptionRegistry.getLiteSubscription(clientId)).thenReturn(subscription); - - // maybe block - liteEventDispatcher.tryDispatchToClient("event", clientId, group); - Assert.assertNotNull(clientEventMap.get(clientId)); - FieldUtils.writeDeclaredField(clientEventMap.get(clientId), "lastAccessTime", 0L, true); - liteEventDispatcher.doFullDispatch(clientId, group); - verify(liteEventDispatcher).scheduleFullDispatch(clientId, group, true); - verify(liteLifecycleManager, never()).getMaxOffsetInQueue(anyString()); - } - - @Test - public void testDoFullDispatch_highWaterMark() throws Exception { - int num = 10; - String clientId = "clientId"; - String group = "group"; - LiteSubscription subscription = new LiteSubscription(); - subscription.setTopic("parentTopic"); - for (int i = 0; i < num; i++) { - subscription.addLiteTopic(LiteUtil.toLmqName(subscription.getTopic(), "l" + i)); - } - when(liteSubscriptionRegistry.getLiteSubscription(clientId)).thenReturn(subscription); - - brokerConfig.setMaxClientEventCount(1); - - // active consuming - liteEventDispatcher.tryDispatchToClient("event", clientId, group); - liteEventDispatcher.doFullDispatch(clientId, group); - - verify(liteEventDispatcher).scheduleFullDispatch(clientId, group, false); - verify(liteLifecycleManager, never()).getMaxOffsetInQueue(anyString()); - - // not active consuming - clientEventMap.clear(); - liteEventDispatcher.tryDispatchToClient("event", clientId, group); - FieldUtils.writeDeclaredField(clientEventMap.get(clientId), "lastAccessTime", System.currentTimeMillis() - 6000L, true); - liteEventDispatcher.doFullDispatch(clientId, group); - - verify(liteEventDispatcher).scheduleFullDispatch(clientId, group, true); - verify(liteLifecycleManager, never()).getMaxOffsetInQueue(anyString()); - } - - @Test - public void testDoFullDispatch_multipleTopics() { - String clientId = "clientId"; - String group = "group"; - - String lmqName1 = "lmqName1"; - String lmqName2 = "lmqName2"; - String lmqName3 = "lmqName2"; - LiteSubscription subscription = new LiteSubscription(); - subscription.setTopic("parentTopic"); - subscription.addLiteTopic(lmqName1); - subscription.addLiteTopic(lmqName2); - subscription.addLiteTopic(lmqName3); - when(liteSubscriptionRegistry.getLiteSubscription(clientId)).thenReturn(subscription); - - - when(liteLifecycleManager.getMaxOffsetInQueue(lmqName1)).thenReturn(0L); - - when(liteLifecycleManager.getMaxOffsetInQueue(lmqName2)).thenReturn(10L); - when(consumerOffsetManager.queryOffset(group, lmqName2, 0)).thenReturn(10L); - - when(liteLifecycleManager.getMaxOffsetInQueue(lmqName3)).thenReturn(10L); - when(consumerOffsetManager.queryOffset(group, lmqName3, 0)).thenReturn(5L); - - liteEventDispatcher.doFullDispatch(clientId, group); - - verify(liteLifecycleManager).getMaxOffsetInQueue(lmqName1); - verify(liteLifecycleManager).getMaxOffsetInQueue(lmqName2); - verify(liteLifecycleManager).getMaxOffsetInQueue(lmqName3); - verify(consumerOffsetManager, never()).queryOffset(group, lmqName1, 0); - verify(consumerOffsetManager).queryOffset(group, lmqName2, 0); - verify(consumerOffsetManager).queryOffset(group, lmqName3, 0); - - verify(liteEventDispatcher, never()).scheduleFullDispatch(clientId, group, true); - verify(popLiteLongPollingService, times(2)).notifyMessageArriving(clientId, true, 0, group); - } - - @Test - public void testDoFullDispatch_eventQueueFull() throws IllegalAccessException { - brokerConfig.setMaxClientEventCount(2); - String clientId = "clientId"; - String group = "group"; - - String lmqName1 = "lmqName1"; - String lmqName2 = "lmqName2"; - String lmqName3 = "lmqName3"; - LiteSubscription subscription = new LiteSubscription(); - subscription.setTopic("parentTopic"); - subscription.addLiteTopic(lmqName1); - subscription.addLiteTopic(lmqName2); - subscription.addLiteTopic(lmqName3); - when(liteSubscriptionRegistry.getLiteSubscription(clientId)).thenReturn(subscription); - - when(liteLifecycleManager.getMaxOffsetInQueue(lmqName1)).thenReturn(10L); - when(consumerOffsetManager.queryOffset(group, lmqName1, 0)).thenReturn(5L); - - when(liteLifecycleManager.getMaxOffsetInQueue(lmqName2)).thenReturn(10L); - when(consumerOffsetManager.queryOffset(group, lmqName2, 0)).thenReturn(5L); - - when(liteLifecycleManager.getMaxOffsetInQueue(lmqName3)).thenReturn(10L); - when(consumerOffsetManager.queryOffset(group, lmqName3, 0)).thenReturn(5L); - - // active consuming - liteEventDispatcher.doFullDispatch(clientId, group); - verify(liteEventDispatcher).scheduleFullDispatch(clientId, group, false); - verify(popLiteLongPollingService, times(2)).notifyMessageArriving(clientId, true, 0, group); - Assert.assertNotNull(clientEventMap.get(clientId).poll()); - Assert.assertNotNull(clientEventMap.get(clientId).poll()); - - // not active consuming - FieldUtils.writeDeclaredField(clientEventMap.get(clientId), "lastAccessTime", System.currentTimeMillis() - 6000L, true); - liteEventDispatcher.doFullDispatch(clientId, group); - verify(liteEventDispatcher).scheduleFullDispatch(clientId, group, true); - verify(popLiteLongPollingService, times(4)).notifyMessageArriving(clientId, true, 0, group); - } - - @Test - public void testDoFullDispatchByGroup() { - String group = "group"; - String clientId1 = "client1"; - String clientId2 = "client2"; - List clientIds = Arrays.asList(clientId1, clientId2); - Mockito.when(liteSubscriptionRegistry.getAllClientIdByGroup(group)).thenReturn(clientIds); - - liteEventDispatcher.doFullDispatchByGroup(group); - - verify(liteSubscriptionRegistry, times(1)).getAllClientIdByGroup(group); - verify(liteEventDispatcher, times(1)).doFullDispatch(clientId1, group); - verify(liteEventDispatcher, times(1)).doFullDispatch(clientId2, group); - } - - @Test - public void testScan() throws Exception { - String clientId = "clientId"; - String group = "group"; - String event = "event"; - liteEventDispatcher.tryDispatchToClient(event, clientId, group); - - Assert.assertNotNull(clientEventMap.get(clientId)); - FieldUtils.writeDeclaredField(clientEventMap.get(clientId), "lastAccessTime", 0L, true); - liteEventDispatcher.scan(); - verify(liteEventDispatcher).getAllSubscriber(group, event); - } - - @Test - public void testFullDispatchDeduplication() throws InterruptedException { + public void testComparatorComparesTimestampsCorrectly() { String clientId1 = "clientId1"; String clientId2 = "clientId2"; String group = "group"; - brokerConfig.setLiteEventFullDispatchDelayTime(10L); - liteEventDispatcher.scheduleFullDispatch(clientId1, group, false); - liteEventDispatcher.scheduleFullDispatch(clientId1, group, false); - liteEventDispatcher.scheduleFullDispatch(clientId1, group, false); - liteEventDispatcher.scheduleFullDispatch(clientId1, group, false); - liteEventDispatcher.scheduleFullDispatch(clientId2, group, false); - Thread.sleep(20L); - liteEventDispatcher.scan(); - verify(liteEventDispatcher, times(1)).doFullDispatch(clientId1, group); - verify(liteEventDispatcher, times(1)).doFullDispatch(clientId2, group); + LiteEventDispatcher.FullDispatchRequest request1 = + new LiteEventDispatcher.FullDispatchRequest(clientId1, group, 1000L); + LiteEventDispatcher.FullDispatchRequest request2 = + new LiteEventDispatcher.FullDispatchRequest(clientId2, group, 2000L); + + assertTrue(LiteEventDispatcher.COMPARATOR.compare(request1, request2) < 0); + assertTrue(LiteEventDispatcher.COMPARATOR.compare(request2, request1) > 0); + assertEquals(0, LiteEventDispatcher.COMPARATOR.compare(request1, request1)); } -} + + @Test + public void testLiteCtlListenerImplOnRegisterForWildcardGroup() throws NoSuchFieldException, IllegalAccessException { + SubscriptionGroupConfig subscriptionGroupConfig = new SubscriptionGroupConfig(); + subscriptionGroupConfig.setWildcardLiteGroup(true); + when(subscriptionGroupManager.findSubscriptionGroupConfig("group")).thenReturn(subscriptionGroupConfig); + + LiteEventDispatcher.LiteCtlListenerImpl listener = + liteEventDispatcher.new LiteCtlListenerImpl(); + + LiteEventDispatcher spyDispatcher = Mockito.spy(liteEventDispatcher); + + // Replace the dispatcher in the listener + java.lang.reflect.Field outerField = listener.getClass().getDeclaredField("this$0"); + outerField.setAccessible(true); + outerField.set(listener, spyDispatcher); + + listener.onRegister("clientId", "group", "lmqName"); + + verify(spyDispatcher).scheduleFullDispatchForWildcardGroup("group", 5000L); + } + + @Test + public void testLiteCtlListenerImplOnRegisterForRegularGroupWithExistingLMQ() throws NoSuchFieldException, IllegalAccessException { + when(liteLifecycleManager.isLmqExist("lmqName")).thenReturn(true); + + LiteEventDispatcher.LiteCtlListenerImpl listener = + liteEventDispatcher.new LiteCtlListenerImpl(); + + LiteEventDispatcher spyDispatcher = Mockito.spy(liteEventDispatcher); + + // Replace the dispatcher in the listener + java.lang.reflect.Field outerField = listener.getClass().getDeclaredField("this$0"); + outerField.setAccessible(true); + outerField.set(listener, spyDispatcher); + + listener.onRegister("clientId", "group", "lmqName"); + + verify(spyDispatcher).doDispatch("group", "lmqName", null); + + } + + @Test + public void testLiteCtlListenerImplOnRemoveAllRemovesClientAndRedispatchesEvents() { + String clientId = "clientId"; + String group = "group"; + + // Add a client event set with an event + LiteEventDispatcher.ClientEventSet eventSet = liteEventDispatcher.new ClientEventSet(group); + eventSet.offer("lmqName"); + liteEventDispatcher.clientEventMap.put(clientId, eventSet); + + LiteEventDispatcher.LiteCtlListenerImpl listener = + liteEventDispatcher.new LiteCtlListenerImpl(); + + LiteEventDispatcher spyDispatcher = Mockito.spy(liteEventDispatcher); + + // Replace the dispatcher in the listener + try { + java.lang.reflect.Field outerField = listener.getClass().getDeclaredField("this$0"); + outerField.setAccessible(true); + outerField.set(listener, spyDispatcher); + + listener.onRemoveAll(clientId, group); + + // Verify client was removed + assertNull(liteEventDispatcher.clientEventMap.get(clientId)); + + // Verify doDispatch was called + verify(spyDispatcher).doDispatch(group, "lmqName", clientId); + } catch (Exception e) { + fail("Exception should not be thrown: " + e.getMessage()); + } + } + + @Test + public void testDoFullDispatchForClientNormalCase() { + String clientId = "testClientId"; + String group = "testGroup"; + String lmqName = "testLmq"; + brokerConfig.setEnableLiteEventMode(true); + + LiteSubscription subscription = new LiteSubscription(); + Set topics = new HashSet<>(); + topics.add(lmqName); + subscription.setLiteTopicSet(topics); + + when(liteSubscriptionRegistry.getLiteSubscription(clientId)).thenReturn(subscription); + when(liteLifecycleManager.getMaxOffsetInQueue(lmqName)).thenReturn(100L); + when(consumerOffsetManager.queryOffset(group, lmqName, 0)).thenReturn(50L); + + LiteEventDispatcher.ClientEventSet eventSet = spy(liteEventDispatcher.new ClientEventSet(group)); + when(eventSet.maybeBlock()).thenReturn(false); + when(eventSet.isLowWaterMark()).thenReturn(true); + when(eventSet.offer(lmqName)).thenReturn(true); + + liteEventDispatcher.clientEventMap.put(clientId, eventSet); + + liteEventDispatcher.doFullDispatchForClient(clientId, group); + + verify(liteSubscriptionRegistry).getLiteSubscription(clientId); + verify(liteLifecycleManager).getMaxOffsetInQueue(lmqName); + verify(consumerOffsetManager).queryOffset(group, lmqName, 0); + verify(eventSet).offer(lmqName); + } + + @Test + public void testScan_FullDispatch() { + LiteEventDispatcher.FullDispatchRequest request = + new LiteEventDispatcher.FullDispatchRequest("testClientId", "testGroup", -1000); + liteEventDispatcher.fullDispatchSet.add(request); + liteEventDispatcher.scan(); + assertTrue(liteEventDispatcher.fullDispatchSet.isEmpty()); + } +} \ No newline at end of file diff --git a/broker/src/test/java/org/apache/rocketmq/broker/lite/LiteSubscriptionRegistryImplTest.java b/broker/src/test/java/org/apache/rocketmq/broker/lite/LiteSubscriptionRegistryImplTest.java index bf300ef4d9..0a555e6e4b 100644 --- a/broker/src/test/java/org/apache/rocketmq/broker/lite/LiteSubscriptionRegistryImplTest.java +++ b/broker/src/test/java/org/apache/rocketmq/broker/lite/LiteSubscriptionRegistryImplTest.java @@ -18,783 +18,587 @@ package org.apache.rocketmq.broker.lite; import io.netty.channel.Channel; +import java.util.Collections; import java.util.HashSet; import java.util.List; import java.util.Set; import java.util.concurrent.ConcurrentHashMap; import org.apache.rocketmq.broker.BrokerController; +import org.apache.rocketmq.broker.client.net.Broker2Client; import org.apache.rocketmq.broker.offset.ConsumerOffsetManager; import org.apache.rocketmq.broker.pop.orderly.QueueLevelConsumerManager; import org.apache.rocketmq.broker.processor.PopLiteMessageProcessor; import org.apache.rocketmq.broker.subscription.SubscriptionGroupManager; import org.apache.rocketmq.common.BrokerConfig; -import org.apache.rocketmq.common.attribute.LiteSubModel; import org.apache.rocketmq.common.entity.ClientGroup; import org.apache.rocketmq.common.lite.LiteSubscription; +import org.apache.rocketmq.common.lite.LiteUtil; import org.apache.rocketmq.common.lite.OffsetOption; +import org.apache.rocketmq.remoting.protocol.header.NotifyUnsubscribeLiteRequestHeader; import org.apache.rocketmq.remoting.protocol.subscription.SubscriptionGroupConfig; import org.junit.Before; import org.junit.Test; +import org.mockito.ArgumentCaptor; -import static org.apache.rocketmq.common.SubscriptionGroupAttributes.LITE_SUB_MODEL_ATTRIBUTE; -import static org.junit.jupiter.api.Assertions.assertEquals; -import static org.junit.jupiter.api.Assertions.assertFalse; -import static org.junit.jupiter.api.Assertions.assertNotNull; -import static org.junit.jupiter.api.Assertions.assertNull; -import static org.junit.jupiter.api.Assertions.assertThrows; -import static org.junit.jupiter.api.Assertions.assertTrue; -import static org.mockito.ArgumentMatchers.anyInt; -import static org.mockito.ArgumentMatchers.anyLong; +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; +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.clearInvocations; import static org.mockito.Mockito.mock; -import static org.mockito.Mockito.never; -import static org.mockito.Mockito.times; import static org.mockito.Mockito.verify; import static org.mockito.Mockito.when; public class LiteSubscriptionRegistryImplTest { private LiteSubscriptionRegistryImpl registry; - private LiteCtlListener mockListener; + private BrokerController mockBrokerController; private AbstractLiteLifecycleManager mockLifecycleManager; - private BrokerConfig mockBrokerConfig; private SubscriptionGroupManager mockSubscriptionGroupManager; + private BrokerConfig mockBrokerConfig; private ConsumerOffsetManager mockConsumerOffsetManager; + private PopLiteMessageProcessor mockPopLiteMessageProcessor; + private QueueLevelConsumerManager mockConsumerOrderInfoManager; + private Broker2Client mockBroker2Client; + private LiteCtlListener mockListener; @Before public void setUp() { - BrokerController mockBrokerController = mock(BrokerController.class); + mockBrokerController = mock(BrokerController.class); mockLifecycleManager = mock(AbstractLiteLifecycleManager.class); - mockBrokerConfig = mock(BrokerConfig.class); mockSubscriptionGroupManager = mock(SubscriptionGroupManager.class); + mockBrokerConfig = mock(BrokerConfig.class); mockConsumerOffsetManager = mock(ConsumerOffsetManager.class); - PopLiteMessageProcessor mockPopLiteMessageProcessor = mock(PopLiteMessageProcessor.class); - QueueLevelConsumerManager mockConsumerOrderInfoManager = mock(QueueLevelConsumerManager.class); + mockPopLiteMessageProcessor = mock(PopLiteMessageProcessor.class); + mockConsumerOrderInfoManager = mock(QueueLevelConsumerManager.class); + mockBroker2Client = mock(Broker2Client.class); - when(mockBrokerController.getBrokerConfig()).thenReturn(mockBrokerConfig); when(mockBrokerController.getSubscriptionGroupManager()).thenReturn(mockSubscriptionGroupManager); + when(mockBrokerController.getBrokerConfig()).thenReturn(mockBrokerConfig); when(mockBrokerController.getConsumerOffsetManager()).thenReturn(mockConsumerOffsetManager); when(mockBrokerController.getPopLiteMessageProcessor()).thenReturn(mockPopLiteMessageProcessor); + when(mockBrokerController.getBroker2Client()).thenReturn(mockBroker2Client); when(mockPopLiteMessageProcessor.getConsumerOrderInfoManager()).thenReturn(mockConsumerOrderInfoManager); when(mockConsumerOrderInfoManager.getTable()).thenReturn(new ConcurrentHashMap<>()); + when(mockPopLiteMessageProcessor.getConsumerOrderInfoManager()).thenReturn(mockConsumerOrderInfoManager); when(mockBrokerConfig.getMaxLiteSubscriptionCount()).thenReturn(1000L); + when(mockBrokerConfig.getLiteSubscriptionCheckInterval()).thenReturn(1000L); when(mockBrokerConfig.getLiteSubscriptionCheckTimeoutMills()).thenReturn(60000L); - when(mockBrokerConfig.getLiteSubscriptionCheckInterval()).thenReturn(10000L); registry = new LiteSubscriptionRegistryImpl(mockBrokerController, mockLifecycleManager); mockListener = mock(LiteCtlListener.class); registry.addListener(mockListener); } - // Test addIncremental method + /** + * Test updateClientChannel updates client channel correctly + */ @Test - public void testAddPartialSubscription_BasicFunctionality() { - String clientId = "client1"; - String group = "group1"; - String topic = "topic1"; - Set liteTopicSet = new HashSet<>(); - liteTopicSet.add("lmq1"); - liteTopicSet.add("lmq2"); + public void testUpdateClientChannel_UpdateChannel() { + String clientId = "testClient"; + Channel mockChannel = mock(Channel.class); - when(mockLifecycleManager.isSubscriptionActive(anyString(), anyString())).thenReturn(true); + registry.updateClientChannel(clientId, mockChannel); - registry.addPartialSubscription(clientId, group, topic, liteTopicSet, null); - - LiteSubscription subscription = registry.getLiteSubscription(clientId); - assertNotNull(subscription); - assertEquals(group, subscription.getGroup()); - assertEquals(topic, subscription.getTopic()); - assertTrue(subscription.getLiteTopicSet().containsAll(liteTopicSet)); - - assertEquals(liteTopicSet.size(), registry.liteTopic2Group.size()); - Set topicGroupSet = registry.liteTopic2Group.get("lmq1"); - assertEquals(1, topicGroupSet.size()); - ClientGroup registeredGroup = topicGroupSet.iterator().next(); - assertEquals(clientId, registeredGroup.clientId); - assertEquals(group, registeredGroup.group); - - verify(mockListener, times(2)).onRegister(eq(clientId), eq(group), anyString()); + assertEquals(mockChannel, registry.clientChannels.get(clientId)); } + /** + * Test addPartialSubscription throws exception when quota exceeded + */ @Test - public void testAddPartialSubscription_ExclusiveMode() { - String existingClientId = "existingClient"; - String newClientId = "newClient"; - String group = "group"; - String topic = "topic"; - String liteTopic = "lmq1"; + public void testAddPartialSubscription_QuotaExceeded() { + // Set quota to 0 so any new subscription exceeds quota + when(mockBrokerConfig.getMaxLiteSubscriptionCount()).thenReturn(0L); - Set liteTopicSet = new HashSet<>(); - liteTopicSet.add(liteTopic); - - when(mockLifecycleManager.isSubscriptionActive(anyString(), anyString())).thenReturn(true); - - // Mock subscription group config for reset offset behavior - SubscriptionGroupConfig subscriptionGroupConfig = new SubscriptionGroupConfig(); - subscriptionGroupConfig.setGroupName(group); - subscriptionGroupConfig.getAttributes().put(LITE_SUB_MODEL_ATTRIBUTE.getName(), LiteSubModel.Exclusive.name()); - when(mockSubscriptionGroupManager.findSubscriptionGroupConfig(group)).thenReturn(subscriptionGroupConfig); - - // Add existing client - registry.addPartialSubscription(existingClientId, group, topic, liteTopicSet, null); - - // Verify that the existing client is correctly registered - LiteSubscription existingSubscription = registry.getLiteSubscription(existingClientId); - assertNotNull(existingSubscription); - assertTrue(existingSubscription.getLiteTopicSet().contains(liteTopic)); - - // Execute exclusive mode addition - Set newLiteTopicSet = new HashSet<>(); - newLiteTopicSet.add(liteTopic); - registry.addPartialSubscription(newClientId, group, topic, newLiteTopicSet, null); - - // Verify that new client subscription has been added. - LiteSubscription newSubscription = registry.getLiteSubscription(newClientId); - assertNotNull(newSubscription); - assertTrue(newSubscription.getLiteTopicSet().contains(liteTopic)); - - assertEquals(liteTopicSet.size(), registry.liteTopic2Group.size()); - Set topicGroupSet = registry.liteTopic2Group.get(liteTopic); - assertEquals(1, topicGroupSet.size()); - ClientGroup registeredGroup = topicGroupSet.iterator().next(); - assertEquals(newClientId, registeredGroup.clientId); - assertEquals(group, registeredGroup.group); - - verify(mockListener).onRegister(existingClientId, group, liteTopic); - verify(mockListener).onRegister(newClientId, group, liteTopic); - verify(mockListener).onUnregister(existingClientId, group, liteTopic); - } - - @Test - public void testAddPartialSubscription_NonExclusiveMode() { - // Add an existing client subscription first - String existingClientId = "existingClient"; - String newClientId = "newClient"; - String group = "group1"; - String topic = "topic1"; - String liteTopic = "lmq1"; - - Set existingLiteTopicSet = new HashSet<>(); - existingLiteTopicSet.add(liteTopic); - - when(mockLifecycleManager.isSubscriptionActive(anyString(), anyString())).thenReturn(true); - - // Mock subscription group config - SubscriptionGroupConfig subscriptionGroupConfig = new SubscriptionGroupConfig(); - subscriptionGroupConfig.setGroupName(group); - when(mockSubscriptionGroupManager.findSubscriptionGroupConfig(group)).thenReturn(subscriptionGroupConfig); - - // Add existing client - registry.addPartialSubscription(existingClientId, group, topic, existingLiteTopicSet, null); - - // Add new client in non-exclusive mode - Set newLiteTopicSet = new HashSet<>(); - newLiteTopicSet.add(liteTopic); - registry.addPartialSubscription(newClientId, group, topic, newLiteTopicSet, null); - - // Verify both client subscriptions exist - LiteSubscription existingSubscription = registry.getLiteSubscription(existingClientId); - LiteSubscription newSubscription = registry.getLiteSubscription(newClientId); - assertNotNull(existingSubscription); - assertNotNull(newSubscription); - assertTrue(existingSubscription.getLiteTopicSet().contains(liteTopic)); - assertTrue(newSubscription.getLiteTopicSet().contains(liteTopic)); - - // Verify listener was only called for registration, not unregistration - verify(mockListener, times(2)).onRegister(anyString(), eq(group), eq(liteTopic)); - verify(mockListener, never()).onUnregister(anyString(), anyString(), anyString()); - } - - @Test - public void testAddPartialSubscription_WithEmptyLiteTopicSet() { - String clientId = "client1"; - String group = "group1"; - String topic = "topic1"; - Set liteTopicSet = new HashSet<>(); - - registry.addPartialSubscription(clientId, group, topic, liteTopicSet, null); - - LiteSubscription subscription = registry.getLiteSubscription(clientId); - assertNotNull(subscription); - assertEquals(group, subscription.getGroup()); - assertEquals(topic, subscription.getTopic()); - assertTrue(subscription.getLiteTopicSet().isEmpty()); - - // Verify listener was not called - verify(mockListener, never()).onRegister(anyString(), anyString(), anyString()); - } - - @Test - public void testAddPartialSubscription_InactiveSubscription() { - String clientId = "client1"; - String group = "group1"; - String topic = "topic1"; - String inactiveLiteTopic = "inactive_lmq1"; - - Set liteTopicSet = new HashSet<>(); - liteTopicSet.add(inactiveLiteTopic); - - // Mock inactive subscription - when(mockLifecycleManager.isSubscriptionActive(topic, inactiveLiteTopic)).thenReturn(false); - - // Should not add inactive subscriptions - registry.addPartialSubscription(clientId, group, topic, liteTopicSet, null); - - LiteSubscription subscription = registry.getLiteSubscription(clientId); - assertNotNull(subscription); - assertFalse(subscription.getLiteTopicSet().contains(inactiveLiteTopic)); - assertEquals(0, registry.getActiveSubscriptionNum()); - } - - @Test - public void testAddPartialSubscription_ExclusiveModeDifferentGroups() { - // Add two clients from different groups - String client1 = "client1"; - String group1 = "group1"; - String client2 = "client2"; - String group2 = "group2"; - String topic = "topic1"; - String liteTopic = "lmq1"; - - Set liteTopicSet = new HashSet<>(); - liteTopicSet.add(liteTopic); - - when(mockLifecycleManager.isSubscriptionActive(anyString(), anyString())).thenReturn(true); - - // Mock subscription group configs - SubscriptionGroupConfig subscriptionGroupConfig1 = new SubscriptionGroupConfig(); - subscriptionGroupConfig1.setGroupName(group1); - subscriptionGroupConfig1.getAttributes().put(LITE_SUB_MODEL_ATTRIBUTE.getName(), LiteSubModel.Exclusive.name()); - when(mockSubscriptionGroupManager.findSubscriptionGroupConfig(group1)).thenReturn(subscriptionGroupConfig1); - - SubscriptionGroupConfig subscriptionGroupConfig2 = new SubscriptionGroupConfig(); - subscriptionGroupConfig2.setGroupName(group2); - subscriptionGroupConfig2.getAttributes().put(LITE_SUB_MODEL_ATTRIBUTE.getName(), LiteSubModel.Exclusive.name()); - when(mockSubscriptionGroupManager.findSubscriptionGroupConfig(group2)).thenReturn(subscriptionGroupConfig2); - - // Add first client - registry.addPartialSubscription(client1, group1, topic, liteTopicSet, null); - - // Add second client - registry.addPartialSubscription(client2, group2, topic, liteTopicSet, null); - - // Verify both clients are registered for the same topic - Set observers = registry.getSubscriber(liteTopic); - assertEquals(2, observers.size()); - - // Add new client in exclusive mode from the same group as client1 - String client3 = "client3"; - registry.addPartialSubscription(client3, group1, topic, liteTopicSet, null); - - // Verify only client1 was removed (same group), client2 remains (different group) - observers = registry.getSubscriber(liteTopic); - assertEquals(2, observers.size()); // client2(group2) and client3(group1) - - boolean hasClient2 = false; - boolean hasClient3 = false; - for (ClientGroup cg : observers) { - if (cg.clientId.equals(client2) && cg.group.equals(group2)) { - hasClient2 = true; - } - if (cg.clientId.equals(client3) && cg.group.equals(group1)) { - hasClient3 = true; - } - } - - assertTrue(hasClient2, "Client2 (group2) should still be registered"); - assertTrue(hasClient3, "Client3 (group1) should be registered"); - - // Verify listener calls - verify(mockListener).onUnregister(client1, group1, liteTopic); // Same group client1 removed - verify(mockListener, never()).onUnregister(client2, group2, liteTopic); // Different group client2 retained - } - - @Test - public void testAddPartialSubscription_QuotaLimit() { - // Set quota to 1 - when(mockBrokerConfig.getMaxLiteSubscriptionCount()).thenReturn(1L); - - when(mockLifecycleManager.isSubscriptionActive(anyString(), anyString())).thenReturn(true); - - // Add first subscription - String clientId1 = "client1"; - String group1 = "group1"; - String topic1 = "topic1"; - Set liteTopicSet1 = new HashSet<>(); - liteTopicSet1.add("lmq1"); - - registry.addPartialSubscription(clientId1, group1, topic1, liteTopicSet1, null); - - // Try to add second subscription, should throw exception - String clientId2 = "client2"; - String group2 = "group2"; - String topic2 = "topic2"; - Set liteTopicSet2 = new HashSet<>(); - liteTopicSet2.add("lmq2"); + String clientId = "testClient"; + String group = "testGroup"; + String topic = "testTopic"; + Set lmqNameSet = Collections.singleton("lmq1"); assertThrows(LiteQuotaException.class, () -> { - registry.addPartialSubscription(clientId2, group2, topic2, liteTopicSet2, null); + registry.addPartialSubscription(clientId, group, topic, lmqNameSet, null); }); } - // Test removeIncremental method + /** + * Test addPartialSubscription throws exception for wildcard group + */ @Test - public void testRemovePartialSubscription() { - String clientId = "client1"; - String group = "group1"; - String topic = "topic1"; - String liteTopic1 = "lmq1"; - String liteTopic2 = "lmq2"; + public void testAddPartialSubscription_WildcardGroup() { + String clientId = "testClient"; + String group = "testGroup"; + String topic = "testTopic"; + Set lmqNameSet = Collections.singleton("lmq1"); - Set liteTopicSet = new HashSet<>(); - liteTopicSet.add(liteTopic1); - liteTopicSet.add(liteTopic2); + // Simulate wildcard group + SubscriptionGroupConfig groupConfig = new SubscriptionGroupConfig(); + groupConfig.setGroupName(group); + groupConfig.setWildcardLiteGroup(true); + when(mockSubscriptionGroupManager.findSubscriptionGroupConfig(group)).thenReturn(groupConfig); - when(mockLifecycleManager.isSubscriptionActive(anyString(), anyString())).thenReturn(true); - - // Add subscriptions first - registry.addPartialSubscription(clientId, group, topic, liteTopicSet, null); - - // Verify subscriptions were added - LiteSubscription subscription = registry.getLiteSubscription(clientId); - assertTrue(subscription.getLiteTopicSet().contains(liteTopic1)); - assertTrue(subscription.getLiteTopicSet().contains(liteTopic2)); - - // Remove some subscriptions - Set toRemove = new HashSet<>(); - toRemove.add(liteTopic1); - registry.removePartialSubscription(clientId, group, topic, toRemove); - - // Verify removal was successful - subscription = registry.getLiteSubscription(clientId); - assertFalse(subscription.getLiteTopicSet().contains(liteTopic1)); - assertTrue(subscription.getLiteTopicSet().contains(liteTopic2)); - - verify(mockListener).onUnregister(clientId, group, liteTopic1); - verify(mockListener, never()).onUnregister(clientId, group, liteTopic2); + assertThrows(IllegalStateException.class, () -> { + registry.addPartialSubscription(clientId, group, topic, lmqNameSet, null); + }); } - // Test addAll method + /** + * Test addPartialSubscription does not add inactive subscription + */ @Test - public void testAddCompleteSubscription() { - String clientId = "client1"; - String group = "group1"; - String topic = "topic1"; - String liteTopic1 = "lmq1"; - String liteTopic2 = "lmq2"; - String liteTopic3 = "lmq3"; + public void testAddPartialSubscription_InactiveSubscription() { + String clientId = "testClient"; + String group = "testGroup"; + String topic = "testTopic"; + Set lmqNameSet = Collections.singleton("lmq1"); - // Initial subscriptions - Set initialSet = new HashSet<>(); - initialSet.add(liteTopic1); - initialSet.add(liteTopic2); + // Simulate non-wildcard group + SubscriptionGroupConfig groupConfig = new SubscriptionGroupConfig(); + groupConfig.setGroupName(group); - // New full subscription set - Set newFullSet = new HashSet<>(); - newFullSet.add(liteTopic2); - newFullSet.add(liteTopic3); + when(mockSubscriptionGroupManager.findSubscriptionGroupConfig(group)).thenReturn(groupConfig); + when(mockLifecycleManager.isSubscriptionActive(topic, "lmq1")).thenReturn(false); - when(mockLifecycleManager.isSubscriptionActive(anyString(), anyString())).thenReturn(true); + registry.addPartialSubscription(clientId, group, topic, lmqNameSet, null); - // Add initial subscriptions - registry.addPartialSubscription(clientId, group, topic, initialSet, null); - - // Reset mock to ignore previous interactions - clearInvocations(mockListener); - - // Update with addAll - registry.addCompleteSubscription(clientId, group, topic, newFullSet, 1L); - - // Verify update results LiteSubscription subscription = registry.getLiteSubscription(clientId); - assertFalse(subscription.getLiteTopicSet().contains(liteTopic1)); // Should be removed - assertTrue(subscription.getLiteTopicSet().contains(liteTopic2)); // Should be retained - assertTrue(subscription.getLiteTopicSet().contains(liteTopic3)); // Should be added - - // Verify that liteTopic1 was unregistered (no longer in new set) - verify(mockListener).onUnregister(clientId, group, liteTopic1); - - // Verify that liteTopic3 was registered (new in the set) - verify(mockListener).onRegister(clientId, group, liteTopic3); - - // Verify that liteTopic2 was neither unregistered nor registered again - // (it was already registered and remains in the new set) - verify(mockListener, never()).onUnregister(clientId, group, liteTopic2); + assertNotNull(subscription); + assertFalse(subscription.getLiteTopicSet().contains("lmq1")); + assertEquals(0, registry.getActiveSubscriptionNum()); } - // Test removeAll method + /** + * Test addPartialSubscription adds subscription normally + */ @Test - public void testRemoveCompleteSubscription() { - String clientId = "client1"; - String group = "group1"; - String topic = "topic1"; - String liteTopic1 = "lmq1"; - String liteTopic2 = "lmq2"; + public void testAddPartialSubscription_NormalCase() { + String clientId = "testClient"; + String group = "testGroup"; + String topic = "testTopic"; + Set lmqNameSet = Collections.singleton("lmq1"); - Set liteTopicSet = new HashSet<>(); - liteTopicSet.add(liteTopic1); - liteTopicSet.add(liteTopic2); + // Simulate non-wildcard group + SubscriptionGroupConfig groupConfig = new SubscriptionGroupConfig(); + groupConfig.setGroupName(group); + when(mockSubscriptionGroupManager.findSubscriptionGroupConfig(group)).thenReturn(groupConfig); + when(mockLifecycleManager.isSubscriptionActive(topic, "lmq1")).thenReturn(true); - when(mockLifecycleManager.isSubscriptionActive(anyString(), anyString())).thenReturn(true); + registry.addPartialSubscription(clientId, group, topic, lmqNameSet, null); - // Add subscriptions - registry.addPartialSubscription(clientId, group, topic, liteTopicSet, null); + LiteSubscription subscription = registry.getLiteSubscription(clientId); + assertNotNull(subscription); + assertTrue(subscription.getLiteTopicSet().contains("lmq1")); + assertEquals(1, registry.getActiveSubscriptionNum()); - // Verify subscriptions were added - assertNotNull(registry.getLiteSubscription(clientId)); + verify(mockListener).onRegister(clientId, group, "lmq1"); + } + + /** + * Test addPartialSubscription excludes client in exclusive mode + */ + @Test + public void testAddPartialSubscription_ExclusiveMode() { + String clientId1 = "testClient1"; + String clientId2 = "testClient2"; + String group = "testGroup"; + String topic = "testTopic"; + Set lmqNameSet = Collections.singleton("lmq1"); + + // Simulate non-wildcard group + SubscriptionGroupConfig groupConfig = new SubscriptionGroupConfig(); + groupConfig.setGroupName(group); + groupConfig.setLiteSubExclusive(true); + when(mockSubscriptionGroupManager.findSubscriptionGroupConfig(group)).thenReturn(groupConfig); + when(mockLifecycleManager.isSubscriptionActive(topic, "lmq1")).thenReturn(true); + + // Add first client + registry.addPartialSubscription(clientId1, group, topic, lmqNameSet, null); + + LiteSubscription subscription1 = registry.getLiteSubscription(clientId1); + assertNotNull(subscription1); + assertTrue(subscription1.getLiteTopicSet().contains("lmq1")); + assertEquals(1, registry.getActiveSubscriptionNum()); + + // Add second client, should exclude first client + registry.addPartialSubscription(clientId2, group, topic, lmqNameSet, null); + + LiteSubscription subscription2 = registry.getLiteSubscription(clientId2); + assertNotNull(subscription2); + assertTrue(subscription2.getLiteTopicSet().contains("lmq1")); + assertNull(registry.getLiteSubscription(clientId1)); + assertEquals(1, registry.getActiveSubscriptionNum()); + + verify(mockListener).onRegister(clientId1, group, "lmq1"); + verify(mockListener).onUnregister(clientId1, group, "lmq1"); + verify(mockListener).onRegister(clientId2, group, "lmq1"); + } + + /** + * Test removePartialSubscription removes partial subscription correctly + */ + @Test + public void testRemovePartialSubscription_RemoveSubscription() { + String clientId = "testClient"; + String group = "testGroup"; + String topic = "testTopic"; + Set lmqNameSet = new HashSet<>(); + lmqNameSet.add("lmq1"); + lmqNameSet.add("lmq2"); + + // Simulate non-wildcard group + SubscriptionGroupConfig groupConfig = new SubscriptionGroupConfig(); + groupConfig.setGroupName(group); + when(mockSubscriptionGroupManager.findSubscriptionGroupConfig(group)).thenReturn(groupConfig); + when(mockLifecycleManager.isSubscriptionActive(eq(topic), anyString())).thenReturn(true); + + // Add subscription first + registry.addPartialSubscription(clientId, group, topic, lmqNameSet, null); + + LiteSubscription subscription = registry.getLiteSubscription(clientId); + assertNotNull(subscription); + assertTrue(subscription.getLiteTopicSet().contains("lmq1")); + assertTrue(subscription.getLiteTopicSet().contains("lmq2")); assertEquals(2, registry.getActiveSubscriptionNum()); - // Remove all subscriptions + // Remove partial subscription + Set toRemove = Collections.singleton("lmq1"); + registry.removePartialSubscription(clientId, group, topic, toRemove); + + subscription = registry.getLiteSubscription(clientId); + assertNotNull(subscription); + assertFalse(subscription.getLiteTopicSet().contains("lmq1")); + assertTrue(subscription.getLiteTopicSet().contains("lmq2")); + assertEquals(1, registry.getActiveSubscriptionNum()); + + verify(mockListener).onUnregister(clientId, group, "lmq1"); + } + + /** + * Test addCompleteSubscription handles wildcard group + */ + @Test + public void testAddCompleteSubscription_WildcardGroup() { + String clientId = "testClient"; + String group = "testGroup"; + String topic = "testTopic"; + Set lmqNameAll = new HashSet<>(); + lmqNameAll.add("lmq1"); + lmqNameAll.add("lmq2"); + + // Simulate wildcard group + SubscriptionGroupConfig groupConfig = new SubscriptionGroupConfig(); + groupConfig.setGroupName(group); + groupConfig.setWildcardLiteGroup(true); + when(mockSubscriptionGroupManager.findSubscriptionGroupConfig(group)).thenReturn(groupConfig); + when(mockLifecycleManager.isSubscriptionActive(eq(topic), anyString())).thenReturn(true); + + registry.addCompleteSubscription(clientId, group, topic, lmqNameAll, 1L); + + assertTrue(registry.wildcardGroupMap.containsKey(topic)); + assertTrue(registry.wildcardGroupMap.get(topic).contains(group)); + + LiteSubscription subscription = registry.getLiteSubscription(clientId); + assertNotNull(subscription); + assertTrue(subscription.getLiteTopicSet().contains(topic + "@" + group)); + assertEquals(1, registry.getActiveSubscriptionNum()); + } + + /** + * Test addCompleteSubscription updates complete subscription + */ + @Test + public void testAddCompleteSubscription_UpdateSubscription() { + String clientId = "testClient"; + String group = "testGroup"; + String topic = "testTopic"; + Set lmqNameAll = new HashSet<>(); + lmqNameAll.add("lmq1"); + lmqNameAll.add("lmq2"); + + Set lmqNameNew = new HashSet<>(); + lmqNameNew.add("lmq2"); + lmqNameNew.add("lmq3"); + + // Simulate non-wildcard group + SubscriptionGroupConfig groupConfig = new SubscriptionGroupConfig(); + groupConfig.setGroupName(group); + when(mockSubscriptionGroupManager.findSubscriptionGroupConfig(group)).thenReturn(groupConfig); + + when(mockLifecycleManager.isSubscriptionActive(eq(topic), anyString())).thenReturn(true); + + // Add initial subscription + registry.addCompleteSubscription(clientId, group, topic, lmqNameAll, 1L); + + LiteSubscription subscription = registry.getLiteSubscription(clientId); + assertNotNull(subscription); + assertTrue(subscription.getLiteTopicSet().contains("lmq1")); + assertTrue(subscription.getLiteTopicSet().contains("lmq2")); + assertEquals(2, registry.getActiveSubscriptionNum()); + + // Update subscription + registry.addCompleteSubscription(clientId, group, topic, lmqNameNew, 2L); + + subscription = registry.getLiteSubscription(clientId); + assertNotNull(subscription); + assertFalse(subscription.getLiteTopicSet().contains("lmq1")); + assertTrue(subscription.getLiteTopicSet().contains("lmq2")); + assertTrue(subscription.getLiteTopicSet().contains("lmq3")); + assertEquals(2, registry.getActiveSubscriptionNum()); + } + + /** + * Test removeCompleteSubscription removes all subscriptions + */ + @Test + public void testRemoveCompleteSubscription_RemoveAll() { + String clientId = "testClient"; + String group = "testGroup"; + String topic = "testTopic"; + Set lmqNameSet = new HashSet<>(); + lmqNameSet.add("lmq1"); + lmqNameSet.add("lmq2"); + + // Simulate non-wildcard group + SubscriptionGroupConfig groupConfig = new SubscriptionGroupConfig(); + groupConfig.setGroupName(group); + when(mockSubscriptionGroupManager.findSubscriptionGroupConfig(group)).thenReturn(groupConfig); + + when(mockLifecycleManager.isSubscriptionActive(eq(topic), anyString())).thenReturn(true); + + // Add subscription first + registry.addPartialSubscription(clientId, group, topic, lmqNameSet, null); + + LiteSubscription subscription = registry.getLiteSubscription(clientId); + assertNotNull(subscription); + assertTrue(subscription.getLiteTopicSet().contains("lmq1")); + assertTrue(subscription.getLiteTopicSet().contains("lmq2")); + assertEquals(2, registry.getActiveSubscriptionNum()); + + // Remove complete subscription registry.removeCompleteSubscription(clientId); - // Verify all subscriptions were removed assertNull(registry.getLiteSubscription(clientId)); + assertNull(registry.clientChannels.get(clientId)); assertEquals(0, registry.getActiveSubscriptionNum()); verify(mockListener).onRemoveAll(clientId, group); } + /** + * Test addListener adds listener + */ @Test - public void testRemoveCompleteSubscription_NonExistentClient() { - String nonExistentClientId = "nonexistent"; + public void testAddListener_AddListener() { + LiteCtlListener listener = mock(LiteCtlListener.class); - // Should not throw exception - registry.removeCompleteSubscription(nonExistentClientId); + registry.addListener(listener); - // Verify no changes to registry state - assertEquals(0, registry.getActiveSubscriptionNum()); - assertNull(registry.getLiteSubscription(nonExistentClientId)); + assertTrue(registry.listeners.contains(listener)); } - // Test cleanSubscription method + /** + * Test getAllSubscriber gets wildcard subscribers + */ @Test - public void testCleanSubscription() { - String clientId = "client1"; - String group = "group1"; - String topic = "topic1"; - String liteTopic1 = "lmq1"; - String liteTopic2 = "lmq2"; + public void testGetAllSubscriber_WildcardGroup() { + String group = "testGroup"; + String topic = "testTopic"; + String lmqName = topic + "@" + group; - Set liteTopicSet = new HashSet<>(); - liteTopicSet.add(liteTopic1); - liteTopicSet.add(liteTopic2); + // Simulate wildcard group + SubscriptionGroupConfig groupConfig = new SubscriptionGroupConfig(); + groupConfig.setGroupName(group); + groupConfig.setWildcardLiteGroup(true); + when(mockSubscriptionGroupManager.findSubscriptionGroupConfig(group)).thenReturn(groupConfig); - when(mockLifecycleManager.isSubscriptionActive(anyString(), anyString())).thenReturn(true); + SubscriberWrapper result = registry.getAllSubscriber(group, lmqName); - // Add subscription - registry.addPartialSubscription(clientId, group, topic, liteTopicSet, null); - assertEquals(2, registry.getActiveSubscriptionNum()); - - // Verify subscription was added - LiteSubscription subscription = registry.getLiteSubscription(clientId); - assertTrue(subscription.getLiteTopicSet().contains(liteTopic1)); - assertTrue(subscription.getLiteTopicSet().contains(liteTopic2)); - - // Clean subscription - registry.cleanSubscription(liteTopic1, true); - registry.cleanSubscription(liteTopic2, false); - - // Verify subscription was cleaned - subscription = registry.getLiteSubscription(clientId); - assertFalse(subscription.getLiteTopicSet().contains(liteTopic1)); - assertFalse(subscription.getLiteTopicSet().contains(liteTopic2)); - assertNull(registry.getSubscriber(liteTopic1)); - assertNull(registry.getSubscriber(liteTopic2)); - assertEquals(0, registry.getActiveSubscriptionNum()); + assertNotNull(result); + assertInstanceOf(SubscriberWrapper.ListWrapper.class, result); } - // Test getSubscriber method + /** + * Test getAllSubscriber gets subscribers for specific group + */ @Test - public void testGetSubscriber() { - String clientId = "client1"; - String group = "group1"; - String topic = "topic1"; - String liteTopic = "lmq1"; - - Set liteTopicSet = new HashSet<>(); - liteTopicSet.add(liteTopic); - - when(mockLifecycleManager.isSubscriptionActive(anyString(), anyString())).thenReturn(true); - - registry.addPartialSubscription(clientId, group, topic, liteTopicSet, null); - - Set observers = registry.getSubscriber(liteTopic); - assertNotNull(observers); - assertEquals(1, observers.size()); - ClientGroup clientGroup = observers.iterator().next(); - assertEquals(clientId, clientGroup.clientId); - assertEquals(group, clientGroup.group); - } - - @Test - public void testGetSubscriber_NonExistentTopic() { - String nonExistentTopic = "nonexistent_lmq"; - - Set result = registry.getSubscriber(nonExistentTopic); - - // Should return null for non-existent topic - assertNull(result); - } - - // Test updateClientChannel method - @Test - public void testUpdateClientChannel() { - String clientId = "client1"; - Channel mockChannel = mock(Channel.class); - - registry.updateClientChannel(clientId, mockChannel); - - // Verify channel was updated - assertEquals(mockChannel, registry.clientChannels.get(clientId)); - } - - // Test getActiveSubscriptionNum method - @Test - public void testGetActiveSubscriptionNum() { - String clientId1 = "client1"; - String clientId2 = "client2"; - String group = "group1"; - String topic = "topic1"; - String liteTopic1 = "lmq1"; - String liteTopic2 = "lmq2"; - - Set liteTopicSet1 = new HashSet<>(); - liteTopicSet1.add(liteTopic1); - - Set liteTopicSet2 = new HashSet<>(); - liteTopicSet2.add(liteTopic1); // Same topic - liteTopicSet2.add(liteTopic2); // New topic - - when(mockLifecycleManager.isSubscriptionActive(anyString(), anyString())).thenReturn(true); - - // Initial state - assertEquals(0, registry.getActiveSubscriptionNum()); - - // Add first client - registry.addPartialSubscription(clientId1, group, topic, liteTopicSet1, null); - assertEquals(1, registry.getActiveSubscriptionNum()); - - // Add second client - registry.addPartialSubscription(clientId2, group, topic, liteTopicSet2, null); - assertEquals(3, registry.getActiveSubscriptionNum()); // 3 references: client1->topic1, client2->topic1, client2->topic2 - } - - // Test cleanupExpiredSubscriptions method - @Test - public void testCleanupExpiredSubscriptions_NoExpiredClients() { - String clientId = "client1"; - String group = "group1"; - String topic = "topic1"; - Set liteTopics = new HashSet<>(); - liteTopics.add("lmq1"); - liteTopics.add("lmq2"); - - LiteSubscription subscription = new LiteSubscription(); - subscription.setGroup(group); - subscription.setTopic(topic); - subscription.addLiteTopic(liteTopics); - subscription.setUpdateTime(System.currentTimeMillis()); // Not expired - - Channel channel = mock(Channel.class); - - registry.client2Subscription.put(clientId, subscription); - registry.clientChannels.put(clientId, channel); - - // Initialize liteTopic2Group - for (String lmq : liteTopics) { - registry.liteTopic2Group.computeIfAbsent(lmq, k -> ConcurrentHashMap.newKeySet()) - .add(new ClientGroup(clientId, group)); - } - - registry.activeNum.set(liteTopics.size()); - - // Perform cleanup with a timeout of 10 seconds - registry.cleanupExpiredSubscriptions(10000); - - // Verify that the client has not been cleaned up - assertNotNull(registry.client2Subscription.get(clientId)); - assertNotNull(registry.clientChannels.get(clientId)); - assertEquals(liteTopics.size(), registry.activeNum.get()); - } - - @Test - public void testCleanupExpiredSubscriptions_WithExpiredClients() { - String clientId = "client1"; - String group = "group1"; - String topic = "topic1"; - Set liteTopics = new HashSet<>(); - liteTopics.add("lmq1"); - liteTopics.add("lmq2"); - - LiteSubscription subscription = new LiteSubscription(); - subscription.setGroup(group); - subscription.setTopic(topic); - subscription.addLiteTopic(liteTopics); - subscription.setUpdateTime(System.currentTimeMillis() - 20000); - - Channel channel = mock(Channel.class); - - registry.client2Subscription.put(clientId, subscription); - registry.clientChannels.put(clientId, channel); - - // Initialize liteTopic2Group - for (String lmq : liteTopics) { - registry.liteTopic2Group.computeIfAbsent(lmq, k -> ConcurrentHashMap.newKeySet()) - .add(new ClientGroup(clientId, group)); - } - - registry.activeNum.set(liteTopics.size()); - - LiteCtlListener mockListener = mock(LiteCtlListener.class); - registry.addListener(mockListener); - - // Perform cleanup with a timeout of 10 seconds - registry.cleanupExpiredSubscriptions(10000); - - // Verify that the client has been cleaned up - assertNull(registry.client2Subscription.get(clientId)); - assertNull(registry.clientChannels.get(clientId)); - assertEquals(0, registry.activeNum.get()); - - // Verify that the listener was called - verify(mockListener, times(1)).onUnregister(eq(clientId), eq(group), eq("lmq1")); - verify(mockListener, times(1)).onUnregister(eq(clientId), eq(group), eq("lmq2")); - verify(mockListener, times(1)).onRemoveAll(eq(clientId), eq(group)); - - // Verify that topics in liteTopic2Group have been removed - assertNull(registry.liteTopic2Group.get("lmq1")); - assertNull(registry.liteTopic2Group.get("lmq2")); - } - - @Test - public void testCleanupExpiredSubscriptions_ExpiredClientWithNoSubscriptions() { - String clientId = "client1"; - String group = "group1"; - String topic = "topic1"; - Set liteTopics = new HashSet<>(); - - LiteSubscription subscription = new LiteSubscription(); - subscription.setGroup(group); - subscription.setTopic(topic); - subscription.addLiteTopic(liteTopics); - subscription.setUpdateTime(System.currentTimeMillis() - 20000); // Expired - - Channel channel = mock(Channel.class); - - registry.client2Subscription.put(clientId, subscription); - registry.clientChannels.put(clientId, channel); - - registry.activeNum.set(0); - - LiteCtlListener mockListener = mock(LiteCtlListener.class); - registry.addListener(mockListener); - - // Perform cleanup with 10 second timeout - registry.cleanupExpiredSubscriptions(10000); - - // Verify that the client has been cleaned up - assertNull(registry.client2Subscription.get(clientId)); - assertNull(registry.clientChannels.get(clientId)); - assertEquals(0, registry.activeNum.get()); - - // Verify that the listener was not called - verify(mockListener, never()).onUnregister(anyString(), anyString(), anyString()); - } - - // Test removeTopicGroup method - @Test - public void testRemoveTopicGroup_EmptyTopicGroupSet() { - String clientId = "client1"; - String group = "group1"; - String liteTopic = "lmq1"; - - ClientGroup clientGroup = new ClientGroup(clientId, group); - - // Initialize with a single client - Set topicGroupSet = ConcurrentHashMap.newKeySet(); - topicGroupSet.add(clientGroup); - registry.liteTopic2Group.put(liteTopic, topicGroupSet); - registry.activeNum.set(1); - - // Remove the only client - registry.removeTopicGroup(clientGroup, liteTopic, false); - - // Verify that the topic is completely removed from liteTopic2Group - assertNull(registry.liteTopic2Group.get(liteTopic)); - assertEquals(0, registry.getActiveSubscriptionNum()); - } - - // Test excludeClientByLmqName method - @Test - public void testExcludeClientByLmqName_EmptyClientSet() { - String newClientId = "newClient"; - String group = "group1"; + public void testGetAllSubscriber_SpecificGroup() { + String clientId = "testClient"; + String group = "testGroup"; String lmqName = "lmq1"; - // Ensure the liteTopic2Group map exists but is empty - registry.liteTopic2Group.put(lmqName, ConcurrentHashMap.newKeySet()); + // Add subscription + ClientGroup clientGroup = new ClientGroup(clientId, group); + Set clientSet = ConcurrentHashMap.newKeySet(); + clientSet.add(clientGroup); + registry.liteTopic2Group.put(lmqName, clientSet); - // Should not throw any exception - registry.excludeClientByLmqName(newClientId, group, lmqName); + SubscriberWrapper result = registry.getAllSubscriber(group, lmqName); - // Verify no changes - assertTrue(registry.liteTopic2Group.get(lmqName).isEmpty()); + 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); } + /** + * Test getAllSubscriber gets subscribers for all groups + */ @Test - public void testGetAllClientIdByGroup() { - String group1 = "group1"; - String group2 = "group2"; - String clientId1 = "client1"; - String clientId2 = "client2"; - String clientId3 = "client3"; - String topic = "parentTopic"; + public void testGetAllSubscriber_AllGroups() { + String clientId1 = "testClient1"; + String clientId2 = "testClient2"; + String group1 = "testGroup1"; + String group2 = "testGroup2"; + String topic = "testTopic"; + String lmqName = LiteUtil.toLmqName(topic, "lmq1"); - LiteSubscription sub1 = new LiteSubscription(); - sub1.setGroup(group1); - sub1.setTopic(topic); + // Add subscription + ClientGroup clientGroup1 = new ClientGroup(clientId1, group1); + ClientGroup clientGroup2 = new ClientGroup(clientId2, group2); + Set clientSet = ConcurrentHashMap.newKeySet(); + clientSet.add(clientGroup1); + clientSet.add(clientGroup2); + registry.liteTopic2Group.put(lmqName, clientSet); - LiteSubscription sub2 = new LiteSubscription(); - sub2.setGroup(group1); - sub2.setTopic(topic); + SubscriberWrapper result = registry.getAllSubscriber(null, lmqName); - LiteSubscription sub3 = new LiteSubscription(); - sub3.setGroup(group2); - sub3.setTopic(topic); + 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()); + } - registry.client2Subscription.put(clientId1, sub1); - registry.client2Subscription.put(clientId2, sub2); - registry.client2Subscription.put(clientId3, sub3); + /** + * Test cleanSubscription cleans subscription + */ + @Test + public void testCleanSubscription_CleanSubscription() { + String clientId = "testClient"; + String group = "testGroup"; + String lmqName = "lmq1"; - List result; + // Add subscription + ClientGroup clientGroup = new ClientGroup(clientId, group); + Set clientSet = ConcurrentHashMap.newKeySet(); + clientSet.add(clientGroup); + registry.liteTopic2Group.put(lmqName, clientSet); + + LiteSubscription subscription = new LiteSubscription(); + subscription.setGroup(group); + subscription.addLiteTopic(lmqName); + registry.client2Subscription.put(clientId, subscription); + registry.activeNum.set(1); + + registry.cleanSubscription(lmqName, false); + + assertFalse(registry.liteTopic2Group.containsKey(lmqName)); + assertFalse(subscription.getLiteTopicSet().contains(lmqName)); + assertEquals(0, registry.getActiveSubscriptionNum()); + } + + /** + * Test getLiteSubscription gets LiteSubscription + */ + @Test + public void testGetLiteSubscription_GetSubscription() { + String clientId = "testClient"; + String group = "testGroup"; + String topic = "testTopic"; + + LiteSubscription subscription = new LiteSubscription(); + subscription.setGroup(group); + subscription.setTopic(topic); + registry.client2Subscription.put(clientId, subscription); + + LiteSubscription result = registry.getLiteSubscription(clientId); + + assertNotNull(result); + assertEquals(group, result.getGroup()); + assertEquals(topic, result.getTopic()); + } + + /** + * Test getActiveSubscriptionNum gets active subscription count + */ + @Test + public void testGetActiveSubscriptionNum_GetCount() { + registry.activeNum.set(5); + + int count = registry.getActiveSubscriptionNum(); + + assertEquals(5, count); + } + + /** + * Test getAllClientIdByGroup gets all client IDs by group + */ + @Test + public void testGetAllClientIdByGroup_GetClientIds() { + String clientId1 = "testClient1"; + String clientId2 = "testClient2"; + String clientId3 = "testClient3"; + String group1 = "testGroup1"; + String group2 = "testGroup2"; + String topic = "testTopic"; + + LiteSubscription subscription1 = new LiteSubscription(); + subscription1.setGroup(group1); + subscription1.setTopic(topic); + registry.client2Subscription.put(clientId1, subscription1); + + LiteSubscription subscription2 = new LiteSubscription(); + subscription2.setGroup(group1); + subscription2.setTopic(topic); + registry.client2Subscription.put(clientId2, subscription2); + + LiteSubscription subscription3 = new LiteSubscription(); + subscription3.setGroup(group2); + subscription3.setTopic(topic); + registry.client2Subscription.put(clientId3, subscription3); + + List result = registry.getAllClientIdByGroup(group1); - // group1 - result = registry.getAllClientIdByGroup(group1); assertEquals(2, result.size()); assertTrue(result.contains(clientId1)); assertTrue(result.contains(clientId2)); - - // group2 - result = registry.getAllClientIdByGroup(group2); - assertEquals(1, result.size()); - assertTrue(result.contains(clientId3)); - - // not exist - result = registry.getAllClientIdByGroup("notExistGroup"); - assertTrue(result.isEmpty()); - - // null - result = registry.getAllClientIdByGroup(null); - assertTrue(result.isEmpty()); } + /** + * Test resetOffset resets offset to specific value + */ @Test - public void testResetOffset_minOffset() { + public void testResetOffset_SpecificOffset() { String lmqName = "lmq1"; - String group = "group1"; - String clientId = "client1"; + String group = "testGroup"; + String clientId = "testClient"; + long specifiedOffset = 250L; + + when(mockConsumerOffsetManager.queryOffset(group, lmqName, 0)).thenReturn(100L); + + OffsetOption offsetOption = new OffsetOption(OffsetOption.Type.OFFSET, specifiedOffset); + registry.resetOffset(lmqName, group, clientId, offsetOption); + + verify(mockConsumerOffsetManager).assignResetOffset(lmqName, group, 0, specifiedOffset); + } + + /** + * Test resetOffset resets offset to minimum + */ + @Test + public void testResetOffset_MinOffset() { + String lmqName = "lmq1"; + String group = "testGroup"; + String clientId = "testClient"; when(mockConsumerOffsetManager.queryOffset(group, lmqName, 0)).thenReturn(100L); @@ -804,11 +608,14 @@ public class LiteSubscriptionRegistryImplTest { verify(mockConsumerOffsetManager).assignResetOffset(lmqName, group, 0, 0L); } + /** + * Test resetOffset resets offset to maximum + */ @Test - public void testResetOffset_maxOffset() { + public void testResetOffset_MaxOffset() { String lmqName = "lmq1"; - String group = "group1"; - String clientId = "client1"; + String group = "testGroup"; + String clientId = "testClient"; long maxOffset = 500L; when(mockConsumerOffsetManager.queryOffset(group, lmqName, 0)).thenReturn(100L); @@ -820,55 +627,48 @@ public class LiteSubscriptionRegistryImplTest { verify(mockConsumerOffsetManager).assignResetOffset(lmqName, group, 0, maxOffset); } + /** + * Test notifyUnsubscribeLite notifies client to unsubscribe + */ @Test - public void testResetOffset_absolute() { - String lmqName = "lmq1"; - String group = "group1"; - String clientId = "client1"; - long specifiedOffset = 250L; + public void testNotifyUnsubscribeLite_NotifyClient() { + String clientId = "testClient"; + String group = "testGroup"; + String lmqName = LiteUtil.toLmqName("testTopic", "lmq1"); + Channel mockChannel = mock(Channel.class); - when(mockConsumerOffsetManager.queryOffset(group, lmqName, 0)).thenReturn(100L); + registry.clientChannels.put(clientId, mockChannel); - OffsetOption offsetOption = new OffsetOption(OffsetOption.Type.OFFSET, specifiedOffset); - registry.resetOffset(lmqName, group, clientId, offsetOption); + registry.notifyUnsubscribeLite(clientId, group, lmqName); - verify(mockConsumerOffsetManager).assignResetOffset(lmqName, group, 0, specifiedOffset); + ArgumentCaptor captor = ArgumentCaptor.forClass(NotifyUnsubscribeLiteRequestHeader.class); + verify(mockBroker2Client).notifyUnsubscribeLite(eq(mockChannel), captor.capture()); + NotifyUnsubscribeLiteRequestHeader header = captor.getValue(); + assertEquals(clientId, header.getClientId()); + assertEquals(group, header.getConsumerGroup()); + assertEquals("lmq1", header.getLiteTopic()); } + /** + * Test cleanupExpiredSubscriptions cleans expired subscriptions + */ @Test - public void testResetOffset_LastN() { - String lmqName = "lmq1"; - String group1 = "group1"; - String group2 = "group2"; - String clientId = "client1"; - long currentOffset = 100L; - long lastN = 20L; - long expectedTargetOffset = 80L; + public void testCleanupExpiredSubscriptions_CleanExpired() { + String clientId = "testClient"; + String group = "testGroup"; + String topic = "testTopic"; + long timeout = 10000L; // 10 seconds - when(mockConsumerOffsetManager.queryOffset(group1, lmqName, 0)).thenReturn(currentOffset); - when(mockConsumerOffsetManager.queryOffset(group2, lmqName, 0)).thenReturn(-1L); + LiteSubscription subscription = new LiteSubscription(); + subscription.setGroup(group); + subscription.setTopic(topic); + // Updated 20 seconds ago, expired + subscription.setUpdateTime(System.currentTimeMillis() - 20000L); - OffsetOption offsetOption = new OffsetOption(OffsetOption.Type.TAIL_N, lastN); + registry.client2Subscription.put(clientId, subscription); + registry.cleanupExpiredSubscriptions(timeout); - registry.resetOffset(lmqName, group1, clientId, offsetOption); - registry.resetOffset(lmqName, group2, clientId, offsetOption); - - verify(mockConsumerOffsetManager).assignResetOffset(lmqName, group1, 0, expectedTargetOffset); - verify(mockConsumerOffsetManager, never()).assignResetOffset(lmqName, group2, 0, expectedTargetOffset); + assertFalse(registry.client2Subscription.containsKey(clientId)); + assertEquals(0, registry.getActiveSubscriptionNum()); } - - @Test - public void testResetOffset_timestamp_not_supported() { - String lmqName = "lmq1"; - String group = "group1"; - String clientId = "client1"; - long timestamp = System.currentTimeMillis(); - - when(mockConsumerOffsetManager.queryOffset(group, lmqName, 0)).thenReturn(100L); - - OffsetOption offsetOption = new OffsetOption(OffsetOption.Type.TIMESTAMP, timestamp); - registry.resetOffset(lmqName, group, clientId, offsetOption); - - verify(mockConsumerOffsetManager, never()).assignResetOffset(anyString(), anyString(), anyInt(), anyLong()); - } -} \ No newline at end of file +} diff --git a/broker/src/test/java/org/apache/rocketmq/broker/processor/LiteManagerProcessorTest.java b/broker/src/test/java/org/apache/rocketmq/broker/processor/LiteManagerProcessorTest.java index c6cf731281..5518a2fa10 100644 --- a/broker/src/test/java/org/apache/rocketmq/broker/processor/LiteManagerProcessorTest.java +++ b/broker/src/test/java/org/apache/rocketmq/broker/processor/LiteManagerProcessorTest.java @@ -19,6 +19,7 @@ package org.apache.rocketmq.broker.processor; import io.netty.channel.ChannelHandlerContext; import java.util.ArrayList; +import java.util.Collections; import java.util.HashSet; import java.util.List; import java.util.Set; @@ -30,6 +31,7 @@ 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; @@ -70,6 +72,7 @@ import org.junit.runner.RunWith; import org.mockito.Mock; import org.mockito.junit.MockitoJUnitRunner; +import static org.apache.rocketmq.common.SubscriptionGroupAttributes.LITE_SUB_WILDCARD_ATTRIBUTE; import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertNotNull; import static org.junit.Assert.assertNull; @@ -346,9 +349,10 @@ public class LiteManagerProcessorTest { when(liteLifecycleManager.getMaxOffsetInQueue(lmqName)).thenReturn(maxOffset); when(messageStore.getMinOffsetInQueue(lmqName, 0)).thenReturn(minOffset); when(messageStore.getMessageStoreTimeStamp(lmqName, 0, maxOffset - 1)).thenReturn(lastUpdateTimestamp); - Set subscribers = new HashSet<>(); - subscribers.add(new ClientGroup("clientId1", "group1")); - when(liteSubscriptionRegistry.getSubscriber(lmqName)).thenReturn(subscribers); + + SubscriberWrapper.MapWrapper wrapper = new SubscriberWrapper.MapWrapper(); + wrapper.getGroupMap().put("group", Collections.singletonList(new ClientGroup("clientId", "group"))); + when(liteSubscriptionRegistry.getAllSubscriber(null, lmqName)).thenReturn(wrapper); when(brokerController.getBrokerConfig()).thenReturn(mock(BrokerConfig.class)); when(brokerController.getBrokerConfig().getBrokerName()).thenReturn("broker1"); when(liteSharding.shardingByLmqName("parent_topic", lmqName)).thenReturn("broker1"); @@ -361,7 +365,7 @@ public class LiteManagerProcessorTest { GetLiteTopicInfoResponseBody body = GetLiteTopicInfoResponseBody.decode(response.getBody(), GetLiteTopicInfoResponseBody.class); assertEquals("parent_topic", body.getParentTopic()); assertEquals("lite_topic", body.getLiteTopic()); - assertEquals(subscribers, body.getSubscriber()); + assertEquals("clientId", body.getSubscriber().iterator().next().clientId); TopicOffset topicOffset = body.getTopicOffset(); assertEquals(minOffset, topicOffset.getMinOffset()); @@ -722,7 +726,7 @@ public class LiteManagerProcessorTest { assertNotNull(response); assertEquals(ResponseCode.SUCCESS, response.getCode()); - verify(liteEventDispatcher, times(1)).doFullDispatch(clientId, group); + verify(liteEventDispatcher, times(1)).doFullDispatchForClient(clientId, group); verify(liteEventDispatcher, never()).doFullDispatchByGroup(group); // without clientId @@ -735,7 +739,49 @@ public class LiteManagerProcessorTest { assertNotNull(response); assertEquals(ResponseCode.SUCCESS, response.getCode()); - verify(liteEventDispatcher, times(1)).doFullDispatch(clientId, group); + verify(liteEventDispatcher, times(1)).doFullDispatchForClient(clientId, group); verify(liteEventDispatcher, times(1)).doFullDispatchByGroup(group); } + + @Test + public void testGetSubscriber_null() { + String lmqName = "lmqName"; + when(liteSubscriptionRegistry.getAllSubscriber(null, lmqName)).thenReturn(new SubscriberWrapper.ListWrapper()); + + Set result = processor.getSubscriber(lmqName); + assertEquals(0, result.size()); + } + + @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); + + Set result = processor.getSubscriber(lmqName); + assertEquals(1, result.size()); + assertEquals("clientId", result.iterator().next().clientId); + } + + @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"))); + SubscriptionGroupConfig groupConfig = new SubscriptionGroupConfig(); + groupConfig.getAttributes().put(LITE_SUB_WILDCARD_ATTRIBUTE.getName(), "xxx"); + + when(liteSubscriptionRegistry.getAllSubscriber(null, lmqName)).thenReturn(wrapper); + when(subscriptionGroupManager.findSubscriptionGroupConfig("wildcardGroup")).thenReturn(groupConfig); + + Set result = processor.getSubscriber(lmqName); + assertEquals(2, result.size()); + result.forEach(clientGroup -> { + if (clientGroup.group.equals("wildcardGroup")) { + assertEquals("*", clientGroup.clientId); + } + }); + } } diff --git a/common/src/main/java/org/apache/rocketmq/common/BrokerConfig.java b/common/src/main/java/org/apache/rocketmq/common/BrokerConfig.java index 7271c12b18..4dfbe39f9e 100644 --- a/common/src/main/java/org/apache/rocketmq/common/BrokerConfig.java +++ b/common/src/main/java/org/apache/rocketmq/common/BrokerConfig.java @@ -523,6 +523,8 @@ public class BrokerConfig extends BrokerIdentity { private long liteEventFullDispatchDelayTime = 10 * 1000; + private long liteEventFullDispatchDelayTimeForWildcardGroup = 10 * 1000; + // lite metrics // whether to collect storeTime in popLiteProcessor private boolean liteLagLatencyCollectEnable = false; @@ -2378,6 +2380,14 @@ public class BrokerConfig extends BrokerIdentity { this.liteEventFullDispatchDelayTime = liteEventFullDispatchDelayTime; } + public long getLiteEventFullDispatchDelayTimeForWildcardGroup() { + return liteEventFullDispatchDelayTimeForWildcardGroup; + } + + public void setLiteEventFullDispatchDelayTimeForWildcardGroup(long liteEventFullDispatchDelayTimeForWildcardGroup) { + this.liteEventFullDispatchDelayTimeForWildcardGroup = liteEventFullDispatchDelayTimeForWildcardGroup; + } + public boolean isLiteLagLatencyCollectEnable() { return liteLagLatencyCollectEnable; } diff --git a/common/src/main/java/org/apache/rocketmq/common/SubscriptionGroupAttributes.java b/common/src/main/java/org/apache/rocketmq/common/SubscriptionGroupAttributes.java index 12f5dbf67e..3329188f8a 100644 --- a/common/src/main/java/org/apache/rocketmq/common/SubscriptionGroupAttributes.java +++ b/common/src/main/java/org/apache/rocketmq/common/SubscriptionGroupAttributes.java @@ -73,7 +73,7 @@ public class SubscriptionGroupAttributes { 2000 ); - public static final LongRangeAttribute LITE_SUB_CLIENT_MAX_EVENT_COUNT = new LongRangeAttribute( + public static final LongRangeAttribute LITE_SUB_CLIENT_MAX_EVENT_COUNT_ATTRIBUTE = new LongRangeAttribute( "lite.sub.client.max.event.cnt", true, 10, @@ -81,6 +81,11 @@ public class SubscriptionGroupAttributes { 400 ); + public static final StringAttribute LITE_SUB_WILDCARD_ATTRIBUTE = new StringAttribute( + "lite.sub.wildcard", + true + ); + static { ALL = new HashMap<>(); ALL.put(PRIORITY_FACTOR_ATTRIBUTE.getName(), PRIORITY_FACTOR_ATTRIBUTE); @@ -89,6 +94,7 @@ public class SubscriptionGroupAttributes { ALL.put(LITE_SUB_MODEL_ATTRIBUTE.getName(), LITE_SUB_MODEL_ATTRIBUTE); ALL.put(LITE_SUB_RESET_OFFSET_EXCLUSIVE_ATTRIBUTE.getName(), LITE_SUB_RESET_OFFSET_EXCLUSIVE_ATTRIBUTE); ALL.put(LITE_SUB_RESET_OFFSET_UNSUBSCRIBE_ATTRIBUTE.getName(), LITE_SUB_RESET_OFFSET_UNSUBSCRIBE_ATTRIBUTE); - ALL.put(LITE_SUB_CLIENT_MAX_EVENT_COUNT.getName(), LITE_SUB_CLIENT_MAX_EVENT_COUNT); + ALL.put(LITE_SUB_CLIENT_MAX_EVENT_COUNT_ATTRIBUTE.getName(), LITE_SUB_CLIENT_MAX_EVENT_COUNT_ATTRIBUTE); + ALL.put(LITE_SUB_WILDCARD_ATTRIBUTE.getName(), LITE_SUB_WILDCARD_ATTRIBUTE); } -} +} \ No newline at end of file diff --git a/pom.xml b/pom.xml index f28beaf9e1..7e66370b1b 100644 --- a/pom.xml +++ b/pom.xml @@ -127,7 +127,7 @@ 6.0.53 1.0-beta-4 1.4.2 - 2.1.1 + 2.1.2 1.53.0 3.20.1 1.2.10 diff --git a/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/v2/consumer/ChangeInvisibleDurationActivity.java b/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/v2/consumer/ChangeInvisibleDurationActivity.java index f90d658ef2..f314da6c0a 100644 --- a/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/v2/consumer/ChangeInvisibleDurationActivity.java +++ b/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/v2/consumer/ChangeInvisibleDurationActivity.java @@ -60,8 +60,12 @@ public class ChangeInvisibleDurationActivity extends AbstractMessagingActivity { request.getMessageId(), group, request.getTopic().getName(), - Durations.toMillis(request.getInvisibleDuration()) - ).thenApply(ackResult -> convertToChangeInvisibleDurationResponse(ctx, request, ackResult)); + Durations.toMillis(request.getInvisibleDuration()), + request.getLiteTopic(), + MessagingProcessor.DEFAULT_TIMEOUT_MILLS, + request.getSuspend() + ).thenApply( + ackResult -> convertToChangeInvisibleDurationResponse(ctx, request, ackResult)); } catch (Throwable t) { future.completeExceptionally(t); } diff --git a/proxy/src/main/java/org/apache/rocketmq/proxy/processor/ReceiptHandleProcessor.java b/proxy/src/main/java/org/apache/rocketmq/proxy/processor/ReceiptHandleProcessor.java index 3038690109..bc3730aed9 100644 --- a/proxy/src/main/java/org/apache/rocketmq/proxy/processor/ReceiptHandleProcessor.java +++ b/proxy/src/main/java/org/apache/rocketmq/proxy/processor/ReceiptHandleProcessor.java @@ -40,8 +40,10 @@ public class ReceiptHandleProcessor extends AbstractProcessor { .setChannel(event.getKey().getChannel()); MessageReceiptHandle messageReceiptHandle = event.getMessageReceiptHandle(); ReceiptHandle handle = ReceiptHandle.decode(messageReceiptHandle.getReceiptHandleStr()); - messagingProcessor.changeInvisibleTime(context, handle, messageReceiptHandle.getMessageId(), - messageReceiptHandle.getGroup(), messageReceiptHandle.getTopic(), event.getRenewTime(), messageReceiptHandle.getLiteTopic()) + messagingProcessor + .changeInvisibleTime(context, handle, messageReceiptHandle.getMessageId(), + messageReceiptHandle.getGroup(), messageReceiptHandle.getTopic(), + event.getRenewTime(), messageReceiptHandle.getLiteTopic()) .whenComplete((v, t) -> { if (t != null) { event.getFuture().completeExceptionally(t); diff --git a/proxy/src/test/java/org/apache/rocketmq/proxy/grpc/v2/consumer/ChangeInvisibleDurationActivityTest.java b/proxy/src/test/java/org/apache/rocketmq/proxy/grpc/v2/consumer/ChangeInvisibleDurationActivityTest.java index 2de9a066be..0201a058bc 100644 --- a/proxy/src/test/java/org/apache/rocketmq/proxy/grpc/v2/consumer/ChangeInvisibleDurationActivityTest.java +++ b/proxy/src/test/java/org/apache/rocketmq/proxy/grpc/v2/consumer/ChangeInvisibleDurationActivityTest.java @@ -37,6 +37,8 @@ import org.mockito.ArgumentCaptor; import static org.junit.Assert.assertEquals; import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.anyBoolean; +import static org.mockito.ArgumentMatchers.anyLong; import static org.mockito.ArgumentMatchers.anyString; import static org.mockito.Mockito.when; @@ -61,7 +63,15 @@ public class ChangeInvisibleDurationActivityTest extends BaseActivityTest { ackResult.setExtraInfo(newHandle); ackResult.setStatus(AckStatus.OK); when(this.messagingProcessor.changeInvisibleTime( - any(), any(), anyString(), anyString(), anyString(), invisibleTimeArgumentCaptor.capture() + any(), + any(), + anyString(), + anyString(), + anyString(), + invisibleTimeArgumentCaptor.capture(), + anyString(), // request.getLiteTopic() + anyLong(), // MessagingProcessor.DEFAULT_TIMEOUT_MILLS + anyBoolean() // request.getSuspend() )).thenReturn(CompletableFuture.completedFuture(ackResult)); ChangeInvisibleDurationResponse response = this.changeInvisibleDurationActivity.changeInvisibleDuration( @@ -90,7 +100,15 @@ public class ChangeInvisibleDurationActivityTest extends BaseActivityTest { String savedHandleStr = buildReceiptHandle("topic", System.currentTimeMillis(),3000); ArgumentCaptor receiptHandleCaptor = ArgumentCaptor.forClass(ReceiptHandle.class); when(this.messagingProcessor.changeInvisibleTime( - any(), receiptHandleCaptor.capture(), anyString(), anyString(), anyString(), invisibleTimeArgumentCaptor.capture() + any(), + receiptHandleCaptor.capture(), + anyString(), + anyString(), + anyString(), + invisibleTimeArgumentCaptor.capture(), + anyString(), // request.getLiteTopic() + anyLong(), // MessagingProcessor.DEFAULT_TIMEOUT_MILLS + anyBoolean() // request.getSuspend() )).thenReturn(CompletableFuture.completedFuture(ackResult)); when(messagingProcessor.removeReceiptHandle(any(), any(), anyString(), anyString(), anyString())) .thenReturn(new MessageReceiptHandle("group", "topic", 0, savedHandleStr, "msgId", 0, 0)); @@ -119,9 +137,16 @@ public class ChangeInvisibleDurationActivityTest extends BaseActivityTest { AckResult ackResult = new AckResult(); ackResult.setStatus(AckStatus.NO_EXIST); when(this.messagingProcessor.changeInvisibleTime( - any(), any(), anyString(), anyString(), anyString(), invisibleTimeArgumentCaptor.capture() + any(), + any(), + anyString(), + anyString(), + anyString(), + invisibleTimeArgumentCaptor.capture(), + anyString(), // request.getLiteTopic() + anyLong(), // MessagingProcessor.DEFAULT_TIMEOUT_MILLS + anyBoolean() // request.getSuspend() )).thenReturn(CompletableFuture.completedFuture(ackResult)); - ChangeInvisibleDurationResponse response = this.changeInvisibleDurationActivity.changeInvisibleDuration( createContext(), ChangeInvisibleDurationRequest.newBuilder() diff --git a/remoting/src/main/java/org/apache/rocketmq/remoting/protocol/subscription/SubscriptionGroupConfig.java b/remoting/src/main/java/org/apache/rocketmq/remoting/protocol/subscription/SubscriptionGroupConfig.java index fa8a9804f4..ef8b443ad5 100644 --- a/remoting/src/main/java/org/apache/rocketmq/remoting/protocol/subscription/SubscriptionGroupConfig.java +++ b/remoting/src/main/java/org/apache/rocketmq/remoting/protocol/subscription/SubscriptionGroupConfig.java @@ -28,12 +28,13 @@ import org.apache.commons.lang3.math.NumberUtils; import org.apache.rocketmq.common.MixAll; import org.apache.rocketmq.common.attribute.LiteSubModel; -import static org.apache.rocketmq.common.SubscriptionGroupAttributes.LITE_SUB_CLIENT_MAX_EVENT_COUNT; +import static org.apache.rocketmq.common.SubscriptionGroupAttributes.LITE_SUB_CLIENT_MAX_EVENT_COUNT_ATTRIBUTE; import static org.apache.rocketmq.common.SubscriptionGroupAttributes.LITE_SUB_CLIENT_QUOTA_ATTRIBUTE; import static org.apache.rocketmq.common.SubscriptionGroupAttributes.LITE_SUB_MODEL_ATTRIBUTE; import static org.apache.rocketmq.common.SubscriptionGroupAttributes.LITE_SUB_RESET_OFFSET_EXCLUSIVE_ATTRIBUTE; import static org.apache.rocketmq.common.SubscriptionGroupAttributes.LITE_BIND_TOPIC_ATTRIBUTE; import static org.apache.rocketmq.common.SubscriptionGroupAttributes.LITE_SUB_RESET_OFFSET_UNSUBSCRIBE_ATTRIBUTE; +import static org.apache.rocketmq.common.SubscriptionGroupAttributes.LITE_SUB_WILDCARD_ATTRIBUTE; import static org.apache.rocketmq.common.SubscriptionGroupAttributes.PRIORITY_FACTOR_ATTRIBUTE; @@ -214,6 +215,13 @@ public class SubscriptionGroupConfig { return Math.toIntExact(quota); } + @JSONField(serialize = false, deserialize = false) + public void setLiteSubExclusive(boolean liteSubExclusive) { + if (liteSubExclusive) { + attributes.put(LITE_SUB_MODEL_ATTRIBUTE.getName(), LiteSubModel.Exclusive.name()); + } + } + @JSONField(serialize = false, deserialize = false) public boolean isLiteSubExclusive() { String subLiteModel = attributes.get(LITE_SUB_MODEL_ATTRIBUTE.getName()); @@ -237,13 +245,25 @@ public class SubscriptionGroupConfig { @JSONField(serialize = false, deserialize = false) public int getMaxClientEventCount() { - String content = attributes.get(LITE_SUB_CLIENT_MAX_EVENT_COUNT.getName()); + String content = attributes.get(LITE_SUB_CLIENT_MAX_EVENT_COUNT_ATTRIBUTE.getName()); if (content == null) { return -1; } return NumberUtils.toInt(content, -1); } + @JSONField(serialize = false, deserialize = false) + public void setWildcardLiteGroup(boolean wildcard) { + if (wildcard) { + attributes.put(LITE_SUB_WILDCARD_ATTRIBUTE.getName(), "true"); + } + } + + @JSONField(serialize = false, deserialize = false) + public boolean isWildcardLiteGroup() { + return attributes.containsKey(LITE_SUB_WILDCARD_ATTRIBUTE.getName()); + } + @Override public int hashCode() { final int prime = 31;