diff --git a/broker/src/main/java/org/apache/rocketmq/broker/out/BrokerOuterAPI.java b/broker/src/main/java/org/apache/rocketmq/broker/out/BrokerOuterAPI.java index 83edd88408..21ba349c84 100644 --- a/broker/src/main/java/org/apache/rocketmq/broker/out/BrokerOuterAPI.java +++ b/broker/src/main/java/org/apache/rocketmq/broker/out/BrokerOuterAPI.java @@ -23,9 +23,14 @@ import java.nio.ByteBuffer; import java.util.ArrayList; import java.util.Arrays; import java.util.List; +import java.util.Map; +import java.util.Objects; +import java.util.Optional; import java.util.Set; import java.util.concurrent.ArrayBlockingQueue; import java.util.concurrent.CompletableFuture; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ConcurrentMap; import java.util.concurrent.CopyOnWriteArrayList; import java.util.concurrent.CountDownLatch; import java.util.concurrent.ExecutorService; @@ -102,6 +107,10 @@ import org.apache.rocketmq.remoting.protocol.body.TopicConfigSerializeWrapper; import org.apache.rocketmq.remoting.protocol.body.UnlockBatchRequestBody; import org.apache.rocketmq.remoting.protocol.header.ExchangeHAInfoRequestHeader; import org.apache.rocketmq.remoting.protocol.header.ExchangeHAInfoResponseHeader; +import org.apache.rocketmq.remoting.protocol.header.GetAllSubscriptionGroupRequestHeader; +import org.apache.rocketmq.remoting.protocol.header.GetAllSubscriptionGroupResponseHeader; +import org.apache.rocketmq.remoting.protocol.header.GetAllTopicConfigRequestHeader; +import org.apache.rocketmq.remoting.protocol.header.GetAllTopicConfigResponseHeader; import org.apache.rocketmq.remoting.protocol.header.GetBrokerMemberGroupRequestHeader; import org.apache.rocketmq.remoting.protocol.header.GetMaxOffsetRequestHeader; import org.apache.rocketmq.remoting.protocol.header.GetMaxOffsetResponseHeader; @@ -139,6 +148,8 @@ import org.apache.rocketmq.remoting.protocol.namesrv.RegisterBrokerResult; import org.apache.rocketmq.remoting.protocol.route.BrokerData; import org.apache.rocketmq.remoting.protocol.route.QueueData; import org.apache.rocketmq.remoting.protocol.route.TopicRouteData; +import org.apache.rocketmq.remoting.protocol.statictopic.TopicQueueMappingDetail; +import org.apache.rocketmq.remoting.protocol.subscription.SubscriptionGroupConfig; import org.apache.rocketmq.remoting.rpc.ClientMetadata; import org.apache.rocketmq.remoting.rpc.RpcClient; import org.apache.rocketmq.remoting.rpc.RpcClientImpl; @@ -761,22 +772,86 @@ public class BrokerOuterAPI { return changedList; } - public TopicConfigAndMappingSerializeWrapper getAllTopicConfig( - final String addr) throws RemotingConnectException, RemotingSendRequestException, - RemotingTimeoutException, InterruptedException, MQBrokerException { - RemotingCommand request = RemotingCommand.createRequestCommand(RequestCode.GET_ALL_TOPIC_CONFIG, null); + public TopicConfigAndMappingSerializeWrapper getAllTopicConfig(final String addr) + throws RemotingConnectException, RemotingSendRequestException, RemotingTimeoutException, + InterruptedException, MQBrokerException, RemotingCommandException { - RemotingCommand response = this.remotingClient.invokeSync(MixAll.brokerVIPChannel(true, addr), request, 3000); - assert response != null; - switch (response.getCode()) { - case ResponseCode.SUCCESS: { - return TopicConfigSerializeWrapper.decode(response.getBody(), TopicConfigAndMappingSerializeWrapper.class); + DataVersion topicConfigDataVersion = null; + DataVersion mappingDataVersion = null; + long timeoutMills = getTimeoutMillis(); + int topicSeq = 0; + long beginTime = System.nanoTime(); + ConcurrentHashMap topicConfigTable = new ConcurrentHashMap<>(); + Map topicQueueMappingDetailMap = new ConcurrentHashMap<>(); + while (true) { + long leftTime = timeoutMills - TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - beginTime); + if (leftTime < 0) { + throw new RemotingTimeoutException("invokeSync call timeout"); } - default: - break; + + GetAllTopicConfigRequestHeader requestHeader = new GetAllTopicConfigRequestHeader(); + requestHeader.setTopicSeq(topicSeq); + requestHeader.setMaxTopicNum(getMaxPageSize()); + requestHeader.setDataVersion(Optional.ofNullable(topicConfigDataVersion). + map(DataVersion::toJson).orElse(StringUtils.EMPTY)); + LOGGER.info("getAllTopicConfig from seq {}, max {}, dataVersion {}", + topicSeq, requestHeader.getMaxTopicNum(), requestHeader.getDataVersion()); + RemotingCommand request = RemotingCommand.createRequestCommand(RequestCode.GET_ALL_TOPIC_CONFIG, requestHeader); + + RemotingCommand response = this.remotingClient.invokeSync( + MixAll.brokerVIPChannel(true, addr), request, 30000); + + assert response != null; + if (response.getCode() == SUCCESS) { + TopicConfigAndMappingSerializeWrapper topicConfigSerializeWrapper = + TopicConfigAndMappingSerializeWrapper.decode(response.getBody(), TopicConfigAndMappingSerializeWrapper.class); + topicConfigTable.putAll(topicConfigSerializeWrapper.getTopicConfigTable()); + topicQueueMappingDetailMap.putAll(topicConfigSerializeWrapper.getTopicQueueMappingDetailMap()); + topicSeq += topicConfigSerializeWrapper.getTopicConfigTable().size(); + + + DataVersion newDataVersion = topicConfigSerializeWrapper.getDataVersion(); + if (topicConfigDataVersion == null) { + // fill dataVersion before break the loop to compatible with old version server + topicConfigDataVersion = newDataVersion; + mappingDataVersion = topicConfigSerializeWrapper.getMappingDataVersion(); + } + + GetAllTopicConfigResponseHeader responseHeader = + response.decodeCommandCustomHeader(GetAllTopicConfigResponseHeader.class); + Integer totalTopicNum = Optional.ofNullable(responseHeader) + .map(GetAllTopicConfigResponseHeader::getTotalTopicNum).orElse(null); + + if (Objects.isNull(totalTopicNum)) { // compatible with old version server + // the server side don't support totalTopicNum, all data is returned + break; + } + + if (!Objects.equals(topicConfigDataVersion, newDataVersion)) { + LOGGER.error("dataVersion changed, currentDataVersion: {}, newDataVersion: {}", topicConfigDataVersion, newDataVersion); + topicConfigDataVersion = newDataVersion; + mappingDataVersion = topicConfigSerializeWrapper.getMappingDataVersion(); + topicSeq = 0; + topicConfigTable.clear(); + continue; + } + + if (topicSeq >= totalTopicNum - 1) { + LOGGER.info("get all topic config, totalTopicNum: {}", totalTopicNum); + break; + } + } else { + throw new MQBrokerException(response.getCode(), response.getRemark(), addr); + } + } - throw new MQBrokerException(response.getCode(), response.getRemark(), addr); + TopicConfigAndMappingSerializeWrapper topicConfigSerializeWrapper = new TopicConfigAndMappingSerializeWrapper(); + topicConfigSerializeWrapper.setDataVersion(topicConfigDataVersion); + topicConfigSerializeWrapper.setTopicConfigTable(topicConfigTable); + topicConfigSerializeWrapper.setMappingDataVersion(mappingDataVersion); + topicConfigSerializeWrapper.setTopicQueueMappingDetailMap(topicQueueMappingDetailMap); + return topicConfigSerializeWrapper; } public TimerCheckpoint getTimerCheckPoint( @@ -849,21 +924,82 @@ public class BrokerOuterAPI { throw new MQBrokerException(response.getCode(), response.getRemark(), addr); } - public SubscriptionGroupWrapper getAllSubscriptionGroupConfig( - final String addr) throws InterruptedException, RemotingTimeoutException, - RemotingSendRequestException, RemotingConnectException, MQBrokerException { - RemotingCommand request = RemotingCommand.createRequestCommand(RequestCode.GET_ALL_SUBSCRIPTIONGROUP_CONFIG, null); - RemotingCommand response = this.remotingClient.invokeSync(addr, request, 3000); - assert response != null; - switch (response.getCode()) { - case ResponseCode.SUCCESS: { - return SubscriptionGroupWrapper.decode(response.getBody(), SubscriptionGroupWrapper.class); + public SubscriptionGroupWrapper getAllSubscriptionGroupConfig(final String addr) + throws InterruptedException, RemotingTimeoutException, RemotingSendRequestException, + RemotingConnectException, MQBrokerException, RemotingCommandException { + + long timeoutMills = getTimeoutMillis(); + DataVersion currentDataVersion = null; + int groupSeq = 0; + long beginTime = System.nanoTime(); + ConcurrentMap subscriptionGroupTable = new ConcurrentHashMap<>(); + ConcurrentMap> forbiddenTable = new ConcurrentHashMap<>(); + + while (true) { + long leftTime = timeoutMills - TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - beginTime); + if (leftTime < 0) { + throw new RemotingTimeoutException("invokeSync call timeout"); + } + + GetAllSubscriptionGroupRequestHeader requestHeader = new GetAllSubscriptionGroupRequestHeader(); + requestHeader.setGroupSeq(groupSeq); + requestHeader.setMaxGroupNum(getMaxPageSize()); + requestHeader.setDataVersion(Optional.ofNullable(currentDataVersion) + .map(DataVersion::toJson).orElse(StringUtils.EMPTY)); + LOGGER.info("getAllSubscriptionGroup from seq {}, max {}, dataVersion {}", + groupSeq, requestHeader.getMaxGroupNum(), requestHeader.getDataVersion()); + RemotingCommand request = RemotingCommand.createRequestCommand(RequestCode.GET_ALL_SUBSCRIPTIONGROUP_CONFIG, requestHeader); + RemotingCommand response = this.remotingClient.invokeSync(addr, request, 30000); + + assert response != null; + if (response.getCode() == SUCCESS) { + SubscriptionGroupWrapper subscriptionGroupWrapper = + SubscriptionGroupWrapper.decode(response.getBody(), SubscriptionGroupWrapper.class); + subscriptionGroupTable.putAll(subscriptionGroupWrapper.getSubscriptionGroupTable()); + forbiddenTable.putAll(subscriptionGroupWrapper.getForbiddenTable()); + + DataVersion newDataVersion = subscriptionGroupWrapper.getDataVersion(); + if (currentDataVersion == null) { + // fill dataVersion before break the loop to compatible with old version server + currentDataVersion = newDataVersion; + } + + groupSeq += subscriptionGroupWrapper.getSubscriptionGroupTable().size(); + + GetAllSubscriptionGroupResponseHeader responseHeader = + response.decodeCommandCustomHeader(GetAllSubscriptionGroupResponseHeader.class); + Integer totalGroupNum = Optional.ofNullable(responseHeader) + .map(GetAllSubscriptionGroupResponseHeader::getTotalGroupNum).orElse(null); + + if (Objects.isNull(totalGroupNum)) { + // the server side don't support totalGroupNum, all data is returned + break; + } + + if (!Objects.equals(currentDataVersion, newDataVersion)) { + LOGGER.error("dataVersion changed, currentDataVersion: {}, newDataVersion: {}", + currentDataVersion, newDataVersion); + currentDataVersion = newDataVersion; + groupSeq = 0; + subscriptionGroupTable.clear(); + forbiddenTable.clear(); + continue; + } + + if (groupSeq >= totalGroupNum - 1) { + LOGGER.info("get all subscription group config, totalGroupNum: {}", totalGroupNum); + break; + } + } else { + throw new MQBrokerException(response.getCode(), response.getRemark(), addr); } - default: - break; } - throw new MQBrokerException(response.getCode(), response.getRemark(), addr); + SubscriptionGroupWrapper allSubscriptionGroup = new SubscriptionGroupWrapper(); + allSubscriptionGroup.setSubscriptionGroupTable(subscriptionGroupTable); + allSubscriptionGroup.setForbiddenTable(forbiddenTable); + allSubscriptionGroup.setDataVersion(currentDataVersion); + return allSubscriptionGroup; } public void registerRPCHook(RPCHook rpcHook) { @@ -1491,4 +1627,12 @@ public class BrokerOuterAPI { return pullResult; } + private int getMaxPageSize() { + return 2000; + } + + private long getTimeoutMillis() { + return TimeUnit.SECONDS.toMillis(60); + } + } diff --git a/broker/src/main/java/org/apache/rocketmq/broker/processor/AdminBrokerProcessor.java b/broker/src/main/java/org/apache/rocketmq/broker/processor/AdminBrokerProcessor.java index fd8d925e3b..4203b3af82 100644 --- a/broker/src/main/java/org/apache/rocketmq/broker/processor/AdminBrokerProcessor.java +++ b/broker/src/main/java/org/apache/rocketmq/broker/processor/AdminBrokerProcessor.java @@ -32,6 +32,7 @@ import java.util.HashSet; import java.util.Iterator; import java.util.List; import java.util.Map; +import java.util.Optional; import java.util.Properties; import java.util.Set; import java.util.concurrent.CompletableFuture; @@ -44,6 +45,7 @@ import java.util.concurrent.SynchronousQueue; import java.util.concurrent.ThreadPoolExecutor; import java.util.concurrent.TimeUnit; import org.apache.commons.collections.CollectionUtils; +import org.apache.commons.lang3.ObjectUtils; import org.apache.commons.lang3.StringUtils; import org.apache.rocketmq.auth.authentication.enums.UserType; import org.apache.rocketmq.auth.authentication.exception.AuthenticationException; @@ -68,6 +70,8 @@ import org.apache.rocketmq.broker.metrics.BrokerMetricsManager; import org.apache.rocketmq.broker.metrics.InvocationStatus; import org.apache.rocketmq.broker.plugin.BrokerAttachedPlugin; import org.apache.rocketmq.broker.subscription.SubscriptionGroupManager; +import org.apache.rocketmq.broker.topic.TopicConfigManager; +import org.apache.rocketmq.broker.topic.TopicQueueMappingManager; import org.apache.rocketmq.broker.transaction.queue.TransactionalMessageUtil; import org.apache.rocketmq.common.BrokerConfig; import org.apache.rocketmq.common.CheckRocksdbCqWriteResult; @@ -137,6 +141,7 @@ import org.apache.rocketmq.remoting.protocol.body.QuerySubscriptionResponseBody; import org.apache.rocketmq.remoting.protocol.body.QueueTimeSpan; import org.apache.rocketmq.remoting.protocol.body.ResetOffsetBody; import org.apache.rocketmq.remoting.protocol.body.SubscriptionGroupList; +import org.apache.rocketmq.remoting.protocol.body.SubscriptionGroupWrapper; import org.apache.rocketmq.remoting.protocol.body.SyncStateSet; import org.apache.rocketmq.remoting.protocol.body.TopicConfigAndMappingSerializeWrapper; import org.apache.rocketmq.remoting.protocol.body.TopicList; @@ -157,6 +162,9 @@ import org.apache.rocketmq.remoting.protocol.header.ExchangeHAInfoResponseHeader import org.apache.rocketmq.remoting.protocol.header.ExportRocksDBConfigToJsonRequestHeader; import org.apache.rocketmq.remoting.protocol.header.GetAclRequestHeader; import org.apache.rocketmq.remoting.protocol.header.GetAllProducerInfoRequestHeader; +import org.apache.rocketmq.remoting.protocol.header.GetAllSubscriptionGroupRequestHeader; +import org.apache.rocketmq.remoting.protocol.header.GetAllSubscriptionGroupResponseHeader; +import org.apache.rocketmq.remoting.protocol.header.GetAllTopicConfigRequestHeader; import org.apache.rocketmq.remoting.protocol.header.GetAllTopicConfigResponseHeader; import org.apache.rocketmq.remoting.protocol.header.GetBrokerConfigResponseHeader; import org.apache.rocketmq.remoting.protocol.header.GetConsumeStatsInBrokerHeader; @@ -816,40 +824,54 @@ public class AdminBrokerProcessor implements NettyRequestProcessor { return response; } - private RemotingCommand getAllTopicConfig(ChannelHandlerContext ctx, RemotingCommand request) { + private RemotingCommand getAllTopicConfig(ChannelHandlerContext ctx, RemotingCommand request) + throws RemotingCommandException { final RemotingCommand response = RemotingCommand.createResponseCommand(GetAllTopicConfigResponseHeader.class); - // final GetAllTopicConfigResponseHeader responseHeader = - // (GetAllTopicConfigResponseHeader) response.readCustomHeader(); + final GetAllTopicConfigResponseHeader responseHeader = + (GetAllTopicConfigResponseHeader) response.readCustomHeader(); + final GetAllTopicConfigRequestHeader requestHeader = + request.decodeCommandCustomHeader(GetAllTopicConfigRequestHeader.class); + + String dataVersionStr = requestHeader.getDataVersion(); + Integer topicSeq = requestHeader.getTopicSeq(); + Integer maxTopicNum = requestHeader.getMaxTopicNum(); + + TopicConfigManager tcManager = brokerController.getTopicConfigManager(); + TopicQueueMappingManager tqmManager = brokerController.getTopicQueueMappingManager(); TopicConfigAndMappingSerializeWrapper topicConfigAndMappingSerializeWrapper = new TopicConfigAndMappingSerializeWrapper(); + if (!brokerController.getBrokerConfig().isEnableSplitMetadata() + || ObjectUtils.allNull(dataVersionStr, topicSeq, maxTopicNum)) { // old client, return all topic config - topicConfigAndMappingSerializeWrapper.setDataVersion(this.brokerController.getTopicConfigManager().getDataVersion()); - topicConfigAndMappingSerializeWrapper.setTopicConfigTable(this.brokerController.getTopicConfigManager().getTopicConfigTable()); + topicConfigAndMappingSerializeWrapper.setDataVersion(tcManager.getDataVersion()); + topicConfigAndMappingSerializeWrapper.setTopicConfigTable(tcManager.getTopicConfigTable()); - topicConfigAndMappingSerializeWrapper.setMappingDataVersion(this.brokerController.getTopicQueueMappingManager().getDataVersion()); - topicConfigAndMappingSerializeWrapper.setTopicQueueMappingDetailMap(this.brokerController.getTopicQueueMappingManager().getTopicQueueMappingTable()); + topicConfigAndMappingSerializeWrapper.setMappingDataVersion(tqmManager.getDataVersion()); + topicConfigAndMappingSerializeWrapper.setTopicQueueMappingDetailMap(tqmManager.getTopicQueueMappingTable()); + } else { + int topicNum = Math.min(brokerController.getBrokerConfig().getSplitMetadataSize(), + Optional.ofNullable(maxTopicNum).orElse(Integer.MAX_VALUE)); // use smaller value + ConcurrentHashMap subTopicConfigTable = + tcManager.subTopicConfigTable(dataVersionStr, topicSeq, topicNum); + topicConfigAndMappingSerializeWrapper.setTopicConfigTable(subTopicConfigTable); + topicConfigAndMappingSerializeWrapper.setDataVersion(tcManager.getDataVersion()); + topicConfigAndMappingSerializeWrapper.setMappingDataVersion(tqmManager.getDataVersion()); + topicConfigAndMappingSerializeWrapper.setTopicQueueMappingDetailMap( + tqmManager.subTopicQueueMappingTable(subTopicConfigTable.keySet())); + } + + responseHeader.setTotalTopicNum(tcManager.getTopicConfigTable().size()); String content = topicConfigAndMappingSerializeWrapper.toJson(); - if (content != null && content.length() > 0) { - try { - response.setBody(content.getBytes(MixAll.DEFAULT_CHARSET)); - } catch (UnsupportedEncodingException e) { - LOGGER.error("", e); - - response.setCode(ResponseCode.SYSTEM_ERROR); - response.setRemark("UnsupportedEncodingException " + e.getMessage()); - return response; - } + if (StringUtils.isNotBlank(content)) { + response.setCode(ResponseCode.SUCCESS); + response.setRemark(null); + response.setBody(content.getBytes(StandardCharsets.UTF_8)); } else { LOGGER.error("No topic in this broker, client: {}", ctx.channel().remoteAddress()); response.setCode(ResponseCode.SYSTEM_ERROR); response.setRemark("No topic in this broker"); - return response; } - - response.setCode(ResponseCode.SUCCESS); - response.setRemark(null); - return response; } @@ -1584,28 +1606,45 @@ public class AdminBrokerProcessor implements NettyRequestProcessor { private RemotingCommand getAllSubscriptionGroup(ChannelHandlerContext ctx, RemotingCommand request) throws RemotingCommandException { - final RemotingCommand response = RemotingCommand.createResponseCommand(null); - String content = this.brokerController.getSubscriptionGroupManager().encode(); - if (content != null && content.length() > 0) { - try { - response.setBody(content.getBytes(MixAll.DEFAULT_CHARSET)); - } catch (UnsupportedEncodingException e) { - LOGGER.error("UnsupportedEncodingException getAllSubscriptionGroup", e); + final RemotingCommand response = RemotingCommand.createResponseCommand(GetAllSubscriptionGroupResponseHeader.class); + final GetAllSubscriptionGroupResponseHeader responseHeader = + (GetAllSubscriptionGroupResponseHeader) response.readCustomHeader(); + final GetAllSubscriptionGroupRequestHeader requestHeader = + request.decodeCommandCustomHeader(GetAllSubscriptionGroupRequestHeader.class); - response.setCode(ResponseCode.SYSTEM_ERROR); - response.setRemark("UnsupportedEncodingException " + e.getMessage()); - return response; - } + String dataVersionStr = requestHeader.getDataVersion(); + Integer groupSeq = requestHeader.getGroupSeq(); + Integer maxGroupNum = requestHeader.getMaxGroupNum(); + + SubscriptionGroupManager sgManager = this.brokerController.getSubscriptionGroupManager(); + + SubscriptionGroupWrapper subscriptionGroupWrapper = new SubscriptionGroupWrapper(); + if (!brokerController.getBrokerConfig().isEnableSplitMetadata() + || ObjectUtils.allNull(dataVersionStr, groupSeq, maxGroupNum)) { + subscriptionGroupWrapper.setSubscriptionGroupTable(sgManager.getSubscriptionGroupTable()); + subscriptionGroupWrapper.setForbiddenTable(sgManager.getForbiddenTable()); + subscriptionGroupWrapper.setDataVersion(sgManager.getDataVersion()); + } else { + int groupNum = Math.min(brokerController.getBrokerConfig().getSplitMetadataSize(), + Optional.ofNullable(maxGroupNum).orElse(Integer.MAX_VALUE)); + ConcurrentMap subGroupTable = + sgManager.subGroupTable(dataVersionStr, groupSeq, groupNum); + subscriptionGroupWrapper.setSubscriptionGroupTable(subGroupTable); + subscriptionGroupWrapper.setDataVersion(sgManager.getDataVersion()); + subscriptionGroupWrapper.setForbiddenTable(sgManager.subForbiddenTable(subGroupTable.keySet())); + } + + responseHeader.setTotalGroupNum(sgManager.getSubscriptionGroupTable().size()); + String content = subscriptionGroupWrapper.toJson(); + if (StringUtils.isNotBlank(content)) { + response.setBody(content.getBytes(StandardCharsets.UTF_8)); + response.setCode(ResponseCode.SUCCESS); + response.setRemark(null); } else { LOGGER.error("No subscription group in this broker, client:{} ", ctx.channel().remoteAddress()); response.setCode(ResponseCode.SYSTEM_ERROR); response.setRemark("No subscription group in this broker"); - return response; } - - response.setCode(ResponseCode.SUCCESS); - response.setRemark(null); - return response; } diff --git a/broker/src/main/java/org/apache/rocketmq/broker/subscription/SubscriptionGroupManager.java b/broker/src/main/java/org/apache/rocketmq/broker/subscription/SubscriptionGroupManager.java index f3e669fb3e..c7083365be 100644 --- a/broker/src/main/java/org/apache/rocketmq/broker/subscription/SubscriptionGroupManager.java +++ b/broker/src/main/java/org/apache/rocketmq/broker/subscription/SubscriptionGroupManager.java @@ -22,8 +22,16 @@ import java.util.Iterator; import java.util.List; import java.util.Map; import java.util.Map.Entry; +import java.util.Objects; +import java.util.Set; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ConcurrentMap; +import java.util.stream.Collectors; + +import com.google.common.collect.ImmutableSortedMap; +import com.google.common.collect.Maps; +import org.apache.commons.collections.CollectionUtils; +import org.apache.commons.collections.MapUtils; import org.apache.commons.lang3.StringUtils; import org.apache.rocketmq.broker.BrokerController; import org.apache.rocketmq.broker.BrokerPathConfigHelper; @@ -322,10 +330,42 @@ public class SubscriptionGroupManager extends ConfigManager { return subscriptionGroupTable; } + public ConcurrentHashMap subGroupTable(String dataVersion, int groupSeq, + int maxGroupNum) { + // [groupSeq, groupSeq + maxGroupNum) + int beginIndex = groupSeq; + if (StringUtils.isBlank(dataVersion) || !Objects.equals(DataVersion.fromJson(dataVersion, DataVersion.class), this.dataVersion)) { + beginIndex = 0; + log.info("get sub subscription group table from {} due to {}", beginIndex, + StringUtils.isBlank(dataVersion) ? "DataVersion Empty" : "DataVersion Changed"); + } + + ConcurrentHashMap subGroupTable = new ConcurrentHashMap<>(); + if (beginIndex < subscriptionGroupTable.size()) { + int endIndex = Math.min(beginIndex + maxGroupNum, subscriptionGroupTable.size()); + + ImmutableSortedMap sortedMap = ImmutableSortedMap.copyOf(subscriptionGroupTable); + subGroupTable.putAll(sortedMap.subMap(sortedMap.keySet().asList().get(beginIndex),true, + sortedMap.keySet().asList().get(endIndex - 1),true)); + } + + return subGroupTable; + } + public ConcurrentMap> getForbiddenTable() { return forbiddenTable; } + public ConcurrentMap> subForbiddenTable(Set groupSet) { + if (MapUtils.isEmpty(forbiddenTable) || CollectionUtils.isEmpty(groupSet)) { + return Maps.newConcurrentMap(); + } + + return forbiddenTable.entrySet().stream() + .filter(e -> groupSet.contains(e.getKey())) + .collect(Collectors.toConcurrentMap(Map.Entry::getKey, Map.Entry::getValue)); + } + public void setForbiddenTable( ConcurrentMap> forbiddenTable) { this.forbiddenTable = forbiddenTable; diff --git a/broker/src/main/java/org/apache/rocketmq/broker/topic/TopicConfigManager.java b/broker/src/main/java/org/apache/rocketmq/broker/topic/TopicConfigManager.java index b20cafc101..8211a8fb09 100644 --- a/broker/src/main/java/org/apache/rocketmq/broker/topic/TopicConfigManager.java +++ b/broker/src/main/java/org/apache/rocketmq/broker/topic/TopicConfigManager.java @@ -21,6 +21,7 @@ import java.util.Iterator; import java.util.List; import java.util.Map; import java.util.Map.Entry; +import java.util.Objects; import java.util.Set; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ConcurrentMap; @@ -29,7 +30,7 @@ import java.util.concurrent.locks.Lock; import java.util.concurrent.locks.ReentrantLock; import com.google.common.collect.ImmutableMap; - +import com.google.common.collect.ImmutableSortedMap; import com.google.common.collect.Maps; import org.apache.commons.lang3.StringUtils; import org.apache.rocketmq.broker.BrokerController; @@ -703,6 +704,28 @@ public class TopicConfigManager extends ConfigManager { return topicConfigTable; } + public ConcurrentHashMap subTopicConfigTable(String dataVersion, int topicSeq, + int maxTopicNum) { + // [topicSeq, topicSeq + maxTopicNum) + int beginIndex = topicSeq; + if (StringUtils.isBlank(dataVersion) || !Objects.equals(DataVersion.fromJson(dataVersion, DataVersion.class), this.dataVersion)) { + beginIndex = 0; + log.info("get sub topic config table from {} due to {}", beginIndex, + StringUtils.isBlank(dataVersion) ? "DataVersion Empty" : "DataVersion Changed"); + } + + ConcurrentHashMap subTopicConfigTable = new ConcurrentHashMap<>(); + if (beginIndex < topicConfigTable.size()) { + int endIndex = Math.min(beginIndex + maxTopicNum, topicConfigTable.size()); + + ImmutableSortedMap sortedMap = ImmutableSortedMap.copyOf(topicConfigTable); + subTopicConfigTable.putAll(sortedMap.subMap(sortedMap.keySet().asList().get(beginIndex),true, + sortedMap.keySet().asList().get(endIndex - 1),true)); + } + + return subTopicConfigTable; + } + private Map request(TopicConfig topicConfig) { return topicConfig.getAttributes() == null ? new HashMap<>() : topicConfig.getAttributes(); } diff --git a/broker/src/main/java/org/apache/rocketmq/broker/topic/TopicQueueMappingManager.java b/broker/src/main/java/org/apache/rocketmq/broker/topic/TopicQueueMappingManager.java index 6b9cf15938..4b0714decb 100644 --- a/broker/src/main/java/org/apache/rocketmq/broker/topic/TopicQueueMappingManager.java +++ b/broker/src/main/java/org/apache/rocketmq/broker/topic/TopicQueueMappingManager.java @@ -17,12 +17,19 @@ package org.apache.rocketmq.broker.topic; import com.alibaba.fastjson.JSON; +import com.google.common.collect.Maps; import java.util.List; +import java.util.Map; +import java.util.Set; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ConcurrentMap; import java.util.concurrent.TimeUnit; import java.util.concurrent.locks.Lock; import java.util.concurrent.locks.ReentrantLock; +import java.util.stream.Collectors; + +import org.apache.commons.collections.CollectionUtils; +import org.apache.commons.collections.MapUtils; import org.apache.rocketmq.broker.BrokerController; import org.apache.rocketmq.broker.BrokerPathConfigHelper; import org.apache.rocketmq.common.ConfigManager; @@ -178,6 +185,16 @@ public class TopicQueueMappingManager extends ConfigManager { return topicQueueMappingTable; } + public ConcurrentMap subTopicQueueMappingTable(Set topicSet) { + if (MapUtils.isEmpty(topicQueueMappingTable) || CollectionUtils.isEmpty(topicSet)) { + return Maps.newConcurrentMap(); + } + + return topicQueueMappingTable.entrySet().stream() + .filter(e -> topicSet.contains(e.getKey())) + .collect(Collectors.toConcurrentMap(Map.Entry::getKey, Map.Entry::getValue)); + } + public DataVersion getDataVersion() { return dataVersion; } diff --git a/broker/src/test/java/org/apache/rocketmq/broker/processor/AdminBrokerProcessorTest.java b/broker/src/test/java/org/apache/rocketmq/broker/processor/AdminBrokerProcessorTest.java index a6bcca954d..f3d0eb0782 100644 --- a/broker/src/test/java/org/apache/rocketmq/broker/processor/AdminBrokerProcessorTest.java +++ b/broker/src/test/java/org/apache/rocketmq/broker/processor/AdminBrokerProcessorTest.java @@ -17,9 +17,12 @@ package org.apache.rocketmq.broker.processor; import com.alibaba.fastjson.JSON; +import com.google.common.collect.ImmutableMap; +import com.google.common.collect.Maps; import com.google.common.collect.Sets; import io.netty.channel.Channel; import io.netty.channel.ChannelHandlerContext; +import org.apache.commons.lang3.StringUtils; import org.apache.rocketmq.auth.authentication.enums.UserType; import org.apache.rocketmq.auth.authentication.manager.AuthenticationMetadataManager; import org.apache.rocketmq.auth.authentication.model.Subject; @@ -61,6 +64,7 @@ import org.apache.rocketmq.remoting.exception.RemotingSendRequestException; import org.apache.rocketmq.remoting.exception.RemotingTimeoutException; import org.apache.rocketmq.remoting.netty.NettyClientConfig; import org.apache.rocketmq.remoting.netty.NettyServerConfig; +import org.apache.rocketmq.remoting.protocol.DataVersion; import org.apache.rocketmq.remoting.protocol.RemotingCommand; import org.apache.rocketmq.remoting.protocol.RemotingSerializable; import org.apache.rocketmq.remoting.protocol.RequestCode; @@ -71,6 +75,8 @@ import org.apache.rocketmq.remoting.protocol.body.GroupList; import org.apache.rocketmq.remoting.protocol.body.HARuntimeInfo; import org.apache.rocketmq.remoting.protocol.body.LockBatchRequestBody; import org.apache.rocketmq.remoting.protocol.body.QueryCorrectionOffsetBody; +import org.apache.rocketmq.remoting.protocol.body.SubscriptionGroupWrapper; +import org.apache.rocketmq.remoting.protocol.body.TopicConfigSerializeWrapper; import org.apache.rocketmq.remoting.protocol.body.UnlockBatchRequestBody; import org.apache.rocketmq.remoting.protocol.body.UserInfo; import org.apache.rocketmq.remoting.protocol.header.CreateAclRequestHeader; @@ -81,6 +87,9 @@ import org.apache.rocketmq.remoting.protocol.header.DeleteTopicRequestHeader; import org.apache.rocketmq.remoting.protocol.header.DeleteUserRequestHeader; import org.apache.rocketmq.remoting.protocol.header.ExchangeHAInfoResponseHeader; import org.apache.rocketmq.remoting.protocol.header.GetAclRequestHeader; +import org.apache.rocketmq.remoting.protocol.header.GetAllSubscriptionGroupRequestHeader; +import org.apache.rocketmq.remoting.protocol.header.GetAllSubscriptionGroupResponseHeader; +import org.apache.rocketmq.remoting.protocol.header.GetAllTopicConfigRequestHeader; import org.apache.rocketmq.remoting.protocol.header.GetAllTopicConfigResponseHeader; import org.apache.rocketmq.remoting.protocol.header.GetConsumerRunningInfoRequestHeader; import org.apache.rocketmq.remoting.protocol.header.GetConsumerStatusRequestHeader; @@ -137,11 +146,15 @@ import java.util.HashMap; import java.util.HashSet; import java.util.List; import java.util.Map; +import java.util.Objects; +import java.util.Optional; import java.util.Properties; import java.util.Set; import java.util.concurrent.CompletableFuture; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ConcurrentMap; +import java.util.concurrent.ThreadLocalRandom; +import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.LongAdder; import static org.assertj.core.api.Assertions.assertThat; @@ -455,12 +468,166 @@ public class AdminBrokerProcessorTest { @Test public void testGetAllTopicConfig() throws Exception { - GetAllTopicConfigResponseHeader getAllTopicConfigResponseHeader = new GetAllTopicConfigResponseHeader(); + GetAllTopicConfigRequestHeader getAllTopicConfigResponseHeader = new GetAllTopicConfigRequestHeader(); RemotingCommand request = RemotingCommand.createRequestCommand(RequestCode.GET_ALL_TOPIC_CONFIG, getAllTopicConfigResponseHeader); RemotingCommand response = adminBrokerProcessor.processRequest(handlerContext, request); assertThat(response.getCode()).isEqualTo(ResponseCode.SUCCESS); } + + private void getAllTopicConfig(boolean enableSplitMetadata) throws RemotingCommandException { + brokerController.getBrokerConfig().setEnableSplitMetadata(enableSplitMetadata); + + // old client, request null + RemotingCommand requestOldClient = RemotingCommand.createRequestCommand(RequestCode.GET_ALL_TOPIC_CONFIG, null); + RemotingCommand responseOldClient = adminBrokerProcessor.processRequest(handlerContext, requestOldClient); + assertThat(responseOldClient.getCode()).isEqualTo(ResponseCode.SUCCESS); + + TopicConfigSerializeWrapper topicConfigSerializeWrapperOldClient = + TopicConfigSerializeWrapper.decode(responseOldClient.getBody(), TopicConfigSerializeWrapper.class); + assertThat(Maps.difference(topicConfigSerializeWrapperOldClient.getTopicConfigTable(), + brokerController.getTopicConfigManager().getTopicConfigTable()).areEqual()).isTrue(); + + // new client, request seq from 0 + AtomicBoolean dataVersionChanged = new AtomicBoolean(false); + int topicSeq = 0; + DataVersion dataVersion = null; + int pageSize = ThreadLocalRandom.current().nextInt(500, brokerController.getBrokerConfig().getSplitMetadataSize()); + ConcurrentMap topicConfigTable = new ConcurrentHashMap<>(); + while (true) { + GetAllTopicConfigRequestHeader requestHeader = new GetAllTopicConfigRequestHeader(); + requestHeader.setTopicSeq(topicSeq); + requestHeader.setMaxTopicNum(pageSize); + requestHeader.setDataVersion(Optional.ofNullable(dataVersion).map(DataVersion::toJson).orElse(StringUtils.EMPTY)); + RemotingCommand requestNewClient = RemotingCommand.createRequestCommand(RequestCode.GET_ALL_TOPIC_CONFIG, requestHeader); + requestNewClient.makeCustomHeaderToNet(); + + RemotingCommand responseNewClient = adminBrokerProcessor.processRequest(handlerContext, requestNewClient); + GetAllTopicConfigResponseHeader responseHeader = (GetAllTopicConfigResponseHeader) responseNewClient.readCustomHeader(); + assertThat(responseNewClient.getCode()).isEqualTo(ResponseCode.SUCCESS); + + TopicConfigSerializeWrapper topicConfigSerializeWrapper = + TopicConfigSerializeWrapper.decode(responseNewClient.getBody(), TopicConfigSerializeWrapper.class); + topicSeq += topicConfigSerializeWrapper.getTopicConfigTable().size(); + + assertThat(responseHeader.getTotalTopicNum()) + .isEqualTo(brokerController.getTopicConfigManager().getTopicConfigTable().size()); + assertThat(topicConfigSerializeWrapper.getDataVersion()) + .isEqualTo(brokerController.getTopicConfigManager().getDataVersion()); + + DataVersion newDataVersion = topicConfigSerializeWrapper.getDataVersion(); + if (dataVersion == null) { + dataVersion = newDataVersion; + } + + // mock server side data version changed + if (topicSeq > responseHeader.getTotalTopicNum() / 2 && dataVersionChanged.compareAndSet(false, true)) { + brokerController.getTopicConfigManager().getDataVersion().nextVersion(); + } + + if (!Objects.equals(dataVersion, newDataVersion)) { + dataVersion = newDataVersion; + topicSeq = 0; // data version diff, from 0 + topicConfigTable.clear(); + continue; + } + + + topicConfigTable.putAll(topicConfigSerializeWrapper.getTopicConfigTable()); + if (topicSeq >= responseHeader.getTotalTopicNum() - 1) { + break; + } else { + assertThat(topicConfigSerializeWrapper.getTopicConfigTable().size()).isEqualTo(pageSize); + } + } + assertThat(Maps.difference(topicConfigTable, brokerController.getTopicConfigManager().getTopicConfigTable()).areEqual()).isTrue(); + } + + @Test + public void testGetAllTopicConfigWithRequestHeader() throws RemotingCommandException { + // from [0, 50000) + fillTopicConfigTable(50000); + + getAllTopicConfig(true); + getAllTopicConfig(false); // broker side disable split , will return all topic config + } + + + private void getAllSubscriptionGroup(boolean enableSplitMetadata) throws RemotingCommandException { + brokerController.getBrokerConfig().setEnableSplitMetadata(enableSplitMetadata); + + // old client, request null + RemotingCommand requestOldClient = RemotingCommand.createRequestCommand(RequestCode.GET_ALL_SUBSCRIPTIONGROUP_CONFIG, null); + RemotingCommand responseOldClient = adminBrokerProcessor.processRequest(handlerContext, requestOldClient); + assertThat(responseOldClient.getCode()).isEqualTo(ResponseCode.SUCCESS); + + // new client, request from 0 + AtomicBoolean dataVersionChanged = new AtomicBoolean(false); + int groupSeq = 0; + int pageSize = ThreadLocalRandom.current().nextInt(500, brokerController.getBrokerConfig().getSplitMetadataSize()); + DataVersion dataVersion = null; + ConcurrentMap subscriptionGroupTable = new ConcurrentHashMap<>(); + ConcurrentMap> forbiddenTable = new ConcurrentHashMap<>(); + while (true) { + GetAllSubscriptionGroupRequestHeader requestHeader = new GetAllSubscriptionGroupRequestHeader(); + requestHeader.setGroupSeq(groupSeq); + requestHeader.setMaxGroupNum(pageSize); + requestHeader.setDataVersion(Optional.ofNullable(dataVersion).map(DataVersion::toJson).orElse(StringUtils.EMPTY)); + RemotingCommand requestNewClient = RemotingCommand.createRequestCommand(RequestCode.GET_ALL_SUBSCRIPTIONGROUP_CONFIG, requestHeader); + requestNewClient.makeCustomHeaderToNet(); + RemotingCommand responseNewClient = adminBrokerProcessor.processRequest(handlerContext, requestNewClient); + GetAllSubscriptionGroupResponseHeader responseHeader = (GetAllSubscriptionGroupResponseHeader) responseNewClient.readCustomHeader(); + assertThat(responseNewClient.getCode()).isEqualTo(ResponseCode.SUCCESS); + + SubscriptionGroupWrapper subscriptionGroupWrapper = + SubscriptionGroupWrapper.decode(responseNewClient.getBody(), SubscriptionGroupWrapper.class); + + groupSeq += subscriptionGroupWrapper.getSubscriptionGroupTable().size(); + DataVersion newDataVersion = subscriptionGroupWrapper.getDataVersion(); + + assertThat(responseHeader.getTotalGroupNum()).isEqualTo( + brokerController.getSubscriptionGroupManager().getSubscriptionGroupTable().size()); + assertThat(newDataVersion).isEqualTo(brokerController.getSubscriptionGroupManager().getDataVersion()); + + if (dataVersion == null) { + dataVersion = newDataVersion; + } + + + // mock server side data version changed + if (groupSeq > responseHeader.getTotalGroupNum() / 2 && dataVersionChanged.compareAndSet(false, true)) { + brokerController.getSubscriptionGroupManager().getDataVersion().nextVersion(); + } + + if (!Objects.equals(dataVersion, newDataVersion)) { + dataVersion = newDataVersion; + groupSeq = 0; // data version diff, from 0 + subscriptionGroupTable.clear(); + forbiddenTable.clear(); + continue; + } + + subscriptionGroupTable.putAll(subscriptionGroupWrapper.getSubscriptionGroupTable()); + forbiddenTable.putAll(subscriptionGroupWrapper.getForbiddenTable()); + if (groupSeq >= responseHeader.getTotalGroupNum() - 1) { + break; + } else { + assertThat(subscriptionGroupWrapper.getSubscriptionGroupTable().size()).isEqualTo(pageSize); + } + } + assertThat(Maps.difference(subscriptionGroupTable, brokerController.getSubscriptionGroupManager().getSubscriptionGroupTable()).areEqual()).isTrue(); + assertThat(Maps.difference(forbiddenTable, brokerController.getSubscriptionGroupManager().getForbiddenTable()).areEqual()).isTrue(); + } + + @Test + public void testGetAllSubscriptionGroupWithRequestHeader() throws RemotingCommandException { + fillSubscriptionGroupManager(50000); + + getAllSubscriptionGroup(true); + getAllSubscriptionGroup(false); + + } + @Test public void testUpdateBrokerConfig() throws Exception { handlerContext = mock(ChannelHandlerContext.class); @@ -1434,4 +1601,24 @@ public class AdminBrokerProcessorTest { private boolean notToBeExecuted() { return MixAll.isMac(); } + + private void fillTopicConfigTable(int num) { + for (int i = num - 1; i >= 0; i--) { + String topicName = String.format("topic%05d", i); + TopicConfig topicConfig = new TopicConfig(topicName, 1, 1, + PermName.PERM_READ | PermName.PERM_WRITE, 0); + brokerController.getTopicConfigManager().getTopicConfigTable().put(topicName, topicConfig); + } + } + + private void fillSubscriptionGroupManager(int num) { + for (int i = num - 1; i >= 0; i--) { + SubscriptionGroupConfig subscriptionGroupConfig = new SubscriptionGroupConfig(); + String groupName = String.format("group-%05d", i); + subscriptionGroupConfig.setGroupName(groupName); + Map attr = ImmutableMap.of("+test", "true"); + subscriptionGroupConfig.setAttributes(attr); + brokerController.getSubscriptionGroupManager().getSubscriptionGroupTable().put(groupName, subscriptionGroupConfig); + } + } } diff --git a/broker/src/test/java/org/apache/rocketmq/broker/slave/SlaveSynchronizeAtomicTest.java b/broker/src/test/java/org/apache/rocketmq/broker/slave/SlaveSynchronizeAtomicTest.java index 75db22e7e7..714dcd967f 100644 --- a/broker/src/test/java/org/apache/rocketmq/broker/slave/SlaveSynchronizeAtomicTest.java +++ b/broker/src/test/java/org/apache/rocketmq/broker/slave/SlaveSynchronizeAtomicTest.java @@ -29,6 +29,7 @@ import org.apache.rocketmq.broker.subscription.SubscriptionGroupManager; import org.apache.rocketmq.broker.topic.TopicConfigManager; import org.apache.rocketmq.client.exception.MQBrokerException; import org.apache.rocketmq.common.BrokerConfig; +import org.apache.rocketmq.remoting.exception.RemotingCommandException; import org.apache.rocketmq.remoting.exception.RemotingConnectException; import org.apache.rocketmq.remoting.exception.RemotingSendRequestException; import org.apache.rocketmq.remoting.exception.RemotingTimeoutException; @@ -113,8 +114,8 @@ public class SlaveSynchronizeAtomicTest { @Test public void testSyncAtomically() - throws RemotingConnectException, RemotingSendRequestException, RemotingTimeoutException, MQBrokerException, - InterruptedException { + throws RemotingConnectException, RemotingSendRequestException, RemotingTimeoutException, MQBrokerException, + InterruptedException, RemotingCommandException { when(brokerOuterAPI.getAllSubscriptionGroupConfig(anyString())).thenReturn(subscriptionGroupWrapper); when(brokerOuterAPI.getAllMessageRequestMode(anyString())).thenReturn(requestModeSerializeWrapper); diff --git a/broker/src/test/java/org/apache/rocketmq/broker/slave/SlaveSynchronizeTest.java b/broker/src/test/java/org/apache/rocketmq/broker/slave/SlaveSynchronizeTest.java index 95db733d0d..c9461c4224 100644 --- a/broker/src/test/java/org/apache/rocketmq/broker/slave/SlaveSynchronizeTest.java +++ b/broker/src/test/java/org/apache/rocketmq/broker/slave/SlaveSynchronizeTest.java @@ -27,6 +27,7 @@ import org.apache.rocketmq.broker.topic.TopicConfigManager; import org.apache.rocketmq.client.exception.MQBrokerException; import org.apache.rocketmq.common.BrokerConfig; import org.apache.rocketmq.common.TopicConfig; +import org.apache.rocketmq.remoting.exception.RemotingCommandException; import org.apache.rocketmq.remoting.exception.RemotingConnectException; import org.apache.rocketmq.remoting.exception.RemotingSendRequestException; import org.apache.rocketmq.remoting.exception.RemotingTimeoutException; @@ -132,7 +133,8 @@ public class SlaveSynchronizeTest { } @Test - public void testSyncAll() throws RemotingConnectException, RemotingSendRequestException, RemotingTimeoutException, MQBrokerException, InterruptedException, UnsupportedEncodingException { + public void testSyncAll() throws RemotingConnectException, RemotingSendRequestException, RemotingTimeoutException, + MQBrokerException, InterruptedException, UnsupportedEncodingException, RemotingCommandException { TopicConfig newTopicConfig = new TopicConfig("NewTopic"); when(brokerOuterAPI.getAllTopicConfig(anyString())).thenReturn(createTopicConfigWrapper(newTopicConfig)); when(brokerOuterAPI.getAllConsumerOffset(anyString())).thenReturn(createConsumerOffsetWrapper()); diff --git a/broker/src/test/java/org/apache/rocketmq/broker/subscription/SubscriptionGroupManagerTest.java b/broker/src/test/java/org/apache/rocketmq/broker/subscription/SubscriptionGroupManagerTest.java index 3384d479c6..3c975a599b 100644 --- a/broker/src/test/java/org/apache/rocketmq/broker/subscription/SubscriptionGroupManagerTest.java +++ b/broker/src/test/java/org/apache/rocketmq/broker/subscription/SubscriptionGroupManagerTest.java @@ -24,13 +24,18 @@ import java.util.LinkedList; import java.util.List; import java.util.Map; import java.util.UUID; +import java.util.concurrent.ThreadLocalRandom; +import java.util.concurrent.atomic.AtomicLong; + import org.apache.rocketmq.broker.BrokerController; import org.apache.rocketmq.common.MixAll; import org.apache.rocketmq.common.SubscriptionGroupAttributes; import org.apache.rocketmq.common.attribute.BooleanAttribute; +import org.apache.rocketmq.remoting.protocol.DataVersion; import org.apache.rocketmq.remoting.protocol.subscription.SubscriptionGroupConfig; import org.apache.rocketmq.store.config.MessageStoreConfig; import org.junit.After; +import org.junit.Assert; import org.junit.Before; import org.junit.Test; import org.junit.runner.RunWith; @@ -45,7 +50,6 @@ import static org.mockito.Mockito.spy; import static org.mockito.Mockito.times; import static org.mockito.Mockito.verify; - @RunWith(MockitoJUnitRunner.class) public class SubscriptionGroupManagerTest { private String group = "group"; @@ -58,9 +62,6 @@ public class SubscriptionGroupManagerTest { @Before public void before() { - if (notToBeExecuted()) { - return; - } SubscriptionGroupAttributes.ALL.put("test", new BooleanAttribute( "test", false, @@ -166,4 +167,75 @@ public class SubscriptionGroupManagerTest { } + @Test + public void testSubGroupTable() { + // Empty SubscriptionGroupManager + subscriptionGroupManager.getSubscriptionGroupTable().clear(); + Map result = + subscriptionGroupManager.subGroupTable(subscriptionGroupManager.getDataVersion().toJson(), 0, 200); + assertThat(result).isEmpty(); + + // fill SubscriptionGroupManager + int totalGroupNum = 50000; + fillSubscriptionGroupManager(totalGroupNum); + + // Null DataVersion + int beginIndex = 0, maxNum = 200; + int endIndex = beginIndex + maxNum - 1; + result = subscriptionGroupManager.subGroupTable(null, beginIndex, maxNum); + + Assert.assertEquals(maxNum, result.size()); + Assert.assertTrue(result.containsKey(String.format("group-%05d", ThreadLocalRandom.current().nextInt(beginIndex, endIndex)))); + Assert.assertFalse(result.containsKey(String.format("group-%05d", beginIndex - 1))); + Assert.assertFalse(result.containsKey(String.format("group-%05d", endIndex + 1))); + + // Different DataVersion + DataVersion differentVersion = new DataVersion(); + differentVersion.setCounter(new AtomicLong(1000L)); // different counter + differentVersion.setTimestamp(System.currentTimeMillis()); + result = subscriptionGroupManager.subGroupTable(differentVersion.toJson(), 300, maxNum); + + Assert.assertEquals(maxNum, result.size()); + Assert.assertTrue(result.containsKey(String.format("group-%05d", ThreadLocalRandom.current().nextInt(beginIndex, endIndex)))); + Assert.assertFalse(result.containsKey(String.format("group-%05d", beginIndex - 1))); + Assert.assertFalse(result.containsKey(String.format("group-%05d", endIndex + 1))); + + // BeginIndexOutOfRange + result = subscriptionGroupManager.subGroupTable(subscriptionGroupManager.getDataVersion().toJson(), totalGroupNum, 200); + + Assert.assertTrue(result.isEmpty()); + + // Normal Case + beginIndex = 300; + endIndex = beginIndex + maxNum - 1; + result = subscriptionGroupManager.subGroupTable(subscriptionGroupManager.getDataVersion().toJson(), beginIndex, maxNum); + + Assert.assertEquals(maxNum, result.size()); + Assert.assertTrue(result.containsKey(String.format("group-%05d", ThreadLocalRandom.current().nextInt(beginIndex, endIndex)))); + Assert.assertFalse(result.containsKey(String.format("group-%05d", beginIndex - 1))); + Assert.assertFalse(result.containsKey(String.format("group-%05d", endIndex + 1))); + + // NotFullTopicConfigTable + beginIndex = 49950; + endIndex = Math.min(subscriptionGroupManager.getSubscriptionGroupTable().size() - 1, beginIndex + maxNum - 1); + result = subscriptionGroupManager.subGroupTable(subscriptionGroupManager.getDataVersion().toJson(), beginIndex, maxNum); + + Assert.assertEquals(totalGroupNum - beginIndex, result.size()); + Assert.assertTrue(result.containsKey(String.format("group-%05d", ThreadLocalRandom.current().nextInt(beginIndex, endIndex)))); + Assert.assertFalse(result.containsKey(String.format("group-%05d", beginIndex - 1))); + Assert.assertFalse(result.containsKey(String.format("group-%05d", endIndex + 1))); + + } + + private void fillSubscriptionGroupManager(int num) { + for (int i = num - 1; i >= 0; i--) { + SubscriptionGroupConfig subscriptionGroupConfig = new SubscriptionGroupConfig(); + String groupName = String.format("group-%05d", i); + subscriptionGroupConfig.setGroupName(groupName); + Map attr = ImmutableMap.of("+test", "true"); + subscriptionGroupConfig.setAttributes(attr); + subscriptionGroupManager.getSubscriptionGroupTable().put(groupName, subscriptionGroupConfig); + } + } + } \ No newline at end of file diff --git a/broker/src/test/java/org/apache/rocketmq/broker/topic/TopicConfigManagerTest.java b/broker/src/test/java/org/apache/rocketmq/broker/topic/TopicConfigManagerTest.java index 3fd1d14c3a..5b2ea0b4d5 100644 --- a/broker/src/test/java/org/apache/rocketmq/broker/topic/TopicConfigManagerTest.java +++ b/broker/src/test/java/org/apache/rocketmq/broker/topic/TopicConfigManagerTest.java @@ -22,6 +22,9 @@ import java.util.List; import java.util.Map; import java.util.Optional; import java.util.UUID; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ThreadLocalRandom; +import java.util.concurrent.atomic.AtomicLong; import org.apache.rocketmq.broker.BrokerController; import org.apache.rocketmq.common.BrokerConfig; @@ -32,7 +35,9 @@ import org.apache.rocketmq.common.attribute.BooleanAttribute; import org.apache.rocketmq.common.attribute.CQType; import org.apache.rocketmq.common.attribute.EnumAttribute; import org.apache.rocketmq.common.attribute.LongRangeAttribute; +import org.apache.rocketmq.common.constant.PermName; import org.apache.rocketmq.common.utils.QueueTypeUtils; +import org.apache.rocketmq.remoting.protocol.DataVersion; import org.apache.rocketmq.store.DefaultMessageStore; import org.apache.rocketmq.store.config.MessageStoreConfig; import org.junit.Assert; @@ -326,4 +331,73 @@ public class TopicConfigManagerTest { TopicAttributes.ALL.putAll(supportedAttributes); } + + private void fillTopicConfigTable(int num) { + ConcurrentHashMap topicConfigTable = new ConcurrentHashMap<>(); + for (int i = num - 1; i >= 0; i--) { + String topicName = String.format("topic%05d", i); + TopicConfig topicConfig = new TopicConfig(topicName, 1, 1, + PermName.PERM_READ | PermName.PERM_WRITE, 0); + topicConfigTable.put(topicName, topicConfig); + } + topicConfigManager.setTopicConfigTable(topicConfigTable); + } + + @Test + public void testSubTopicConfigTable() { + // Empty TopicConfigTable + topicConfigManager.getTopicConfigTable().clear(); + Map result = topicConfigManager.subTopicConfigTable(topicConfigManager.getDataVersion().toJson(), 0, 200); + Assert.assertTrue(result.isEmpty()); + + // init table, topic range [0, N) + final int totalTopicNum = 50000; + fillTopicConfigTable(totalTopicNum); + + // Null DataVersion + int beginIndex = 0, maxNum = 200; + int endIndex = beginIndex + maxNum - 1; + result = topicConfigManager.subTopicConfigTable(null, beginIndex, maxNum); + + Assert.assertEquals(maxNum, result.size()); + Assert.assertTrue(result.containsKey(String.format("topic%05d", ThreadLocalRandom.current().nextInt(beginIndex, endIndex)))); + Assert.assertFalse(result.containsKey(String.format("topic%05d", beginIndex - 1))); + Assert.assertFalse(result.containsKey(String.format("topic%05d", endIndex + 1))); + + // Different DataVersion + DataVersion differentVersion = new DataVersion(); + differentVersion.setCounter(new AtomicLong(1000L)); // different counter + differentVersion.setTimestamp(System.currentTimeMillis()); + result = topicConfigManager.subTopicConfigTable(differentVersion.toJson(), 300, maxNum); + + Assert.assertEquals(maxNum, result.size()); + Assert.assertTrue(result.containsKey(String.format("topic%05d", ThreadLocalRandom.current().nextInt(beginIndex, endIndex)))); + Assert.assertFalse(result.containsKey(String.format("topic%05d", beginIndex - 1))); + Assert.assertFalse(result.containsKey(String.format("topic%05d", endIndex + 1))); + + // BeginIndexOutOfRange + result = topicConfigManager.subTopicConfigTable(topicConfigManager.getDataVersion().toJson(), totalTopicNum, 200); + + Assert.assertTrue(result.isEmpty()); + + // Normal Case + beginIndex = 300; + endIndex = beginIndex + maxNum - 1; + result = topicConfigManager.subTopicConfigTable(topicConfigManager.getDataVersion().toJson(), beginIndex, maxNum); + + Assert.assertEquals(maxNum, result.size()); + Assert.assertTrue(result.containsKey(String.format("topic%05d", ThreadLocalRandom.current().nextInt(beginIndex, endIndex)))); + Assert.assertFalse(result.containsKey(String.format("topic%05d", beginIndex - 1))); + Assert.assertFalse(result.containsKey(String.format("topic%05d", endIndex + 1))); + + // NotFullTopicConfigTable + beginIndex = 49950; + endIndex = Math.min(topicConfigManager.getTopicConfigTable().size() - 1, beginIndex + maxNum - 1); + result = topicConfigManager.subTopicConfigTable(topicConfigManager.getDataVersion().toJson(), beginIndex, maxNum); + + Assert.assertEquals(totalTopicNum - beginIndex, result.size()); + Assert.assertTrue(result.containsKey(String.format("topic%05d", ThreadLocalRandom.current().nextInt(beginIndex, endIndex)))); + Assert.assertFalse(result.containsKey(String.format("topic%05d", beginIndex - 1))); + Assert.assertFalse(result.containsKey(String.format("topic%05d", endIndex + 1))); + } } diff --git a/client/src/main/java/org/apache/rocketmq/client/ClientConfig.java b/client/src/main/java/org/apache/rocketmq/client/ClientConfig.java index 696b073b37..79cb04af1d 100644 --- a/client/src/main/java/org/apache/rocketmq/client/ClientConfig.java +++ b/client/src/main/java/org/apache/rocketmq/client/ClientConfig.java @@ -111,6 +111,8 @@ public class ClientConfig { */ protected String traceTopic; + protected int maxPageSizeInGetMetadata = 2000; + public String buildMQClientId() { StringBuilder sb = new StringBuilder(); sb.append(this.getClientIP()); @@ -515,6 +517,14 @@ public class ClientConfig { this.traceTopic = traceTopic; } + public int getMaxPageSizeInGetMetadata() { + return maxPageSizeInGetMetadata; + } + + public void setMaxPageSizeInGetMetadata(int maxPageSizeInGetMetadata) { + this.maxPageSizeInGetMetadata = maxPageSizeInGetMetadata; + } + @Override public String toString() { return "ClientConfig{" + diff --git a/client/src/main/java/org/apache/rocketmq/client/impl/MQClientAPIImpl.java b/client/src/main/java/org/apache/rocketmq/client/impl/MQClientAPIImpl.java index c001b33fa9..95bb0e8a96 100644 --- a/client/src/main/java/org/apache/rocketmq/client/impl/MQClientAPIImpl.java +++ b/client/src/main/java/org/apache/rocketmq/client/impl/MQClientAPIImpl.java @@ -85,6 +85,7 @@ import org.apache.rocketmq.remoting.exception.RemotingTooMuchRequestException; import org.apache.rocketmq.remoting.netty.NettyClientConfig; import org.apache.rocketmq.remoting.netty.NettyRemotingClient; import org.apache.rocketmq.remoting.netty.ResponseFuture; +import org.apache.rocketmq.remoting.protocol.DataVersion; import org.apache.rocketmq.remoting.protocol.LanguageCode; import org.apache.rocketmq.remoting.protocol.NamespaceUtil; import org.apache.rocketmq.remoting.protocol.RemotingCommand; @@ -151,6 +152,10 @@ import org.apache.rocketmq.remoting.protocol.header.ExportRocksDBConfigToJsonReq import org.apache.rocketmq.remoting.protocol.header.ExtraInfoUtil; import org.apache.rocketmq.remoting.protocol.header.GetAclRequestHeader; import org.apache.rocketmq.remoting.protocol.header.GetAllProducerInfoRequestHeader; +import org.apache.rocketmq.remoting.protocol.header.GetAllSubscriptionGroupRequestHeader; +import org.apache.rocketmq.remoting.protocol.header.GetAllSubscriptionGroupResponseHeader; +import org.apache.rocketmq.remoting.protocol.header.GetAllTopicConfigRequestHeader; +import org.apache.rocketmq.remoting.protocol.header.GetAllTopicConfigResponseHeader; import org.apache.rocketmq.remoting.protocol.header.GetConsumeStatsInBrokerHeader; import org.apache.rocketmq.remoting.protocol.header.GetConsumeStatsRequestHeader; import org.apache.rocketmq.remoting.protocol.header.GetConsumerConnectionListRequestHeader; @@ -242,8 +247,13 @@ import java.util.HashMap; import java.util.Iterator; import java.util.List; import java.util.Map; +import java.util.Objects; +import java.util.Optional; import java.util.Properties; import java.util.Set; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ConcurrentMap; +import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicInteger; import static org.apache.rocketmq.remoting.protocol.RemotingSysResponseCode.SUCCESS; @@ -2806,21 +2816,81 @@ public class MQClientAPIImpl implements NameServerUpdateCallback, StartAndShutdo throw new MQClientException(response.getCode(), response.getRemark()); } - public SubscriptionGroupWrapper getAllSubscriptionGroup(final String brokerAddr, - long timeoutMillis) throws InterruptedException, - RemotingTimeoutException, RemotingSendRequestException, RemotingConnectException, MQBrokerException { - RemotingCommand request = RemotingCommand.createRequestCommand(RequestCode.GET_ALL_SUBSCRIPTIONGROUP_CONFIG, null); - RemotingCommand response = this.remotingClient - .invokeSync(MixAll.brokerVIPChannel(this.clientConfig.isVipChannelEnabled(), brokerAddr), request, timeoutMillis); - assert response != null; - switch (response.getCode()) { - case ResponseCode.SUCCESS: { - return SubscriptionGroupWrapper.decode(response.getBody(), SubscriptionGroupWrapper.class); + public SubscriptionGroupWrapper getAllSubscriptionGroup(final String brokerAddr, long timeoutMillis) + throws InterruptedException, RemotingTimeoutException, RemotingSendRequestException, + RemotingConnectException, MQBrokerException, RemotingCommandException { + + DataVersion currentDataVersion = null; + int groupSeq = 0; + ConcurrentMap subscriptionGroupTable = new ConcurrentHashMap<>(); + ConcurrentMap> forbiddenTable = new ConcurrentHashMap<>(); + long beginTime = System.nanoTime(); + while (true) { + long leftTime = timeoutMillis - TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - beginTime); + if (leftTime < 0) { + throw new RemotingTimeoutException("invokeSync call timeout"); + } + + GetAllSubscriptionGroupRequestHeader requestHeader = new GetAllSubscriptionGroupRequestHeader(); + requestHeader.setGroupSeq(groupSeq); + requestHeader.setMaxGroupNum(clientConfig.getMaxPageSizeInGetMetadata()); + requestHeader.setDataVersion(Optional.ofNullable(currentDataVersion) + .map(DataVersion::toJson).orElse(StringUtils.EMPTY)); + log.info("getAllSubscriptionGroup from seq {}, max {}, dataVersion {}", + groupSeq, requestHeader.getMaxGroupNum(), requestHeader.getDataVersion()); + RemotingCommand request = RemotingCommand.createRequestCommand(RequestCode.GET_ALL_SUBSCRIPTIONGROUP_CONFIG, requestHeader); + RemotingCommand response = this.remotingClient.invokeSync( + MixAll.brokerVIPChannel(this.clientConfig.isVipChannelEnabled(), brokerAddr), request, leftTime); + + assert response != null; + if (response.getCode() == SUCCESS) { + SubscriptionGroupWrapper subscriptionGroupWrapper = + SubscriptionGroupWrapper.decode(response.getBody(), SubscriptionGroupWrapper.class); + subscriptionGroupTable.putAll(subscriptionGroupWrapper.getSubscriptionGroupTable()); + forbiddenTable.putAll(subscriptionGroupWrapper.getForbiddenTable()); + + DataVersion newDataVersion = subscriptionGroupWrapper.getDataVersion(); + if (currentDataVersion == null) { + // fill dataVersion before break the loop to compatible with old version server + currentDataVersion = newDataVersion; + } + + groupSeq += subscriptionGroupWrapper.getSubscriptionGroupTable().size(); + + GetAllSubscriptionGroupResponseHeader responseHeader = + response.decodeCommandCustomHeader(GetAllSubscriptionGroupResponseHeader.class); + Integer totalGroupNum = Optional.ofNullable(responseHeader) + .map(GetAllSubscriptionGroupResponseHeader::getTotalGroupNum).orElse(null); + + if (Objects.isNull(totalGroupNum)) { + // the server side don't support totalGroupNum, all data is returned + break; + } + + if (!Objects.equals(currentDataVersion, newDataVersion)) { + log.error("dataVersion changed, currentDataVersion: {}, newDataVersion: {}", + currentDataVersion, newDataVersion); + currentDataVersion = newDataVersion; + groupSeq = 0; + subscriptionGroupTable.clear(); + forbiddenTable.clear(); + continue; + } + + if (groupSeq >= totalGroupNum - 1) { + log.info("get all subscription group config, totalGroupNum: {}", totalGroupNum); + break; + } + } else { + throw new MQBrokerException(response.getCode(), response.getRemark(), brokerAddr); } - default: - break; } - throw new MQBrokerException(response.getCode(), response.getRemark(), brokerAddr); + + SubscriptionGroupWrapper allSubscriptionGroup = new SubscriptionGroupWrapper(); + allSubscriptionGroup.setSubscriptionGroupTable(subscriptionGroupTable); + allSubscriptionGroup.setForbiddenTable(forbiddenTable); + allSubscriptionGroup.setDataVersion(currentDataVersion); + return allSubscriptionGroup; } public SubscriptionGroupConfig getSubscriptionGroupConfig(final String brokerAddr, String group, @@ -2842,23 +2912,77 @@ public class MQClientAPIImpl implements NameServerUpdateCallback, StartAndShutdo throw new MQBrokerException(response.getCode(), response.getRemark(), brokerAddr); } - public TopicConfigSerializeWrapper getAllTopicConfig(final String addr, - long timeoutMillis) throws RemotingConnectException, - RemotingSendRequestException, RemotingTimeoutException, InterruptedException, MQBrokerException { - RemotingCommand request = RemotingCommand.createRequestCommand(RequestCode.GET_ALL_TOPIC_CONFIG, null); + public TopicConfigSerializeWrapper getAllTopicConfig(final String addr, long timeoutMillis) + throws RemotingConnectException, RemotingSendRequestException, RemotingTimeoutException, + InterruptedException, MQBrokerException, RemotingCommandException { - RemotingCommand response = this.remotingClient.invokeSync(MixAll.brokerVIPChannel(this.clientConfig.isVipChannelEnabled(), addr), - request, timeoutMillis); - assert response != null; - switch (response.getCode()) { - case ResponseCode.SUCCESS: { - return TopicConfigSerializeWrapper.decode(response.getBody(), TopicConfigSerializeWrapper.class); + DataVersion currentDataVersion = null; + int topicSeq = 0; + ConcurrentMap topicConfigTable = new ConcurrentHashMap<>(); + long beginTime = System.nanoTime(); + while (true) { + long leftTime = timeoutMillis - TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - beginTime); + if (leftTime <= 0) { + throw new RemotingTimeoutException("invokeSync call timeout"); } - default: - break; + + GetAllTopicConfigRequestHeader requestHeader = new GetAllTopicConfigRequestHeader(); + requestHeader.setTopicSeq(topicSeq); + requestHeader.setMaxTopicNum(clientConfig.getMaxPageSizeInGetMetadata()); + requestHeader.setDataVersion(Optional.ofNullable(currentDataVersion). + map(DataVersion::toJson).orElse(StringUtils.EMPTY)); + log.info("getAllTopicConfig from seq {}, max {}, dataVersion {}", + topicSeq, requestHeader.getMaxTopicNum(), requestHeader.getDataVersion()); + RemotingCommand request = RemotingCommand.createRequestCommand(RequestCode.GET_ALL_TOPIC_CONFIG, requestHeader); + + RemotingCommand response = this.remotingClient.invokeSync( + MixAll.brokerVIPChannel(this.clientConfig.isVipChannelEnabled(), addr), request, leftTime); + + assert response != null; + if (response.getCode() == SUCCESS) { + TopicConfigSerializeWrapper topicConfigSerializeWrapper = + TopicConfigSerializeWrapper.decode(response.getBody(), TopicConfigSerializeWrapper.class); + topicConfigTable.putAll(topicConfigSerializeWrapper.getTopicConfigTable()); + topicSeq += topicConfigSerializeWrapper.getTopicConfigTable().size(); + + DataVersion newDataVersion = topicConfigSerializeWrapper.getDataVersion(); + if (currentDataVersion == null) { + // fill dataVersion before break the loop to compatible with old version server + currentDataVersion = newDataVersion; + } + + GetAllTopicConfigResponseHeader responseHeader = + response.decodeCommandCustomHeader(GetAllTopicConfigResponseHeader.class); + Integer totalTopicNum = Optional.ofNullable(responseHeader) + .map(GetAllTopicConfigResponseHeader::getTotalTopicNum).orElse(null); + + if (Objects.isNull(totalTopicNum)) { // compatible with old version server + // the server side don't support totalTopicNum, all data is returned + break; + } + + if (!Objects.equals(currentDataVersion, newDataVersion)) { + log.error("dataVersion changed, currentDataVersion: {}, newDataVersion: {}", currentDataVersion, newDataVersion); + currentDataVersion = newDataVersion; + topicSeq = 0; + topicConfigTable.clear(); + continue; + } + + if (topicSeq >= totalTopicNum - 1) { + log.info("get all topic config, totalTopicNum: {}", totalTopicNum); + break; + } + } else { + throw new MQBrokerException(response.getCode(), response.getRemark(), addr); + } + } - throw new MQBrokerException(response.getCode(), response.getRemark(), addr); + TopicConfigSerializeWrapper topicConfigSerializeWrapper = new TopicConfigSerializeWrapper(); + topicConfigSerializeWrapper.setDataVersion(currentDataVersion); + topicConfigSerializeWrapper.setTopicConfigTable(topicConfigTable); + return topicConfigSerializeWrapper; } public void updateNameServerConfig(final Properties properties, final List nameServers, long timeoutMillis) 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 3d0feec8a7..77f49554a5 100644 --- a/common/src/main/java/org/apache/rocketmq/common/BrokerConfig.java +++ b/common/src/main/java/org/apache/rocketmq/common/BrokerConfig.java @@ -425,6 +425,9 @@ public class BrokerConfig extends BrokerIdentity { */ private boolean enableSplitRegistration = false; + private boolean enableSplitMetadata = true; + private int splitMetadataSize = 2000; + private long popInflightMessageThreshold = 10000; private boolean enablePopMessageThreshold = false; @@ -2060,4 +2063,20 @@ public class BrokerConfig extends BrokerIdentity { public void setEnableCreateSysGroup(boolean enableCreateSysGroup) { this.enableCreateSysGroup = enableCreateSysGroup; } + + public boolean isEnableSplitMetadata() { + return enableSplitMetadata; + } + + public void setEnableSplitMetadata(boolean enableSplitMetadata) { + this.enableSplitMetadata = enableSplitMetadata; + } + + public int getSplitMetadataSize() { + return splitMetadataSize; + } + + public void setSplitMetadataSize(int splitMetadataSize) { + this.splitMetadataSize = splitMetadataSize; + } } diff --git a/remoting/src/main/java/org/apache/rocketmq/remoting/protocol/body/SubscriptionGroupWrapper.java b/remoting/src/main/java/org/apache/rocketmq/remoting/protocol/body/SubscriptionGroupWrapper.java index 7c159021aa..a1828d3d53 100644 --- a/remoting/src/main/java/org/apache/rocketmq/remoting/protocol/body/SubscriptionGroupWrapper.java +++ b/remoting/src/main/java/org/apache/rocketmq/remoting/protocol/body/SubscriptionGroupWrapper.java @@ -26,6 +26,8 @@ import org.apache.rocketmq.remoting.protocol.subscription.SubscriptionGroupConfi public class SubscriptionGroupWrapper extends RemotingSerializable { private ConcurrentMap subscriptionGroupTable = new ConcurrentHashMap<>(1024); + private ConcurrentMap> forbiddenTable = + new ConcurrentHashMap<>(1024); private DataVersion dataVersion = new DataVersion(); public ConcurrentMap getSubscriptionGroupTable() { @@ -37,6 +39,14 @@ public class SubscriptionGroupWrapper extends RemotingSerializable { this.subscriptionGroupTable = subscriptionGroupTable; } + public ConcurrentMap> getForbiddenTable() { + return forbiddenTable; + } + + public void setForbiddenTable(ConcurrentMap> forbiddenTable) { + this.forbiddenTable = forbiddenTable; + } + public DataVersion getDataVersion() { return dataVersion; } diff --git a/remoting/src/main/java/org/apache/rocketmq/remoting/protocol/header/GetAllSubscriptionGroupRequestHeader.java b/remoting/src/main/java/org/apache/rocketmq/remoting/protocol/header/GetAllSubscriptionGroupRequestHeader.java new file mode 100644 index 0000000000..6d67afdb9c --- /dev/null +++ b/remoting/src/main/java/org/apache/rocketmq/remoting/protocol/header/GetAllSubscriptionGroupRequestHeader.java @@ -0,0 +1,65 @@ +/* + * 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.remoting.protocol.header; + +import org.apache.rocketmq.common.action.Action; +import org.apache.rocketmq.common.action.RocketMQAction; +import org.apache.rocketmq.common.resource.ResourceType; +import org.apache.rocketmq.remoting.CommandCustomHeader; +import org.apache.rocketmq.remoting.annotation.CFNotNull; +import org.apache.rocketmq.remoting.exception.RemotingCommandException; +import org.apache.rocketmq.remoting.protocol.RequestCode; + +@RocketMQAction(value = RequestCode.GET_ALL_SUBSCRIPTIONGROUP_CONFIG, resource = ResourceType.GROUP, action = Action.GET) +public class GetAllSubscriptionGroupRequestHeader implements CommandCustomHeader { + @Override + public void checkFields() throws RemotingCommandException { + // nothing + } + + @CFNotNull + private Integer groupSeq; + + private String dataVersion; + + private Integer maxGroupNum; + + public Integer getGroupSeq() { + return groupSeq; + } + + public void setGroupSeq(Integer groupSeq) { + this.groupSeq = groupSeq; + } + + public String getDataVersion() { + return dataVersion; + } + + public void setDataVersion(String dataVersion) { + this.dataVersion = dataVersion; + } + + public Integer getMaxGroupNum() { + return maxGroupNum; + } + + public void setMaxGroupNum(Integer maxGroupNum) { + this.maxGroupNum = maxGroupNum; + } +} diff --git a/remoting/src/main/java/org/apache/rocketmq/remoting/protocol/header/GetAllSubscriptionGroupResponseHeader.java b/remoting/src/main/java/org/apache/rocketmq/remoting/protocol/header/GetAllSubscriptionGroupResponseHeader.java new file mode 100644 index 0000000000..8f42a1b2a8 --- /dev/null +++ b/remoting/src/main/java/org/apache/rocketmq/remoting/protocol/header/GetAllSubscriptionGroupResponseHeader.java @@ -0,0 +1,45 @@ +/* + * 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.remoting.protocol.header; + +import org.apache.rocketmq.common.action.Action; +import org.apache.rocketmq.common.action.RocketMQAction; +import org.apache.rocketmq.common.resource.ResourceType; +import org.apache.rocketmq.remoting.CommandCustomHeader; +import org.apache.rocketmq.remoting.annotation.CFNotNull; +import org.apache.rocketmq.remoting.exception.RemotingCommandException; +import org.apache.rocketmq.remoting.protocol.RequestCode; + +@RocketMQAction(value = RequestCode.GET_ALL_SUBSCRIPTIONGROUP_CONFIG, resource = ResourceType.GROUP, action = Action.LIST) +public class GetAllSubscriptionGroupResponseHeader implements CommandCustomHeader { + @Override + public void checkFields() throws RemotingCommandException { + + } + + @CFNotNull + private Integer totalGroupNum; + + public Integer getTotalGroupNum() { + return totalGroupNum; + } + + public void setTotalGroupNum(Integer totalGroupNum) { + this.totalGroupNum = totalGroupNum; + } +} diff --git a/remoting/src/main/java/org/apache/rocketmq/remoting/protocol/header/GetAllTopicConfigRequestHeader.java b/remoting/src/main/java/org/apache/rocketmq/remoting/protocol/header/GetAllTopicConfigRequestHeader.java new file mode 100644 index 0000000000..769a814d34 --- /dev/null +++ b/remoting/src/main/java/org/apache/rocketmq/remoting/protocol/header/GetAllTopicConfigRequestHeader.java @@ -0,0 +1,65 @@ +/* + * 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.remoting.protocol.header; + +import org.apache.rocketmq.common.action.Action; +import org.apache.rocketmq.common.action.RocketMQAction; +import org.apache.rocketmq.common.resource.ResourceType; +import org.apache.rocketmq.remoting.CommandCustomHeader; +import org.apache.rocketmq.remoting.annotation.CFNotNull; +import org.apache.rocketmq.remoting.exception.RemotingCommandException; +import org.apache.rocketmq.remoting.protocol.RequestCode; + +@RocketMQAction(value = RequestCode.GET_ALL_TOPIC_CONFIG, resource = ResourceType.TOPIC, action = Action.GET) +public class GetAllTopicConfigRequestHeader implements CommandCustomHeader { + @Override + public void checkFields() throws RemotingCommandException { + // nothing + } + + @CFNotNull + private Integer topicSeq; + + private String dataVersion; + + private Integer maxTopicNum; + + public Integer getTopicSeq() { + return topicSeq; + } + + public void setTopicSeq(Integer topicSeq) { + this.topicSeq = topicSeq; + } + + public String getDataVersion() { + return dataVersion; + } + + public void setDataVersion(String dataVersion) { + this.dataVersion = dataVersion; + } + + public Integer getMaxTopicNum() { + return maxTopicNum; + } + + public void setMaxTopicNum(Integer maxTopicNum) { + this.maxTopicNum = maxTopicNum; + } +} diff --git a/remoting/src/main/java/org/apache/rocketmq/remoting/protocol/header/GetAllTopicConfigResponseHeader.java b/remoting/src/main/java/org/apache/rocketmq/remoting/protocol/header/GetAllTopicConfigResponseHeader.java index 566ce16fa4..446ed03c47 100644 --- a/remoting/src/main/java/org/apache/rocketmq/remoting/protocol/header/GetAllTopicConfigResponseHeader.java +++ b/remoting/src/main/java/org/apache/rocketmq/remoting/protocol/header/GetAllTopicConfigResponseHeader.java @@ -24,6 +24,7 @@ import org.apache.rocketmq.common.action.Action; import org.apache.rocketmq.common.action.RocketMQAction; import org.apache.rocketmq.common.resource.ResourceType; import org.apache.rocketmq.remoting.CommandCustomHeader; +import org.apache.rocketmq.remoting.annotation.CFNotNull; import org.apache.rocketmq.remoting.exception.RemotingCommandException; import org.apache.rocketmq.remoting.protocol.RequestCode; @@ -33,4 +34,15 @@ public class GetAllTopicConfigResponseHeader implements CommandCustomHeader { @Override public void checkFields() throws RemotingCommandException { } + + @CFNotNull + private Integer totalTopicNum; + + public Integer getTotalTopicNum() { + return totalTopicNum; + } + + public void setTotalTopicNum(Integer totalTopicNum) { + this.totalTopicNum = totalTopicNum; + } } diff --git a/tools/src/main/java/org/apache/rocketmq/tools/admin/DefaultMQAdminExt.java b/tools/src/main/java/org/apache/rocketmq/tools/admin/DefaultMQAdminExt.java index 9780df13dd..2e72af13ee 100644 --- a/tools/src/main/java/org/apache/rocketmq/tools/admin/DefaultMQAdminExt.java +++ b/tools/src/main/java/org/apache/rocketmq/tools/admin/DefaultMQAdminExt.java @@ -652,23 +652,23 @@ public class DefaultMQAdminExt extends ClientConfig implements MQAdminExt { } @Override - public SubscriptionGroupWrapper getAllSubscriptionGroup(final String brokerAddr, - long timeoutMillis) throws InterruptedException, RemotingTimeoutException, RemotingSendRequestException, - RemotingConnectException, MQBrokerException { + public SubscriptionGroupWrapper getAllSubscriptionGroup(final String brokerAddr, long timeoutMillis) + throws InterruptedException, RemotingTimeoutException, RemotingSendRequestException, + RemotingConnectException, MQBrokerException, RemotingCommandException { return this.defaultMQAdminExtImpl.getAllSubscriptionGroup(brokerAddr, timeoutMillis); } @Override - public SubscriptionGroupWrapper getUserSubscriptionGroup(final String brokerAddr, - long timeoutMillis) throws InterruptedException, RemotingTimeoutException, RemotingSendRequestException, - RemotingConnectException, MQBrokerException { + public SubscriptionGroupWrapper getUserSubscriptionGroup(final String brokerAddr, long timeoutMillis) + throws InterruptedException, RemotingTimeoutException, RemotingSendRequestException, + RemotingConnectException, MQBrokerException, RemotingCommandException { return this.defaultMQAdminExtImpl.getUserSubscriptionGroup(brokerAddr, timeoutMillis); } @Override public TopicConfigSerializeWrapper getAllTopicConfig(final String brokerAddr, long timeoutMillis) throws InterruptedException, RemotingTimeoutException, RemotingSendRequestException, - RemotingConnectException, MQBrokerException { + RemotingConnectException, MQBrokerException, RemotingCommandException { return this.defaultMQAdminExtImpl.getAllTopicConfig(brokerAddr, timeoutMillis); } diff --git a/tools/src/main/java/org/apache/rocketmq/tools/admin/DefaultMQAdminExtImpl.java b/tools/src/main/java/org/apache/rocketmq/tools/admin/DefaultMQAdminExtImpl.java index 1bdcc765d6..7b268cf694 100644 --- a/tools/src/main/java/org/apache/rocketmq/tools/admin/DefaultMQAdminExtImpl.java +++ b/tools/src/main/java/org/apache/rocketmq/tools/admin/DefaultMQAdminExtImpl.java @@ -1636,14 +1636,16 @@ public class DefaultMQAdminExtImpl implements MQAdminExt, MQAdminExtInner { } @Override - public SubscriptionGroupWrapper getAllSubscriptionGroup(final String brokerAddr, - long timeoutMillis) throws InterruptedException, RemotingTimeoutException, RemotingSendRequestException, RemotingConnectException, MQBrokerException { + public SubscriptionGroupWrapper getAllSubscriptionGroup(final String brokerAddr, long timeoutMillis) + throws InterruptedException, RemotingTimeoutException, RemotingSendRequestException, + RemotingConnectException, MQBrokerException, RemotingCommandException { return this.mqClientInstance.getMQClientAPIImpl().getAllSubscriptionGroup(brokerAddr, timeoutMillis); } @Override - public SubscriptionGroupWrapper getUserSubscriptionGroup(final String brokerAddr, - long timeoutMillis) throws InterruptedException, RemotingTimeoutException, RemotingSendRequestException, RemotingConnectException, MQBrokerException { + public SubscriptionGroupWrapper getUserSubscriptionGroup(final String brokerAddr, long timeoutMillis) + throws InterruptedException, RemotingTimeoutException, RemotingSendRequestException, + RemotingConnectException, MQBrokerException, RemotingCommandException { SubscriptionGroupWrapper subscriptionGroupWrapper = this.mqClientInstance.getMQClientAPIImpl().getAllSubscriptionGroup(brokerAddr, timeoutMillis); Iterator> iterator = subscriptionGroupWrapper.getSubscriptionGroupTable().entrySet().iterator(); @@ -1658,8 +1660,9 @@ public class DefaultMQAdminExtImpl implements MQAdminExt, MQAdminExtInner { } @Override - public TopicConfigSerializeWrapper getAllTopicConfig(final String brokerAddr, - long timeoutMillis) throws InterruptedException, RemotingTimeoutException, RemotingSendRequestException, RemotingConnectException, MQBrokerException { + public TopicConfigSerializeWrapper getAllTopicConfig(final String brokerAddr, long timeoutMillis) + throws InterruptedException, RemotingTimeoutException, RemotingSendRequestException, + RemotingConnectException, MQBrokerException, RemotingCommandException { return this.mqClientInstance.getMQClientAPIImpl().getAllTopicConfig(brokerAddr, timeoutMillis); } diff --git a/tools/src/main/java/org/apache/rocketmq/tools/admin/MQAdminExt.java b/tools/src/main/java/org/apache/rocketmq/tools/admin/MQAdminExt.java index a10a58950d..46e2c066cb 100644 --- a/tools/src/main/java/org/apache/rocketmq/tools/admin/MQAdminExt.java +++ b/tools/src/main/java/org/apache/rocketmq/tools/admin/MQAdminExt.java @@ -316,16 +316,18 @@ public interface MQAdminExt extends MQAdmin { final String topic) throws InterruptedException, MQBrokerException, MQClientException, RemotingException; SubscriptionGroupWrapper getAllSubscriptionGroup(final String brokerAddr, - long timeoutMillis) throws InterruptedException, RemotingTimeoutException, RemotingSendRequestException, - RemotingConnectException, MQBrokerException; + long timeoutMillis) + throws InterruptedException, RemotingTimeoutException, RemotingSendRequestException, RemotingConnectException, + MQBrokerException, RemotingCommandException; SubscriptionGroupWrapper getUserSubscriptionGroup(final String brokerAddr, - long timeoutMillis) throws InterruptedException, RemotingTimeoutException, RemotingSendRequestException, - RemotingConnectException, MQBrokerException; + long timeoutMillis) + throws InterruptedException, RemotingTimeoutException, RemotingSendRequestException, RemotingConnectException, + MQBrokerException, RemotingCommandException; TopicConfigSerializeWrapper getAllTopicConfig(final String brokerAddr, long timeoutMillis) throws InterruptedException, RemotingTimeoutException, RemotingSendRequestException, - RemotingConnectException, MQBrokerException; + RemotingConnectException, MQBrokerException, RemotingCommandException; TopicConfigSerializeWrapper getUserTopicConfig(final String brokerAddr, final boolean specialTopic, long timeoutMillis) throws InterruptedException, RemotingException, diff --git a/tools/src/test/java/org/apache/rocketmq/tools/admin/DefaultMQAdminExtTest.java b/tools/src/test/java/org/apache/rocketmq/tools/admin/DefaultMQAdminExtTest.java index ec5f7571d2..884764f853 100644 --- a/tools/src/test/java/org/apache/rocketmq/tools/admin/DefaultMQAdminExtTest.java +++ b/tools/src/test/java/org/apache/rocketmq/tools/admin/DefaultMQAdminExtTest.java @@ -491,7 +491,8 @@ public class DefaultMQAdminExtTest { } @Test - public void testGetAllSubscriptionGroup() throws InterruptedException, MQBrokerException, RemotingTimeoutException, RemotingSendRequestException, RemotingConnectException { + public void testGetAllSubscriptionGroup() throws InterruptedException, MQBrokerException, RemotingTimeoutException, + RemotingSendRequestException, RemotingConnectException, RemotingCommandException { SubscriptionGroupWrapper subscriptionGroupWrapper = defaultMQAdminExt.getAllSubscriptionGroup("127.0.0.1:10911", 10000); assertThat(subscriptionGroupWrapper.getSubscriptionGroupTable().get("Consumer-group-one").getBrokerId()).isEqualTo(1234); assertThat(subscriptionGroupWrapper.getSubscriptionGroupTable().get("Consumer-group-one").getGroupName()).isEqualTo("Consumer-group-one");