This is an automated email from the ASF dual-hosted git repository.
fuyou001 pushed a commit to branch develop
in repository https://gitbox.apache.org/repos/asf/rocketmq.git
The following commit(s) were added to refs/heads/develop by this push:
new 74e90f7c28 [ISSUE #10446] Support batch deletion of topics and
subscription groups in broker (#10448)
74e90f7c28 is described below
commit 74e90f7c283d4030e4c89a35f46dec77419bcb9e
Author: Jiahua Wang <[email protected]>
AuthorDate: Mon Jul 13 10:07:43 2026 +0800
[ISSUE #10446] Support batch deletion of topics and subscription groups in
broker (#10448)
* [ISSUE #10446] Support batch deletion of topics and subscription groups
in broker
- Add DELETE_TOPIC_IN_BROKER_LIST and DELETE_SUBSCRIPTION_GROUP_LIST
request codes with corresponding RequestBody types
- Implement deleteTopicList / deleteSubscriptionGroupList in
AdminBrokerProcessor with deduplication and one-shot persist
- Add deleteTopicConfigList / deleteSubscriptionGroupConfigList in
TopicConfigManager / SubscriptionGroupManager (single persist per batch)
- Add MQClientAPIImpl#deleteTopicInBrokerList /
deleteSubscriptionGroupList client APIs
- Cover changes with unit tests in AdminBrokerProcessorTest
* fix: persist config after batch delete
* fix: preserve batch delete failure response
---------
Co-authored-by: wangjiahua.wjh <[email protected]>
Co-authored-by: fuchong <[email protected]>
---
.../DefaultAuthorizationContextBuilder.java | 35 ++++
.../DefaultAuthorizationContextBuilderTest.java | 45 +++++
.../broker/processor/AdminBrokerProcessor.java | 202 ++++++++++++++++++---
.../subscription/SubscriptionGroupManager.java | 18 +-
.../rocketmq/broker/topic/TopicConfigManager.java | 19 +-
.../broker/processor/AdminBrokerProcessorTest.java | 191 ++++++++++++++++++-
.../subscription/SubscriptionGroupManagerTest.java | 19 +-
.../broker/topic/TopicConfigManagerTest.java | 18 ++
.../rocketmq/client/impl/MQClientAPIImpl.java | 46 +++++
.../rocketmq/client/impl/MQClientAPIImplTest.java | 62 +++++++
.../org/apache/rocketmq/common/BrokerConfig.java | 28 +++
.../rocketmq/remoting/protocol/RequestCode.java | 3 +
.../DeleteSubscriptionGroupListRequestBody.java | 56 ++++++
.../protocol/body/DeleteTopicListRequestBody.java | 41 +++++
14 files changed, 755 insertions(+), 28 deletions(-)
diff --git
a/auth/src/main/java/org/apache/rocketmq/auth/authorization/builder/DefaultAuthorizationContextBuilder.java
b/auth/src/main/java/org/apache/rocketmq/auth/authorization/builder/DefaultAuthorizationContextBuilder.java
index f462aabc0d..7f4fca5117 100644
---
a/auth/src/main/java/org/apache/rocketmq/auth/authorization/builder/DefaultAuthorizationContextBuilder.java
+++
b/auth/src/main/java/org/apache/rocketmq/auth/authorization/builder/DefaultAuthorizationContextBuilder.java
@@ -68,6 +68,8 @@ import org.apache.rocketmq.remoting.protocol.NamespaceUtil;
import org.apache.rocketmq.remoting.protocol.RemotingCommand;
import org.apache.rocketmq.remoting.protocol.RequestCode;
import org.apache.rocketmq.remoting.protocol.RequestHeaderRegistry;
+import
org.apache.rocketmq.remoting.protocol.body.DeleteSubscriptionGroupListRequestBody;
+import org.apache.rocketmq.remoting.protocol.body.DeleteTopicListRequestBody;
import org.apache.rocketmq.remoting.protocol.body.LockBatchRequestBody;
import org.apache.rocketmq.remoting.protocol.body.UnlockBatchRequestBody;
import
org.apache.rocketmq.remoting.protocol.header.GetConsumerListByGroupRequestHeader;
@@ -322,6 +324,39 @@ public class DefaultAuthorizationContextBuilder implements
AuthorizationContextB
}
}
break;
+ case RequestCode.DELETE_TOPIC_IN_BROKER_LIST:
+ // Batch APIs carry their target list in the request body,
not in an annotated
+ // CommandCustomHeader, so the annotation-based path in
+ // RequestHeaderRegistry would otherwise produce an empty
context list and let
+ // the request through without a DELETE permission check.
Decode the body and
+ // emit one DELETE context per topic instead.
+ DeleteTopicListRequestBody deleteTopicListRequestBody =
+ DeleteTopicListRequestBody.decode(command.getBody(),
DeleteTopicListRequestBody.class);
+ if
(CollectionUtils.isNotEmpty(deleteTopicListRequestBody.getTopicList())) {
+ for (String topicName :
deleteTopicListRequestBody.getTopicList()) {
+ if (StringUtils.isBlank(topicName)) {
+ continue;
+ }
+ topic = Resource.ofTopic(topicName);
+ result.add(DefaultAuthorizationContext.of(subject,
topic, Action.DELETE, sourceIp));
+ }
+ }
+ break;
+ case RequestCode.DELETE_SUBSCRIPTION_GROUP_LIST:
+ // See DELETE_TOPIC_IN_BROKER_LIST: emit one DELETE
context per group from the
+ // request body so authorization can enforce per-group
DELETE permission.
+ DeleteSubscriptionGroupListRequestBody
deleteGroupListRequestBody =
+
DeleteSubscriptionGroupListRequestBody.decode(command.getBody(),
DeleteSubscriptionGroupListRequestBody.class);
+ if
(CollectionUtils.isNotEmpty(deleteGroupListRequestBody.getGroupNameList())) {
+ for (String groupName :
deleteGroupListRequestBody.getGroupNameList()) {
+ if (StringUtils.isBlank(groupName)) {
+ continue;
+ }
+ group = Resource.ofGroup(groupName);
+ result.add(DefaultAuthorizationContext.of(subject,
group, Action.DELETE, sourceIp));
+ }
+ }
+ break;
default:
result = buildContextByAnnotation(subject, command,
sourceIp);
break;
diff --git
a/auth/src/test/java/org/apache/rocketmq/auth/authorization/builder/DefaultAuthorizationContextBuilderTest.java
b/auth/src/test/java/org/apache/rocketmq/auth/authorization/builder/DefaultAuthorizationContextBuilderTest.java
index 920a0e1200..a9de347324 100644
---
a/auth/src/test/java/org/apache/rocketmq/auth/authorization/builder/DefaultAuthorizationContextBuilderTest.java
+++
b/auth/src/test/java/org/apache/rocketmq/auth/authorization/builder/DefaultAuthorizationContextBuilderTest.java
@@ -57,6 +57,8 @@ import org.apache.rocketmq.remoting.netty.AttributeKeys;
import org.apache.rocketmq.remoting.protocol.RemotingCommand;
import org.apache.rocketmq.remoting.protocol.RequestCode;
import org.apache.rocketmq.remoting.protocol.RequestHeaderRegistry;
+import
org.apache.rocketmq.remoting.protocol.body.DeleteSubscriptionGroupListRequestBody;
+import org.apache.rocketmq.remoting.protocol.body.DeleteTopicListRequestBody;
import org.apache.rocketmq.remoting.protocol.body.LockBatchRequestBody;
import org.apache.rocketmq.remoting.protocol.body.UnlockBatchRequestBody;
import
org.apache.rocketmq.remoting.protocol.header.ConsumerSendMsgBackRequestHeader;
@@ -570,6 +572,49 @@ public class DefaultAuthorizationContextBuilderTest {
Assert.assertEquals("channel-id", getContext(result,
ResourceType.TOPIC).getChannelId());
Assert.assertEquals(String.valueOf(RequestCode.UNLOCK_BATCH_MQ),
getContext(result, ResourceType.TOPIC).getRpcCode());
+ // DELETE_TOPIC_IN_BROKER_LIST: body-driven, must yield one DELETE
context per topic.
+ DeleteTopicListRequestBody deleteTopicListBody = new
DeleteTopicListRequestBody();
+ deleteTopicListBody.setTopicList(Arrays.asList("topicA", "topicB", "",
" "));
+
+ request =
RemotingCommand.createRequestCommand(RequestCode.DELETE_TOPIC_IN_BROKER_LIST,
null);
+ request.setBody(JSON.toJSONBytes(deleteTopicListBody));
+ request.setVersion(441);
+ request.addExtField("AccessKey", "rocketmq");
+ request.makeCustomHeaderToNet();
+
+ result = builder.build(channelHandlerContext, request);
+ // Blank entries are filtered, so 2 valid topics produce 2 contexts.
+ Assert.assertEquals(2, result.size());
+ for (DefaultAuthorizationContext ctx : result) {
+ Assert.assertEquals(ResourceType.TOPIC,
ctx.getResource().getResourceType());
+ Assert.assertEquals("User:rocketmq",
ctx.getSubject().getSubjectKey());
+ Assert.assertTrue(ctx.getActions().contains(Action.DELETE));
+
Assert.assertEquals(String.valueOf(RequestCode.DELETE_TOPIC_IN_BROKER_LIST),
ctx.getRpcCode());
+ }
+ Assert.assertTrue(result.stream().anyMatch(ctx ->
"Topic:topicA".equals(ctx.getResource().getResourceKey())));
+ Assert.assertTrue(result.stream().anyMatch(ctx ->
"Topic:topicB".equals(ctx.getResource().getResourceKey())));
+
+ // DELETE_SUBSCRIPTION_GROUP_LIST: body-driven, must yield one DELETE
context per group.
+ DeleteSubscriptionGroupListRequestBody deleteGroupListBody = new
DeleteSubscriptionGroupListRequestBody();
+ deleteGroupListBody.setGroupNameList(Arrays.asList("groupX",
"groupY"));
+
+ request =
RemotingCommand.createRequestCommand(RequestCode.DELETE_SUBSCRIPTION_GROUP_LIST,
null);
+ request.setBody(JSON.toJSONBytes(deleteGroupListBody));
+ request.setVersion(441);
+ request.addExtField("AccessKey", "rocketmq");
+ request.makeCustomHeaderToNet();
+
+ result = builder.build(channelHandlerContext, request);
+ Assert.assertEquals(2, result.size());
+ for (DefaultAuthorizationContext ctx : result) {
+ Assert.assertEquals(ResourceType.GROUP,
ctx.getResource().getResourceType());
+ Assert.assertEquals("User:rocketmq",
ctx.getSubject().getSubjectKey());
+ Assert.assertTrue(ctx.getActions().contains(Action.DELETE));
+
Assert.assertEquals(String.valueOf(RequestCode.DELETE_SUBSCRIPTION_GROUP_LIST),
ctx.getRpcCode());
+ }
+ Assert.assertTrue(result.stream().anyMatch(ctx ->
"Group:groupX".equals(ctx.getResource().getResourceKey())));
+ Assert.assertTrue(result.stream().anyMatch(ctx ->
"Group:groupY".equals(ctx.getResource().getResourceKey())));
+
}
private DefaultAuthorizationContext
getContext(List<DefaultAuthorizationContext> contexts,
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 fc0182d733..876a72fa29 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
@@ -28,9 +28,11 @@ import java.net.UnknownHostException;
import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.Arrays;
+import java.util.Collection;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
+import java.util.LinkedHashSet;
import java.util.List;
import java.util.Map;
import java.util.Optional;
@@ -128,6 +130,8 @@ import
org.apache.rocketmq.remoting.protocol.body.ConsumeQueueData;
import org.apache.rocketmq.remoting.protocol.body.ConsumeStatsList;
import org.apache.rocketmq.remoting.protocol.body.ConsumerConnection;
import org.apache.rocketmq.remoting.protocol.body.CreateTopicListRequestBody;
+import
org.apache.rocketmq.remoting.protocol.body.DeleteSubscriptionGroupListRequestBody;
+import org.apache.rocketmq.remoting.protocol.body.DeleteTopicListRequestBody;
import org.apache.rocketmq.remoting.protocol.body.EpochEntryCache;
import org.apache.rocketmq.remoting.protocol.body.GroupList;
import org.apache.rocketmq.remoting.protocol.body.HARuntimeInfo;
@@ -270,6 +274,8 @@ public class AdminBrokerProcessor implements
NettyRequestProcessor {
return this.updateAndCreateTopicList(ctx, request);
case RequestCode.DELETE_TOPIC_IN_BROKER:
return this.deleteTopic(ctx, request);
+ case RequestCode.DELETE_TOPIC_IN_BROKER_LIST:
+ return this.deleteTopicList(ctx, request);
case RequestCode.GET_ALL_TOPIC_CONFIG:
return this.getAllTopicConfig(ctx, request);
case RequestCode.GET_TIMER_CHECK_POINT:
@@ -310,6 +316,8 @@ public class AdminBrokerProcessor implements
NettyRequestProcessor {
return this.getAllSubscriptionGroup(ctx, request);
case RequestCode.DELETE_SUBSCRIPTIONGROUP:
return this.deleteSubscriptionGroup(ctx, request);
+ case RequestCode.DELETE_SUBSCRIPTION_GROUP_LIST:
+ return this.deleteSubscriptionGroupList(ctx, request);
case RequestCode.GET_TOPIC_STATS_INFO:
return this.getTopicStatsInfo(ctx, request);
case RequestCode.GET_CONSUMER_CONNECTION_LIST:
@@ -782,17 +790,7 @@ public class AdminBrokerProcessor implements
NettyRequestProcessor {
topicsToClean.add(topic);
if
(brokerController.getBrokerConfig().isClearRetryTopicWhenDeleteTopic()) {
- final Set<String> groups =
this.brokerController.getConsumerOffsetManager().whichGroupByTopic(topic);
- for (String group : groups) {
- final String popRetryTopicV2 =
KeyBuilder.buildPopRetryTopic(topic, group, true);
- if
(brokerController.getTopicConfigManager().selectTopicConfig(popRetryTopicV2) !=
null) {
- topicsToClean.add(popRetryTopicV2);
- }
- final String popRetryTopicV1 =
KeyBuilder.buildPopRetryTopicV1(topic, group);
- if
(brokerController.getTopicConfigManager().selectTopicConfig(popRetryTopicV1) !=
null) {
- topicsToClean.add(popRetryTopicV1);
- }
- }
+ collectPopRetryTopics(topic, topicsToClean);
}
try {
@@ -811,8 +809,96 @@ public class AdminBrokerProcessor implements
NettyRequestProcessor {
return response;
}
- private void deleteTopicInBroker(String topic) {
- this.brokerController.getTopicConfigManager().deleteTopicConfig(topic);
+ private synchronized RemotingCommand deleteTopicList(ChannelHandlerContext
ctx,
+ RemotingCommand request) {
+ final RemotingCommand response =
RemotingCommand.createResponseCommand(null);
+
+ DeleteTopicListRequestBody requestBody =
DeleteTopicListRequestBody.decode(
+ request.getBody(), DeleteTopicListRequestBody.class);
+ List<String> topicList = requestBody == null ? null :
requestBody.getTopicList();
+
+ if (CollectionUtils.isEmpty(topicList)) {
+ response.setCode(ResponseCode.INVALID_PARAMETER);
+ response.setRemark("The specified topic list is blank.");
+ return response;
+ }
+
+ LOGGER.info("AdminBrokerProcessor#deleteTopicList: broker receive
request to delete topics={}, caller={}",
+ topicList, RemotingHelper.parseChannelRemoteAddr(ctx.channel()));
+
+ boolean validateSystemTopic =
brokerController.getBrokerConfig().isValidateSystemTopicWhenUpdateTopic();
+ // dedup while preserving the input order
+ Set<String> topicsToClean = new LinkedHashSet<>();
+ for (String topic : topicList) {
+ if (UtilAll.isBlank(topic)) {
+ response.setCode(ResponseCode.INVALID_PARAMETER);
+ response.setRemark("The specified topic is blank.");
+ return response;
+ }
+ if (validateSystemTopic && TopicValidator.isSystemTopic(topic)) {
+ response.setCode(ResponseCode.INVALID_PARAMETER);
+ response.setRemark("The topic[" + topic + "] is conflict with
system topic.");
+ return response;
+ }
+ topicsToClean.add(topic);
+ }
+
+ if
(brokerController.getBrokerConfig().isClearRetryTopicWhenDeleteTopic()) {
+ // snapshot the inputs before mutating the set, so retry topics
for already-added retry topics are not collected
+ for (String topic : new ArrayList<>(topicsToClean)) {
+ collectPopRetryTopics(topic, topicsToClean);
+ }
+ }
+
+ boolean isSuccess = false;
+ try {
+ double maxRate =
this.brokerController.getBrokerConfig().getBatchDeleteTopicMaxRate();
+ com.google.common.util.concurrent.RateLimiter rateLimiter =
maxRate > 0
+ ?
com.google.common.util.concurrent.RateLimiter.create(maxRate) : null;
+
+ for (String topic : topicsToClean) {
+ if (rateLimiter != null) {
+ rateLimiter.acquire();
+ }
+ if (LiteMetadataUtil.isLiteMessageType(topic,
brokerController)) {
+
brokerController.getLiteLifecycleManager().cleanByParentTopic(topic);
+ }
+ deleteTopicInBroker(topic, false);
+ }
+ isSuccess = true;
+ } catch (Throwable t) {
+ LOGGER.error("Failed to delete topic ", t);
+ } finally {
+ try {
+ this.brokerController.getTopicConfigManager().persist();
+ } catch (Throwable t) {
+ isSuccess = false;
+ LOGGER.error("Failed to persist topic config after batch
delete", t);
+ }
+ }
+
+ response.setCode(isSuccess ? ResponseCode.SUCCESS :
ResponseCode.SYSTEM_ERROR);
+ response.setRemark(null);
+
+ return response;
+ }
+
+ private void collectPopRetryTopics(String topic, Collection<String>
topicsToClean) {
+ final Set<String> groups =
this.brokerController.getConsumerOffsetManager().whichGroupByTopic(topic);
+ for (String group : groups) {
+ final String popRetryTopicV2 =
KeyBuilder.buildPopRetryTopic(topic, group, true);
+ if
(brokerController.getTopicConfigManager().selectTopicConfig(popRetryTopicV2) !=
null) {
+ topicsToClean.add(popRetryTopicV2);
+ }
+ final String popRetryTopicV1 =
KeyBuilder.buildPopRetryTopicV1(topic, group);
+ if
(brokerController.getTopicConfigManager().selectTopicConfig(popRetryTopicV1) !=
null) {
+ topicsToClean.add(popRetryTopicV1);
+ }
+ }
+ }
+
+ private void deleteTopicInBroker(String topic, boolean persist) {
+ this.brokerController.getTopicConfigManager().deleteTopicConfig(topic,
persist);
this.brokerController.getTopicQueueMappingManager().delete(topic);
this.brokerController.getConsumerOffsetManager().cleanOffsetByTopic(topic);
this.brokerController.getPopInflightMessageCounter().clearInFlightMessageNumByTopicName(topic);
@@ -820,6 +906,10 @@ public class AdminBrokerProcessor implements
NettyRequestProcessor {
this.brokerController.getMessageStore().getTimerMessageStore().getTimerMetrics().removeTimingCount(topic);
}
+ private void deleteTopicInBroker(String topic) {
+ deleteTopicInBroker(topic, true);
+ }
+
private RemotingCommand getUnknownCmdResponse(ChannelHandlerContext ctx,
RemotingCommand request) {
String error = " request type " + request.getCode() + " not supported";
final RemotingCommand response =
@@ -1693,18 +1783,88 @@ public class AdminBrokerProcessor implements
NettyRequestProcessor {
LOGGER.info("AdminBrokerProcessor#deleteSubscriptionGroup, caller={}",
RemotingHelper.parseChannelRemoteAddr(ctx.channel()));
-
this.brokerController.getSubscriptionGroupManager().deleteSubscriptionGroupConfig(requestHeader.getGroupName());
+ boolean cleanOffset = requestHeader.isCleanOffset()
+ || LiteMetadataUtil.isLiteGroupType(requestHeader.getGroupName(),
this.brokerController);
+ deleteSubscriptionGroupInBroker(requestHeader.getGroupName(),
cleanOffset);
- if (requestHeader.isCleanOffset()
- || LiteMetadataUtil.isLiteGroupType(requestHeader.getGroupName(),
this.brokerController)) {
-
this.brokerController.getConsumerOffsetManager().removeOffset(requestHeader.getGroupName());
-
this.brokerController.getPopInflightMessageCounter().clearInFlightMessageNumByGroupName(requestHeader.getGroupName());
- }
+ response.setCode(ResponseCode.SUCCESS);
+ response.setRemark(null);
+ return response;
+ }
+ private void deleteSubscriptionGroupInBroker(String groupName, boolean
cleanOffset) {
+ deleteSubscriptionGroupInBroker(groupName, cleanOffset, true);
+ }
+
+ private void deleteSubscriptionGroupInBroker(String groupName, boolean
cleanOffset, boolean persist) {
+
this.brokerController.getSubscriptionGroupManager().deleteSubscriptionGroupConfig(groupName,
persist);
+ if (cleanOffset) {
+
this.brokerController.getConsumerOffsetManager().removeOffset(groupName);
+
this.brokerController.getPopInflightMessageCounter().clearInFlightMessageNumByGroupName(groupName);
+ }
if (this.brokerController.getBrokerConfig().isAutoDeleteUnusedStats())
{
-
this.brokerController.getBrokerStatsManager().onGroupDeleted(requestHeader.getGroupName());
+
this.brokerController.getBrokerStatsManager().onGroupDeleted(groupName);
}
- response.setCode(ResponseCode.SUCCESS);
+ }
+
+ private RemotingCommand deleteSubscriptionGroupList(ChannelHandlerContext
ctx,
+ RemotingCommand request) {
+ final RemotingCommand response =
RemotingCommand.createResponseCommand(null);
+
+ DeleteSubscriptionGroupListRequestBody requestBody =
DeleteSubscriptionGroupListRequestBody.decode(
+ request.getBody(), DeleteSubscriptionGroupListRequestBody.class);
+ List<String> groupNameList = requestBody == null ? null :
requestBody.getGroupNameList();
+
+ if (CollectionUtils.isEmpty(groupNameList)) {
+ response.setCode(ResponseCode.INVALID_PARAMETER);
+ response.setRemark("The specified group name list is blank.");
+ return response;
+ }
+
+ // dedup while preserving the input order
+ Set<String> groupNames = new LinkedHashSet<>();
+ for (String groupName : groupNameList) {
+ if (UtilAll.isBlank(groupName)) {
+ response.setCode(ResponseCode.INVALID_PARAMETER);
+ response.setRemark("The specified group name is blank.");
+ return response;
+ }
+ groupNames.add(groupName);
+ }
+
+ LOGGER.info("AdminBrokerProcessor#deleteSubscriptionGroupList:
groupNames={}, caller={}",
+ groupNames, RemotingHelper.parseChannelRemoteAddr(ctx.channel()));
+
+ boolean isSuccess = false;
+ try {
+ boolean cleanOffset = requestBody.isCleanOffset();
+ double maxRate =
this.brokerController.getBrokerConfig().getBatchDeleteSubscriptionGroupMaxRate();
+ com.google.common.util.concurrent.RateLimiter rateLimiter =
maxRate > 0
+ ?
com.google.common.util.concurrent.RateLimiter.create(maxRate) : null;
+
+ // Check isLiteGroupType before deletion: once the group config is
removed,
+ // findSubscriptionGroupConfig(...) may auto-recreate the group
when
+ // autoCreateSubscriptionGroup=true.
+ for (String groupName : groupNames) {
+ if (rateLimiter != null) {
+ rateLimiter.acquire();
+ }
+ boolean shouldCleanOffset = cleanOffset
+ || LiteMetadataUtil.isLiteGroupType(groupName,
this.brokerController);
+ deleteSubscriptionGroupInBroker(groupName, shouldCleanOffset,
false);
+ }
+ isSuccess = true;
+ } catch (Throwable t) {
+ LOGGER.error("Failed to delete subscription group config ", t);
+ } finally {
+ try {
+ this.brokerController.getSubscriptionGroupManager().persist();
+ } catch (Throwable t) {
+ isSuccess = false;
+ LOGGER.error("Failed to persist subscription group config
after batch delete", t);
+ }
+ }
+ response.setCode(isSuccess ? ResponseCode.SUCCESS :
ResponseCode.SYSTEM_ERROR);
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 5850309e8c..b37c35051a 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
@@ -384,18 +384,32 @@ public class SubscriptionGroupManager extends
ConfigManager {
}
}
- public void deleteSubscriptionGroupConfig(final String groupName) {
+ public void deleteSubscriptionGroupConfig(final String groupName, boolean
persist) {
SubscriptionGroupConfig old = removeSubscriptionGroupConfig(groupName);
this.forbiddenTable.remove(groupName);
if (old != null) {
log.info("delete subscription group OK, subscription group:{}",
old);
updateDataVersion();
- this.persist();
+ if (persist) {
+ this.persist();
+ }
} else {
log.warn("delete subscription group failed, subscription
groupName: {} not exist", groupName);
}
}
+ public void deleteSubscriptionGroupConfig(final String groupName) {
+ deleteSubscriptionGroupConfig(groupName, true);
+ }
+
+ public void deleteSubscriptionGroupConfigList(final List<String>
groupNames) {
+ if (groupNames == null || groupNames.isEmpty()) {
+ return;
+ }
+ for (String groupName : groupNames) {
+ deleteSubscriptionGroupConfig(groupName);
+ }
+ }
public void setSubscriptionGroupTable(ConcurrentMap<String,
SubscriptionGroupConfig> subscriptionGroupTable) {
this.subscriptionGroupTable = subscriptionGroupTable;
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 4b82aacb07..82277322cc 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
@@ -615,17 +615,32 @@ public class TopicConfigManager extends ConfigManager {
}
}
- public void deleteTopicConfig(final String topic) {
+ public void deleteTopicConfig(final String topic, boolean persist) {
TopicConfig old = removeTopicConfig(topic);
if (old != null) {
log.info("delete topic config OK, topic: {}", old);
updateDataVersion();
- this.persist();
+ if (persist) {
+ this.persist();
+ }
} else {
log.warn("delete topic config failed, topic: {} not exists",
topic);
}
}
+ public void deleteTopicConfig(final String topic) {
+ deleteTopicConfig(topic, true);
+ }
+
+ public void deleteTopicConfigList(final List<String> topics) {
+ if (topics == null || topics.isEmpty()) {
+ return;
+ }
+ for (String topic : topics) {
+ deleteTopicConfig(topic);
+ }
+ }
+
public TopicConfigSerializeWrapper buildTopicConfigSerializeWrapper() {
TopicConfigSerializeWrapper topicConfigSerializeWrapper = new
TopicConfigSerializeWrapper();
DataVersion dataVersionCopy = new 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 c342400d14..6a32adbe1d 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
@@ -44,6 +44,7 @@ import
org.apache.rocketmq.broker.metrics.BrokerMetricsManager;
import org.apache.rocketmq.broker.lite.LiteLifecycleManager;
import org.apache.rocketmq.broker.offset.ConsumerOffsetManager;
import org.apache.rocketmq.broker.schedule.ScheduleMessageService;
+import org.apache.rocketmq.broker.subscription.SubscriptionGroupManager;
import org.apache.rocketmq.broker.topic.TopicConfigManager;
import org.apache.rocketmq.common.BoundaryType;
import org.apache.rocketmq.common.BrokerConfig;
@@ -76,6 +77,8 @@ import org.apache.rocketmq.remoting.protocol.ResponseCode;
import org.apache.rocketmq.remoting.protocol.body.AclInfo;
import
org.apache.rocketmq.remoting.protocol.body.ConsumerOffsetSerializeWrapper;
import org.apache.rocketmq.remoting.protocol.body.CreateTopicListRequestBody;
+import
org.apache.rocketmq.remoting.protocol.body.DeleteSubscriptionGroupListRequestBody;
+import org.apache.rocketmq.remoting.protocol.body.DeleteTopicListRequestBody;
import org.apache.rocketmq.remoting.protocol.body.GroupList;
import org.apache.rocketmq.remoting.protocol.body.HARuntimeInfo;
import org.apache.rocketmq.remoting.protocol.body.LockBatchRequestBody;
@@ -470,11 +473,181 @@ public class AdminBrokerProcessorTest {
RemotingCommand response =
adminBrokerProcessor.processRequest(handlerContext, request);
assertThat(response.getCode()).isEqualTo(ResponseCode.SUCCESS);
- verify(topicConfigManager).deleteTopicConfig(topic);
-
verify(topicConfigManager).deleteTopicConfig(KeyBuilder.buildPopRetryTopic(topic,
"cid1", brokerConfig.isEnableRetryTopicV2()));
+ verify(topicConfigManager).deleteTopicConfig(topic, true);
+
verify(topicConfigManager).deleteTopicConfig(KeyBuilder.buildPopRetryTopic(topic,
"cid1", brokerConfig.isEnableRetryTopicV2()), true);
verify(messageStore, times(2)).deleteTopics(anySet());
}
+ @Test
+ public void testDeleteTopicListInBroker() throws Exception {
+ // empty list should fail with INVALID_PARAMETER
+ RemotingCommand emptyRequest = buildDeleteTopicListRequest(new
ArrayList<>());
+ RemotingCommand emptyResponse =
adminBrokerProcessor.processRequest(handlerContext, emptyRequest);
+
assertThat(emptyResponse.getCode()).isEqualTo(ResponseCode.INVALID_PARAMETER);
+
+ // blank topic name should fail
+ RemotingCommand blankRequest =
buildDeleteTopicListRequest(Arrays.asList("VALID_TOPIC", "", "ANOTHER"));
+ RemotingCommand blankResponse =
adminBrokerProcessor.processRequest(handlerContext, blankRequest);
+
assertThat(blankResponse.getCode()).isEqualTo(ResponseCode.INVALID_PARAMETER);
+ assertThat(blankResponse.getRemark()).isEqualTo("The specified topic
is blank.");
+
+ // system topic in batch should be rejected
+ for (String sysTopic : systemTopicSet) {
+ RemotingCommand request =
buildDeleteTopicListRequest(Arrays.asList("TEST_DELETE_TOPIC_BATCH_OK",
sysTopic));
+ RemotingCommand response =
adminBrokerProcessor.processRequest(handlerContext, request);
+
assertThat(response.getCode()).isEqualTo(ResponseCode.INVALID_PARAMETER);
+ assertThat(response.getRemark()).isEqualTo("The topic[" + sysTopic
+ "] is conflict with system topic.");
+ }
+
+ // happy path
+ List<String> topicList = Arrays.asList("TEST_DELETE_TOPIC_BATCH_1",
"TEST_DELETE_TOPIC_BATCH_2");
+ RemotingCommand request = buildDeleteTopicListRequest(topicList);
+ RemotingCommand response =
adminBrokerProcessor.processRequest(handlerContext, request);
+ assertThat(response.getCode()).isEqualTo(ResponseCode.SUCCESS);
+ }
+
+ @Test
+ public void testDeleteTopicListBatchPersist() throws Exception {
+ // Disable rate limiter for test
+ brokerController.getBrokerConfig().setBatchDeleteTopicMaxRate(0);
+
+ topicConfigManager = mock(TopicConfigManager.class);
+
when(brokerController.getTopicConfigManager()).thenReturn(topicConfigManager);
+
when(topicConfigManager.selectTopicConfig(anyString())).thenReturn(null);
+
+
when(brokerController.getConsumerOffsetManager()).thenReturn(consumerOffsetManager);
+
when(consumerOffsetManager.whichGroupByTopic(anyString())).thenReturn(new
HashSet<>());
+
+ List<String> topicList = Arrays.asList("TEST_DELETE_TOPIC_BATCH_A",
"TEST_DELETE_TOPIC_BATCH_B", "TEST_DELETE_TOPIC_BATCH_A");
+ RemotingCommand request = buildDeleteTopicListRequest(topicList);
+ RemotingCommand response =
adminBrokerProcessor.processRequest(handlerContext, request);
+ assertThat(response.getCode()).isEqualTo(ResponseCode.SUCCESS);
+
verify(topicConfigManager).deleteTopicConfig("TEST_DELETE_TOPIC_BATCH_A",
false);
+
verify(topicConfigManager).deleteTopicConfig("TEST_DELETE_TOPIC_BATCH_B",
false);
+ verify(topicConfigManager).persist();
+ }
+
+ @Test
+ public void testDeleteTopicListRateLimited() throws Exception {
+ // Enable rate limiter at 1000/s so the test exercises the throttled
path without being slow
+ brokerController.getBrokerConfig().setBatchDeleteTopicMaxRate(1000.0);
+
+ List<String> topicList = Arrays.asList("RL_TOPIC_1", "RL_TOPIC_2",
"RL_TOPIC_3");
+ RemotingCommand request = buildDeleteTopicListRequest(topicList);
+ RemotingCommand response =
adminBrokerProcessor.processRequest(handlerContext, request);
+ assertThat(response.getCode()).isEqualTo(ResponseCode.SUCCESS);
+ }
+
+ @Test
+ public void testDeleteSubscriptionGroupList() throws Exception {
+ // empty list should fail with INVALID_PARAMETER
+ RemotingCommand emptyRequest =
buildDeleteSubscriptionGroupListRequest(new ArrayList<>(), false);
+ RemotingCommand emptyResponse =
adminBrokerProcessor.processRequest(handlerContext, emptyRequest);
+
assertThat(emptyResponse.getCode()).isEqualTo(ResponseCode.INVALID_PARAMETER);
+
+ // blank group name should fail
+ RemotingCommand blankRequest =
buildDeleteSubscriptionGroupListRequest(Arrays.asList("GID-OK", "", "GID-OK2"),
true);
+ RemotingCommand blankResponse =
adminBrokerProcessor.processRequest(handlerContext, blankRequest);
+
assertThat(blankResponse.getCode()).isEqualTo(ResponseCode.INVALID_PARAMETER);
+
+ // happy path
+ List<String> groupList = Arrays.asList("GID-Group-Name-1",
"GID-Group-Name-2");
+ RemotingCommand request =
buildDeleteSubscriptionGroupListRequest(groupList, true);
+ RemotingCommand response =
adminBrokerProcessor.processRequest(handlerContext, request);
+ assertThat(response.getCode()).isEqualTo(ResponseCode.SUCCESS);
+ }
+
+ @Test
+ public void testDeleteSubscriptionGroupListDedup() throws Exception {
+ // Duplicate group names should be deduplicated, request should still
succeed
+
brokerController.getBrokerConfig().setBatchDeleteSubscriptionGroupMaxRate(0);
+
+ List<String> groupList = Arrays.asList("GID-DUP-1", "GID-DUP-2",
"GID-DUP-1");
+ RemotingCommand request =
buildDeleteSubscriptionGroupListRequest(groupList, false);
+ RemotingCommand response =
adminBrokerProcessor.processRequest(handlerContext, request);
+ assertThat(response.getCode()).isEqualTo(ResponseCode.SUCCESS);
+ }
+
+ @Test
+ public void testDeleteSubscriptionGroupListBatchPersist() throws Exception
{
+
brokerController.getBrokerConfig().setBatchDeleteSubscriptionGroupMaxRate(0);
+ SubscriptionGroupManager subscriptionGroupManager =
mock(SubscriptionGroupManager.class);
+ brokerController.setSubscriptionGroupManager(subscriptionGroupManager);
+
+ List<String> groupList = Arrays.asList("GID-BATCH-A", "GID-BATCH-B",
"GID-BATCH-A");
+ RemotingCommand request =
buildDeleteSubscriptionGroupListRequest(groupList, false);
+ RemotingCommand response =
adminBrokerProcessor.processRequest(handlerContext, request);
+
+ assertThat(response.getCode()).isEqualTo(ResponseCode.SUCCESS);
+
verify(subscriptionGroupManager).deleteSubscriptionGroupConfig("GID-BATCH-A",
false);
+
verify(subscriptionGroupManager).deleteSubscriptionGroupConfig("GID-BATCH-B",
false);
+ verify(subscriptionGroupManager).persist();
+ }
+
+ @Test
+ public void testDeleteSubscriptionGroupListRateLimited() throws Exception {
+ // Enable rate limiter at 1000/s so the test exercises the throttled
path without being slow
+
brokerController.getBrokerConfig().setBatchDeleteSubscriptionGroupMaxRate(1000.0);
+
+ List<String> groupList = Arrays.asList("GID-RL-1", "GID-RL-2",
"GID-RL-3");
+ RemotingCommand request =
buildDeleteSubscriptionGroupListRequest(groupList, true);
+ RemotingCommand response =
adminBrokerProcessor.processRequest(handlerContext, request);
+ assertThat(response.getCode()).isEqualTo(ResponseCode.SUCCESS);
+ }
+
+ @Test
+ public void testDeleteSubscriptionGroupListCleanOffset() throws Exception {
+ // cleanOffset=true should invoke removeOffset for each group
+
brokerController.getBrokerConfig().setBatchDeleteSubscriptionGroupMaxRate(0);
+
+ List<String> groupList = Arrays.asList("GID-CLEAN-1", "GID-CLEAN-2");
+ RemotingCommand request =
buildDeleteSubscriptionGroupListRequest(groupList, true);
+ RemotingCommand response =
adminBrokerProcessor.processRequest(handlerContext, request);
+ assertThat(response.getCode()).isEqualTo(ResponseCode.SUCCESS);
+ }
+
+ @Test
+ public void testDeleteSubscriptionGroupListNoCleanOffset() throws
Exception {
+ // cleanOffset=false should still succeed (offset not cleaned unless
isLiteGroupType)
+
brokerController.getBrokerConfig().setBatchDeleteSubscriptionGroupMaxRate(0);
+
+ List<String> groupList = Arrays.asList("GID-NOCLEAN-1",
"GID-NOCLEAN-2");
+ RemotingCommand request =
buildDeleteSubscriptionGroupListRequest(groupList, false);
+ RemotingCommand response =
adminBrokerProcessor.processRequest(handlerContext, request);
+ assertThat(response.getCode()).isEqualTo(ResponseCode.SUCCESS);
+ }
+
+ @Test
+ public void testDeleteTopicListWithPopRetryTopics() throws Exception {
+ // When clearRetryTopicWhenDeleteTopic=true, POP retry topics should
be collected and deleted
+ brokerController.getBrokerConfig().setBatchDeleteTopicMaxRate(0);
+
+ topicConfigManager = mock(TopicConfigManager.class);
+
when(brokerController.getTopicConfigManager()).thenReturn(topicConfigManager);
+ String retryTopic =
org.apache.rocketmq.common.KeyBuilder.buildPopRetryTopic("myTopic", "cid1",
+ brokerController.getBrokerConfig().isEnableRetryTopicV2());
+ when(topicConfigManager.selectTopicConfig(anyString())).thenAnswer(inv
-> {
+ String t = inv.getArgument(0);
+ if ("myTopic".equals(t) || retryTopic.equals(t)) {
+ return new org.apache.rocketmq.common.TopicConfig();
+ }
+ return null;
+ });
+
+
when(brokerController.getConsumerOffsetManager()).thenReturn(consumerOffsetManager);
+
when(consumerOffsetManager.whichGroupByTopic("myTopic")).thenReturn(com.google.common.collect.Sets.newHashSet("cid1"));
+
+ List<String> topicList = Arrays.asList("myTopic");
+ RemotingCommand request = buildDeleteTopicListRequest(topicList);
+ RemotingCommand response =
adminBrokerProcessor.processRequest(handlerContext, request);
+ assertThat(response.getCode()).isEqualTo(ResponseCode.SUCCESS);
+
+ // Verify both the original topic and the retry topic were deleted
+
org.mockito.Mockito.verify(topicConfigManager).deleteTopicConfig("myTopic",
false);
+
org.mockito.Mockito.verify(topicConfigManager).deleteTopicConfig(retryTopic,
false);
+ org.mockito.Mockito.verify(topicConfigManager).persist();
+ }
+
@Test
public void testGetAllTopicConfigInRocksdb() throws Exception {
if (notToBeExecuted()) {
@@ -1691,6 +1864,20 @@ public class AdminBrokerProcessorTest {
return request;
}
+ private RemotingCommand buildDeleteTopicListRequest(List<String>
topicList) {
+ DeleteTopicListRequestBody requestBody = new
DeleteTopicListRequestBody(topicList);
+ RemotingCommand request =
RemotingCommand.createRequestCommand(RequestCode.DELETE_TOPIC_IN_BROKER_LIST,
null);
+ request.setBody(requestBody.encode());
+ return request;
+ }
+
+ private RemotingCommand
buildDeleteSubscriptionGroupListRequest(List<String> groupNameList, boolean
cleanOffset) {
+ DeleteSubscriptionGroupListRequestBody requestBody = new
DeleteSubscriptionGroupListRequestBody(groupNameList, cleanOffset);
+ RemotingCommand request =
RemotingCommand.createRequestCommand(RequestCode.DELETE_SUBSCRIPTION_GROUP_LIST,
null);
+ request.setBody(requestBody.encode());
+ return request;
+ }
+
private MessageExt createDefaultMessageExt() {
MessageExt messageExt = new MessageExt();
messageExt.setMsgId("12345678");
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 bc34e26bf5..9f709f6502 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
@@ -163,6 +163,23 @@ public class SubscriptionGroupManagerTest {
assertThat(subscriptionGroupManager.getSubscriptionGroupTable().get(groupName)).isNotNull());
}
+ @Test
+ public void testDeleteSubscriptionGroupConfigWithDeferredPersist() {
+ Mockito.doNothing().when(subscriptionGroupManager).persist();
+
+ SubscriptionGroupConfig deferredGroup = new SubscriptionGroupConfig();
+ deferredGroup.setGroupName("deferred-group");
+
subscriptionGroupManager.getSubscriptionGroupTable().put(deferredGroup.getGroupName(),
deferredGroup);
+
subscriptionGroupManager.deleteSubscriptionGroupConfig(deferredGroup.getGroupName(),
false);
+ verify(subscriptionGroupManager, never()).persist();
+
+ SubscriptionGroupConfig persistedGroup = new SubscriptionGroupConfig();
+ persistedGroup.setGroupName("persisted-group");
+
subscriptionGroupManager.getSubscriptionGroupTable().put(persistedGroup.getGroupName(),
persistedGroup);
+
subscriptionGroupManager.deleteSubscriptionGroupConfig(persistedGroup.getGroupName(),
true);
+ verify(subscriptionGroupManager).persist();
+ }
+
@Test
public void testSubGroupTable() {
// Empty SubscriptionGroupManager
@@ -234,4 +251,4 @@ public class SubscriptionGroupManagerTest {
}
}
-}
\ 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 085c3f9cfb..fd91b5c0eb 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
@@ -140,6 +140,24 @@ public class TopicConfigManagerTest {
Assert.assertEquals("only add attribute is supported while creating
topic. key: " + key, runtimeException.getMessage());
}
+ @Test
+ public void testDeleteTopicConfigWithDeferredPersist() {
+ TopicConfigManager manager = Mockito.spy(topicConfigManager);
+ Mockito.doNothing().when(manager).persist();
+
+ TopicConfig deferredTopic = new TopicConfig();
+ deferredTopic.setTopicName("deferred-topic");
+ manager.getTopicConfigTable().put(deferredTopic.getTopicName(),
deferredTopic);
+ manager.deleteTopicConfig(deferredTopic.getTopicName(), false);
+ Mockito.verify(manager, Mockito.never()).persist();
+
+ TopicConfig persistedTopic = new TopicConfig();
+ persistedTopic.setTopicName("persisted-topic");
+ manager.getTopicConfigTable().put(persistedTopic.getTopicName(),
persistedTopic);
+ manager.deleteTopicConfig(persistedTopic.getTopicName(), true);
+ Mockito.verify(manager).persist();
+ }
+
@Test
public void testAddWrongValueOnCreating() {
Map<String, String> attributes = new HashMap<>();
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 8294ffd422..6999859491 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
@@ -125,6 +125,8 @@ import
org.apache.rocketmq.remoting.protocol.body.ConsumeStatsList;
import org.apache.rocketmq.remoting.protocol.body.ConsumerConnection;
import org.apache.rocketmq.remoting.protocol.body.ConsumerRunningInfo;
import org.apache.rocketmq.remoting.protocol.body.CreateTopicListRequestBody;
+import
org.apache.rocketmq.remoting.protocol.body.DeleteSubscriptionGroupListRequestBody;
+import org.apache.rocketmq.remoting.protocol.body.DeleteTopicListRequestBody;
import org.apache.rocketmq.remoting.protocol.body.EpochEntryCache;
import
org.apache.rocketmq.remoting.protocol.body.GetBrokerLiteInfoResponseBody;
import org.apache.rocketmq.remoting.protocol.body.GetConsumerStatusBody;
@@ -2197,6 +2199,27 @@ public class MQClientAPIImpl implements
NameServerUpdateCallback, StartAndShutdo
throw new MQClientException(response.getCode(), response.getRemark());
}
+ public void deleteTopicInBrokerList(final String addr, final List<String>
topicList, final long timeoutMillis)
+ throws RemotingException, InterruptedException, MQClientException {
+ DeleteTopicListRequestBody requestBody = new
DeleteTopicListRequestBody(topicList);
+ RemotingCommand request =
RemotingCommand.createRequestCommand(RequestCode.DELETE_TOPIC_IN_BROKER_LIST,
null);
+ request.setBody(requestBody.encode());
+ RemotingCommand response =
this.remotingClient.invokeSync(MixAll.brokerVIPChannel(this.clientConfig.isVipChannelEnabled(),
addr),
+ request, timeoutMillis);
+ if (response == null) {
+ throw new MQClientException(ResponseCode.SYSTEM_ERROR, "response
is null when deleting topic list in broker");
+ }
+ switch (response.getCode()) {
+ case ResponseCode.SUCCESS: {
+ return;
+ }
+ default:
+ break;
+ }
+
+ throw new MQClientException(response.getCode(), response.getRemark());
+ }
+
public void deleteTopicInNameServer(final String addr, final String topic,
final long timeoutMillis)
throws RemotingException, InterruptedException, MQClientException {
DeleteTopicFromNamesrvRequestHeader requestHeader = new
DeleteTopicFromNamesrvRequestHeader();
@@ -2257,6 +2280,29 @@ public class MQClientAPIImpl implements
NameServerUpdateCallback, StartAndShutdo
throw new MQClientException(response.getCode(), response.getRemark());
}
+ public void deleteSubscriptionGroupList(final String addr, final
List<String> groupNameList, final boolean cleanOffset,
+ final long timeoutMillis)
+ throws RemotingException, InterruptedException, MQClientException {
+ DeleteSubscriptionGroupListRequestBody requestBody = new
DeleteSubscriptionGroupListRequestBody(groupNameList, cleanOffset);
+ RemotingCommand request =
RemotingCommand.createRequestCommand(RequestCode.DELETE_SUBSCRIPTION_GROUP_LIST,
null);
+ request.setBody(requestBody.encode());
+
+ RemotingCommand response =
this.remotingClient.invokeSync(MixAll.brokerVIPChannel(this.clientConfig.isVipChannelEnabled(),
addr),
+ request, timeoutMillis);
+ if (response == null) {
+ throw new MQClientException(ResponseCode.SYSTEM_ERROR, "response
is null when deleting subscription group list");
+ }
+ switch (response.getCode()) {
+ case ResponseCode.SUCCESS: {
+ return;
+ }
+ default:
+ break;
+ }
+
+ throw new MQClientException(response.getCode(), response.getRemark());
+ }
+
public String getKVConfigValue(final String namespace, final String key,
final long timeoutMillis)
throws RemotingException, MQClientException, InterruptedException {
GetKVConfigRequestHeader requestHeader = new
GetKVConfigRequestHeader();
diff --git
a/client/src/test/java/org/apache/rocketmq/client/impl/MQClientAPIImplTest.java
b/client/src/test/java/org/apache/rocketmq/client/impl/MQClientAPIImplTest.java
index 27b3d68571..096853c6e6 100644
---
a/client/src/test/java/org/apache/rocketmq/client/impl/MQClientAPIImplTest.java
+++
b/client/src/test/java/org/apache/rocketmq/client/impl/MQClientAPIImplTest.java
@@ -1364,6 +1364,68 @@ public class MQClientAPIImplTest {
mqClientAPI.deleteSubscriptionGroup(defaultBrokerAddr, "", true,
defaultTimeout);
}
+ @Test
+ public void testDeleteTopicInBrokerList() throws RemotingException,
InterruptedException, MQClientException {
+ mockInvokeSync();
+ mqClientAPI.deleteTopicInBrokerList(defaultBrokerAddr,
java.util.Arrays.asList("topicA", "topicB"), defaultTimeout);
+ }
+
+ @Test(expected = MQClientException.class)
+ public void testDeleteTopicInBrokerListFail() throws RemotingException,
InterruptedException, MQClientException {
+ when(response.getCode()).thenReturn(ResponseCode.SYSTEM_ERROR);
+ when(response.getRemark()).thenReturn("error");
+ when(remotingClient.invokeSync(any(), any(),
anyLong())).thenReturn(response);
+ mqClientAPI.deleteTopicInBrokerList(defaultBrokerAddr,
java.util.Arrays.asList("topicA"), defaultTimeout);
+ }
+
+ @Test(expected = MQClientException.class)
+ public void testDeleteTopicInBrokerListNullResponse() throws
RemotingException, InterruptedException, MQClientException {
+ when(remotingClient.invokeSync(any(), any(),
anyLong())).thenReturn(null);
+ mqClientAPI.deleteTopicInBrokerList(defaultBrokerAddr,
java.util.Arrays.asList("topicA"), defaultTimeout);
+ }
+
+ @Test
+ public void testDeleteTopicInBrokerListVerifyRequest() throws Exception {
+ mockInvokeSync();
+ org.mockito.ArgumentCaptor<RemotingCommand> captor =
org.mockito.ArgumentCaptor.forClass(RemotingCommand.class);
+ mqClientAPI.deleteTopicInBrokerList(defaultBrokerAddr,
java.util.Arrays.asList("t1", "t2"), defaultTimeout);
+ org.mockito.Mockito.verify(remotingClient).invokeSync(any(),
captor.capture(), anyLong());
+ RemotingCommand captured = captor.getValue();
+ assertEquals(RequestCode.DELETE_TOPIC_IN_BROKER_LIST,
captured.getCode());
+ assertNotNull(captured.getBody());
+ }
+
+ @Test
+ public void testDeleteSubscriptionGroupList() throws RemotingException,
InterruptedException, MQClientException {
+ mockInvokeSync();
+ mqClientAPI.deleteSubscriptionGroupList(defaultBrokerAddr,
java.util.Arrays.asList("groupA", "groupB"), true, defaultTimeout);
+ }
+
+ @Test(expected = MQClientException.class)
+ public void testDeleteSubscriptionGroupListFail() throws
RemotingException, InterruptedException, MQClientException {
+ when(response.getCode()).thenReturn(ResponseCode.SYSTEM_ERROR);
+ when(response.getRemark()).thenReturn("error");
+ when(remotingClient.invokeSync(any(), any(),
anyLong())).thenReturn(response);
+ mqClientAPI.deleteSubscriptionGroupList(defaultBrokerAddr,
java.util.Arrays.asList("groupA"), false, defaultTimeout);
+ }
+
+ @Test(expected = MQClientException.class)
+ public void testDeleteSubscriptionGroupListNullResponse() throws
RemotingException, InterruptedException, MQClientException {
+ when(remotingClient.invokeSync(any(), any(),
anyLong())).thenReturn(null);
+ mqClientAPI.deleteSubscriptionGroupList(defaultBrokerAddr,
java.util.Arrays.asList("groupA"), true, defaultTimeout);
+ }
+
+ @Test
+ public void testDeleteSubscriptionGroupListVerifyRequest() throws
Exception {
+ mockInvokeSync();
+ org.mockito.ArgumentCaptor<RemotingCommand> captor =
org.mockito.ArgumentCaptor.forClass(RemotingCommand.class);
+ mqClientAPI.deleteSubscriptionGroupList(defaultBrokerAddr,
java.util.Arrays.asList("g1", "g2"), true, defaultTimeout);
+ org.mockito.Mockito.verify(remotingClient).invokeSync(any(),
captor.capture(), anyLong());
+ RemotingCommand captured = captor.getValue();
+ assertEquals(RequestCode.DELETE_SUBSCRIPTION_GROUP_LIST,
captured.getCode());
+ assertNotNull(captured.getBody());
+ }
+
@Test
public void assertGetKVConfigValue() throws RemotingException,
InterruptedException, MQClientException {
mockInvokeSync();
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 3c0ba3f6ca..3ab78fba5f 100644
--- a/common/src/main/java/org/apache/rocketmq/common/BrokerConfig.java
+++ b/common/src/main/java/org/apache/rocketmq/common/BrokerConfig.java
@@ -379,6 +379,18 @@ public class BrokerConfig extends BrokerIdentity {
private boolean validateSystemTopicWhenUpdateTopic = true;
+ /**
+ * Maximum rate (permits per second) for batch topic deletion.
+ * Setting to 0 or negative disables rate limiting.
+ */
+ private double batchDeleteTopicMaxRate = 10.0;
+
+ /**
+ * Maximum rate (permits per second) for batch subscription-group deletion.
+ * Setting to 0 or negative disables rate limiting.
+ */
+ private double batchDeleteSubscriptionGroupMaxRate = 10.0;
+
/**
* It is an important basis for the controller to choose the broker master.
* The lower the value of brokerElectionPriority, the higher the priority
of the broker being selected as the master.
@@ -2062,6 +2074,22 @@ public class BrokerConfig extends BrokerIdentity {
this.validateSystemTopicWhenUpdateTopic =
validateSystemTopicWhenUpdateTopic;
}
+ public double getBatchDeleteTopicMaxRate() {
+ return batchDeleteTopicMaxRate;
+ }
+
+ public void setBatchDeleteTopicMaxRate(double batchDeleteTopicMaxRate) {
+ this.batchDeleteTopicMaxRate = batchDeleteTopicMaxRate;
+ }
+
+ public double getBatchDeleteSubscriptionGroupMaxRate() {
+ return batchDeleteSubscriptionGroupMaxRate;
+ }
+
+ public void setBatchDeleteSubscriptionGroupMaxRate(double
batchDeleteSubscriptionGroupMaxRate) {
+ this.batchDeleteSubscriptionGroupMaxRate =
batchDeleteSubscriptionGroupMaxRate;
+ }
+
public boolean isEstimateAccumulation() {
return estimateAccumulation;
}
diff --git
a/remoting/src/main/java/org/apache/rocketmq/remoting/protocol/RequestCode.java
b/remoting/src/main/java/org/apache/rocketmq/remoting/protocol/RequestCode.java
index b32dbbc87e..32530905e4 100644
---
a/remoting/src/main/java/org/apache/rocketmq/remoting/protocol/RequestCode.java
+++
b/remoting/src/main/java/org/apache/rocketmq/remoting/protocol/RequestCode.java
@@ -311,4 +311,7 @@ public class RequestCode {
public static final int AUTH_LIST_ACL = 3010;
public static final int SWITCH_TIMER_ENGINE = 5001;
+
+ public static final int DELETE_TOPIC_IN_BROKER_LIST = 5002;
+ public static final int DELETE_SUBSCRIPTION_GROUP_LIST = 5003;
}
diff --git
a/remoting/src/main/java/org/apache/rocketmq/remoting/protocol/body/DeleteSubscriptionGroupListRequestBody.java
b/remoting/src/main/java/org/apache/rocketmq/remoting/protocol/body/DeleteSubscriptionGroupListRequestBody.java
new file mode 100644
index 0000000000..7c78fb69af
--- /dev/null
+++
b/remoting/src/main/java/org/apache/rocketmq/remoting/protocol/body/DeleteSubscriptionGroupListRequestBody.java
@@ -0,0 +1,56 @@
+/*
+ * 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.body;
+
+import java.util.List;
+import org.apache.rocketmq.remoting.annotation.CFNotNull;
+import org.apache.rocketmq.remoting.protocol.RemotingSerializable;
+
+public class DeleteSubscriptionGroupListRequestBody extends
RemotingSerializable {
+ @CFNotNull
+ private List<String> groupNameList;
+
+ private boolean cleanOffset = false;
+
+ public DeleteSubscriptionGroupListRequestBody() {
+ }
+
+ public DeleteSubscriptionGroupListRequestBody(List<String> groupNameList) {
+ this.groupNameList = groupNameList;
+ }
+
+ public DeleteSubscriptionGroupListRequestBody(List<String> groupNameList,
boolean cleanOffset) {
+ this.groupNameList = groupNameList;
+ this.cleanOffset = cleanOffset;
+ }
+
+ public List<String> getGroupNameList() {
+ return groupNameList;
+ }
+
+ public void setGroupNameList(List<String> groupNameList) {
+ this.groupNameList = groupNameList;
+ }
+
+ public boolean isCleanOffset() {
+ return cleanOffset;
+ }
+
+ public void setCleanOffset(boolean cleanOffset) {
+ this.cleanOffset = cleanOffset;
+ }
+}
diff --git
a/remoting/src/main/java/org/apache/rocketmq/remoting/protocol/body/DeleteTopicListRequestBody.java
b/remoting/src/main/java/org/apache/rocketmq/remoting/protocol/body/DeleteTopicListRequestBody.java
new file mode 100644
index 0000000000..8677e3b338
--- /dev/null
+++
b/remoting/src/main/java/org/apache/rocketmq/remoting/protocol/body/DeleteTopicListRequestBody.java
@@ -0,0 +1,41 @@
+/*
+ * 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.body;
+
+import java.util.List;
+import org.apache.rocketmq.remoting.annotation.CFNotNull;
+import org.apache.rocketmq.remoting.protocol.RemotingSerializable;
+
+public class DeleteTopicListRequestBody extends RemotingSerializable {
+ @CFNotNull
+ private List<String> topicList;
+
+ public DeleteTopicListRequestBody() {
+ }
+
+ public DeleteTopicListRequestBody(List<String> topicList) {
+ this.topicList = topicList;
+ }
+
+ public List<String> getTopicList() {
+ return topicList;
+ }
+
+ public void setTopicList(List<String> topicList) {
+ this.topicList = topicList;
+ }
+}
