[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 <wangjiahua.wjh@alibaba-inc.com> Co-authored-by: fuchong <yubao.fyb@alibaba-inc.com>
2026-08-28 20:09:14 +08:00 · 2026-07-13 10:07:43 +08:00
parent c4ce550a57
commit 74e90f7c28
14 changed files with 757 additions and 30 deletions
@@ -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;
@@ -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,
@@ -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,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<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;
+    }
+
    private RemotingCommand getTopicStatsInfo(ChannelHandlerContext ctx,
        RemotingCommand request) throws RemotingCommandException {
        final RemotingCommand response = RemotingCommand.createResponseCommand(null);
@@ -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;
@@ -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();
@@ -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");
@@ -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 {
        }
    }

-}
+}
@@ -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<>();
@@ -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();
@@ -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();
@@ -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;
    }
@@ -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;
 }
@@ -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;
+    }
+}
@@ -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;
+    }
+}