opensource/rocketmq
3
0
Fork 0
mirror of https://github.com/apache/rocketmq.git synced 2026-08-28 20:09:14 +08:00
Code Issues Packages Projects Releases Wiki Activity

[ISSUE #10203] Support wildcard subscription and and consumer suspend for LiteTopic (#10204)

Browse Source 
- 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
This commit is contained in:
Quan
2026-04-03 11:23:08 +08:00
committed by GitHub
parent 860de80261
commit 614b81693b
24 changed files with 1616 additions and 1345 deletions
Download Patch File Download Diff File Expand all files Collapse all files
WORKSPACE
+1 -1
View File
@@ -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",
broker/src/main/java/org/apache/rocketmq/broker/lite/AbstractLiteLifecycleManager.java
+23 -1
View File
@@ -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<String, Integer> ttlMap = Collections.emptyMap();
protected Map<String, Set<String>> subscriberGroupMap = Collections.emptyMap();
protected Map<String, Integer> 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<String> collectByParentTopic(String parentTopic);
/**
* Iterator of lite topic, for high frequency iteration
* Triple<lmqName, maxOffsetInQueue, lastStoreTimestamp>, lastStoreTimestamp is null for now
* return true to continue, false to break.
*
* @param function consumer func
*/
public abstract void forEachLiteTopic(Function<Triple<String, Long, Long>, 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);
broker/src/main/java/org/apache/rocketmq/broker/lite/LiteEventDispatcher.java
+134 -114
View File
@@ -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<String, ClientEventSet> clientEventMap = new ConcurrentHashMap<>();
private final ConcurrentSkipListSet<FullDispatchRequest> fullDispatchSet = new ConcurrentSkipListSet<>(COMPARATOR);
private final ConcurrentMap<String, Object> fullDispatchMap = new ConcurrentHashMap<>(); // deduplication
protected final ConcurrentMap<String, ClientEventSet> clientEventMap = new ConcurrentHashMap<>();
protected final ConcurrentSkipListSet<FullDispatchRequest> fullDispatchSet = new ConcurrentSkipListSet<>(COMPARATOR);
protected final ConcurrentMap<String, Object> fullDispatchMap = new ConcurrentHashMap<>(); // deduplication
private final Cache<String, Object> 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<ClientGroup>) subscribers, excludeClientId);
if (wrapper instanceof SubscriberWrapper.ListWrapper) {
selectAndDispatch(lmqName, wrapper.asListWrapper().getClients(), excludeClientId);
}
if (subscribers instanceof Map) {
Map<String, List<ClientGroup>> map = (Map<String, List<ClientGroup>>) subscribers;
if (wrapper instanceof SubscriberWrapper.MapWrapper) {
Map<String, List<ClientGroup>> 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<ClientGroup> clients, String excludeClientId) {
public boolean selectAndDispatch(String lmqName, List<ClientGroup> 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<ClientGroup> 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<ClientGroup> 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<String> 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<ClientGroup> clients = liteSubscriptionRegistry.getWildcardSubscriber(group, parentTopic).getClients();
if (CollectionUtils.isEmpty(clients)) {
return;
}
AtomicInteger count = new AtomicInteger();
Function<Triple<String, Long, Long>, 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<ClientGroup>
* 2. When group is specified, return List<ClientGroup> containing subscribers of that group
* 3. When group is null and multiple groups exist, return Map<String, List<ClientGroup>>
* mapping each group to its subscribers
*
* @return Object that can be either List<ClientGroup> or Map<String, List<ClientGroup>> 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<ClientGroup> 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<String> 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<ClientGroup> result = new ArrayList<>(4);
for (ClientGroup ele : observers) {
if (group.equals(ele.group)) {
result.add(ele);
}
}
return !result.isEmpty() ? result : null;
}
}
Map<String, List<ClientGroup>> 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<String> events;
private final ConcurrentMap<String, Object> 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;
broker/src/main/java/org/apache/rocketmq/broker/lite/LiteLifecycleManager.java
+27
View File
@@ -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<Triple<String, Long, Long>, Boolean> function) {
Iterator<Map.Entry<String, ConcurrentMap<Integer, ConsumeQueueInterface>>> iterator =
messageStore.getQueueStore().getConsumeQueueTable().entrySet().iterator();
while (iterator.hasNext()) {
Map.Entry<String, ConcurrentMap<Integer, ConsumeQueueInterface>> entry = iterator.next();
if (!LiteUtil.isLiteTopicQueue(entry.getKey())) {
continue;
}
ConsumeQueueInterface consumeQueueInterface = entry.getValue().get(0);
if (null == consumeQueueInterface) {
continue;
}
Triple<String, Long, Long> 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;
}
}
}
}
broker/src/main/java/org/apache/rocketmq/broker/lite/LiteMetadataUtil.java
+9
View File
@@ -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<String, Integer> getTopicTtlMap(BrokerController brokerController) {
if (null == brokerController) {
return Collections.emptyMap();
broker/src/main/java/org/apache/rocketmq/broker/lite/LiteSubscriptionRegistry.java
+3 -2
View File
@@ -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<ClientGroup> getSubscriber(String lmqName);
SubscriberWrapper getAllSubscriber(String group, String lmqName);
SubscriberWrapper.ListWrapper getWildcardSubscriber(String group, String parentTopic);
List<String> getAllClientIdByGroup(String group);
broker/src/main/java/org/apache/rocketmq/broker/lite/LiteSubscriptionRegistryImpl.java
+116 -7
View File
@@ -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<String/*clientId*/, Channel> clientChannels = new ConcurrentHashMap<>();
protected final ConcurrentMap<String/*clientId*/, LiteSubscription> client2Subscription = new ConcurrentHashMap<>();
protected final ConcurrentMap<String/*lmqName*/, Set<ClientGroup>> liteTopic2Group = new ConcurrentHashMap<>();
protected final ConcurrentMap<String/*topic*/, Set<String/*group*/>> wildcardGroupMap = new ConcurrentHashMap<>();
private final Cache<String/*group*/, List<ClientGroup>> wildcardClientCache =
CacheBuilder.newBuilder().maximumSize(2000).expireAfterWrite(30, TimeUnit.SECONDS).build();
private final List<LiteCtlListener> listeners = new ArrayList<>();
protected final List<LiteCtlListener> 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<String> lmqNameAll, long version) {
Set<String> lmqNameNew = lmqNameAll.stream()
.filter(lmqName -> liteLifecycleManager.isSubscriptionActive(topic, lmqName))
.collect(Collectors.toSet());
Set<String> lmqNameNew;
if (LiteMetadataUtil.isWildcardGroup(group, brokerController)) {
lmqNameNew = Collections.singleton(mockLmqNameForWildcardGroup(topic, group));
markWildcardGroup(topic, group);
} else {
lmqNameNew = lmqNameAll.stream()
.filter(lmqName -> liteLifecycleManager.isSubscriptionActive(topic, lmqName))
.collect(Collectors.toSet());
}
LiteSubscription thisSub = getOrCreateLiteSubscription(clientId, group, topic);
Set<String> 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<ClientGroup>
* 2. When group is specified, return List<ClientGroup> containing subscribers of that group
* 3. When group is null and multiple groups exist, return Map<String, List<ClientGroup>>
* mapping each group to its subscribers
*/
@Override
public Set<ClientGroup> 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<ClientGroup> 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<ClientGroup> subscribers = liteTopic2Group.get(lmqName);
if (subscribers != null) {
for (ClientGroup clientGroup : subscribers) {
wrapper.getGroupMap().computeIfAbsent(clientGroup.group, k -> new ArrayList<>()).add(clientGroup);
}
}
Set<String> wildcardGroups = wildcardGroupMap.get(topic);
if (wildcardGroups != null) {
for (String wildcardGroup : wildcardGroups) {
List<ClientGroup> wildcardClients = getWildcardGroupClients(topic, wildcardGroup);
if (CollectionUtils.isNotEmpty(wildcardClients)) {
wrapper.getGroupMap().putIfAbsent(wildcardGroup, wildcardClients);
}
}
}
return wrapper;
}
}
@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<ClientGroup> getWildcardGroupClients(String topic, String group) {
List<ClientGroup> list = null;
try {
list = wildcardClientCache.get(group, () -> {
Set<ClientGroup> clientSet = liteTopic2Group.get(mockLmqNameForWildcardGroup(topic, group));
return clientSet != null ? new ArrayList<>(clientSet) : Collections.emptyList();
});
} catch (ExecutionException ignored) {
}
return list;
}
@Override
public void run() {
LOGGER.info("Start checking lite subscription.");
broker/src/main/java/org/apache/rocketmq/broker/lite/RocksDBLiteLifecycleManager.java
+22
View File
@@ -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<Triple<String, Long, Long>, Boolean> function) {
for (Map.Entry<String, Long> entry : maxCqOffsetTable.entrySet()) {
String queueAndQid = entry.getKey();
String queueName = queueAndQid.substring(0, queueAndQid.lastIndexOf("-"));
if (!LiteUtil.isLiteTopicQueue(queueName)) {
continue;
}
Triple<String, Long, Long> triple = Triple.of(queueName, entry.getValue() + 1, null);
try {
if (!function.apply(triple)) {
break;
}
} catch (Throwable e) {
LOGGER.error("forEachLiteTopic error. {}", queueName, e);
break;
}
}
}
}
broker/src/main/java/org/apache/rocketmq/broker/lite/SubscriberWrapper.java
+64
View File
@@ -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<ClientGroup> clients;
public ListWrapper() {
this.clients = new ArrayList<>();
}
public ListWrapper(List<ClientGroup> clients) {
this.clients = clients;
}
public List<ClientGroup> getClients() {
return this.clients;
}
}
public static class MapWrapper extends SubscriberWrapper {
private final Map<String, List<ClientGroup>> groupMap = new HashMap<>();
public MapWrapper() {
}
public Map<String, List<ClientGroup>> getGroupMap() {
return groupMap;
}
}
public ListWrapper asListWrapper() {
return this instanceof ListWrapper ? (ListWrapper) this : null;
}
public MapWrapper asMapWrapper() {
return this instanceof MapWrapper ? (MapWrapper) this : null;
}
}
broker/src/main/java/org/apache/rocketmq/broker/offset/MemoryConsumerOrderInfoManager.java
+24
View File
@@ -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<Integer, OrderInfo> 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.
broker/src/main/java/org/apache/rocketmq/broker/pop/orderly/QueueLevelConsumerManager.java
+1 -1
View File
@@ -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;
broker/src/main/java/org/apache/rocketmq/broker/processor/ChangeInvisibleTimeProcessor.java
+9 -5
View File
@@ -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));
broker/src/main/java/org/apache/rocketmq/broker/processor/LiteManagerProcessor.java
+27 -2
View File
@@ -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<ClientGroup> 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;
broker/src/test/java/org/apache/rocketmq/broker/lite/AbstractLiteLifecycleManagerTest.java
+7
View File
@@ -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<String> collectByParentTopic(String parentTopic) {
return PARENT_TOPIC.equals(parentTopic) ? Collections.singletonList(EXIST_LMQ_NAME) : Collections.emptyList();
}
@Override
public void forEachLiteTopic(Function<Triple<String, Long, Long>, Boolean> function) {
}
}
}
broker/src/test/java/org/apache/rocketmq/broker/lite/LiteEventDispatcherTest.java
+516 -492
View File
File diff suppressed because it is too large Load Diff
broker/src/test/java/org/apache/rocketmq/broker/lite/LiteSubscriptionRegistryImplTest.java
+500 -700
View File
File diff suppressed because it is too large Load Diff
broker/src/test/java/org/apache/rocketmq/broker/processor/LiteManagerProcessorTest.java
+52 -6
View File
@@ -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<ClientGroup> 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<ClientGroup> 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<ClientGroup> 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<ClientGroup> result = processor.getSubscriber(lmqName);
assertEquals(2, result.size());
result.forEach(clientGroup -> {
if (clientGroup.group.equals("wildcardGroup")) {
assertEquals("*", clientGroup.clientId);
}
});
}
}
common/src/main/java/org/apache/rocketmq/common/BrokerConfig.java
+10
View File
@@ -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;
}
common/src/main/java/org/apache/rocketmq/common/SubscriptionGroupAttributes.java
+9 -3
View File
@@ -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);
}
}
}
pom.xml
+1 -1
View File
@@ -127,7 +127,7 @@
<annotations-api.version>6.0.53</annotations-api.version>
<extra-enforcer-rules.version>1.0-beta-4</extra-enforcer-rules.version>
<concurrentlinkedhashmap-lru.version>1.4.2</concurrentlinkedhashmap-lru.version>
<rocketmq-proto.version>2.1.1</rocketmq-proto.version>
<rocketmq-proto.version>2.1.2</rocketmq-proto.version>
<grpc.version>1.53.0</grpc.version>
<protobuf.version>3.20.1</protobuf.version>
<disruptor.version>1.2.10</disruptor.version>
proxy/src/main/java/org/apache/rocketmq/proxy/grpc/v2/consumer/ChangeInvisibleDurationActivity.java
+6 -2
View File
@@ -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);
}
proxy/src/main/java/org/apache/rocketmq/proxy/processor/ReceiptHandleProcessor.java
+4 -2
View File
@@ -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);
proxy/src/test/java/org/apache/rocketmq/proxy/grpc/v2/consumer/ChangeInvisibleDurationActivityTest.java
+29 -4
View File
@@ -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<ReceiptHandle> 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()
remoting/src/main/java/org/apache/rocketmq/remoting/protocol/subscription/SubscriptionGroupConfig.java
+22 -2
View File
@@ -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;