From 74e90f7c283d4030e4c89a35f46dec77419bcb9e Mon Sep 17 00:00:00 2001 From: Jiahua Wang <90542846+wang-jiahua@users.noreply.github.com> Date: Mon, 13 Jul 2026 10:07:43 +0800 Subject: [PATCH] [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 Co-authored-by: fuchong --- .../DefaultAuthorizationContextBuilder.java | 35 +++ ...efaultAuthorizationContextBuilderTest.java | 45 ++++ .../processor/AdminBrokerProcessor.java | 206 ++++++++++++++++-- .../SubscriptionGroupManager.java | 18 +- .../broker/topic/TopicConfigManager.java | 19 +- .../processor/AdminBrokerProcessorTest.java | 191 +++++++++++++++- .../SubscriptionGroupManagerTest.java | 19 +- .../broker/topic/TopicConfigManagerTest.java | 18 ++ .../rocketmq/client/impl/MQClientAPIImpl.java | 46 ++++ .../client/impl/MQClientAPIImplTest.java | 62 ++++++ .../apache/rocketmq/common/BrokerConfig.java | 28 +++ .../remoting/protocol/RequestCode.java | 3 + ...eleteSubscriptionGroupListRequestBody.java | 56 +++++ .../body/DeleteTopicListRequestBody.java | 41 ++++ 14 files changed, 757 insertions(+), 30 deletions(-) create mode 100644 remoting/src/main/java/org/apache/rocketmq/remoting/protocol/body/DeleteSubscriptionGroupListRequestBody.java create mode 100644 remoting/src/main/java/org/apache/rocketmq/remoting/protocol/body/DeleteTopicListRequestBody.java 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 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 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 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 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 topicsToClean) { + final Set 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,22 +1783,92 @@ 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()); - } - - if (this.brokerController.getBrokerConfig().isAutoDeleteUnusedStats()) { - this.brokerController.getBrokerStatsManager().onGroupDeleted(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(groupName); + } + } + + private RemotingCommand deleteSubscriptionGroupList(ChannelHandlerContext ctx, + RemotingCommand request) { + final RemotingCommand response = RemotingCommand.createResponseCommand(null); + + DeleteSubscriptionGroupListRequestBody requestBody = DeleteSubscriptionGroupListRequestBody.decode( + request.getBody(), DeleteSubscriptionGroupListRequestBody.class); + List 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 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; + } + private RemotingCommand getTopicStatsInfo(ChannelHandlerContext ctx, RemotingCommand request) throws RemotingCommandException { final RemotingCommand response = RemotingCommand.createResponseCommand(null); 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 groupNames) { + if (groupNames == null || groupNames.isEmpty()) { + return; + } + for (String groupName : groupNames) { + deleteSubscriptionGroupConfig(groupName); + } + } public void setSubscriptionGroupTable(ConcurrentMap 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 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 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 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 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 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 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 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 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 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 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 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 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 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 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 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 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 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 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 groupNameList; + + private boolean cleanOffset = false; + + public DeleteSubscriptionGroupListRequestBody() { + } + + public DeleteSubscriptionGroupListRequestBody(List groupNameList) { + this.groupNameList = groupNameList; + } + + public DeleteSubscriptionGroupListRequestBody(List groupNameList, boolean cleanOffset) { + this.groupNameList = groupNameList; + this.cleanOffset = cleanOffset; + } + + public List getGroupNameList() { + return groupNameList; + } + + public void setGroupNameList(List 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 topicList; + + public DeleteTopicListRequestBody() { + } + + public DeleteTopicListRequestBody(List topicList) { + this.topicList = topicList; + } + + public List getTopicList() { + return topicList; + } + + public void setTopicList(List topicList) { + this.topicList = topicList; + } +}