private synchronized RemotingCommand deleteTopic(ChannelHandlerContext ctx,
        RemotingCommand request) throws RemotingCommandException {
        final RemotingCommand response = RemotingCommand.createResponseCommand(null);
        DeleteTopicRequestHeader requestHeader =
            (DeleteTopicRequestHeader) request.decodeCommandCustomHeader(DeleteTopicRequestHeader.class);

        LOGGER.info("AdminBrokerProcessor#deleteTopic: broker receive request to delete topic={}, caller={}",
            requestHeader.getTopic(), RemotingHelper.parseChannelRemoteAddr(ctx.channel()));

        String topic = requestHeader.getTopic();

        if (UtilAll.isBlank(topic)) {
            response.setCode(ResponseCode.SYSTEM_ERROR);
            response.setRemark("The specified topic is blank.");
            return response;
        }

        if (brokerController.getBrokerConfig().isValidateSystemTopicWhenUpdateTopic()) {
            if (TopicValidator.isSystemTopic(topic)) {
                response.setCode(ResponseCode.SYSTEM_ERROR);
                response.setRemark("The topic[" + topic + "] is conflict with system topic.");
                return response;
            }
        }

        final Set<String> groups = this.brokerController.getConsumerOffsetManager().whichGroupByTopic(topic);
        // delete pop retry topics first
        try {
            for (String group : groups) {
                final String popRetryTopicV2 = KeyBuilder.buildPopRetryTopic(topic, group, true);
                if (brokerController.getTopicConfigManager().selectTopicConfig(popRetryTopicV2) != null) {
                    deleteTopicInBroker(popRetryTopicV2);
                }
                final String popRetryTopicV1 = KeyBuilder.buildPopRetryTopicV1(topic, group);
                if (brokerController.getTopicConfigManager().selectTopicConfig(popRetryTopicV1) != null) {
                    deleteTopicInBroker(popRetryTopicV1);
                }
            }
            // delete topic
            deleteTopicInBroker(topic);
        } catch (Throwable t) {
            return buildErrorResponse(ResponseCode.SYSTEM_ERROR, t.getMessage());
        }
        response.setCode(ResponseCode.SUCCESS);
        response.setRemark(null);
        return response;
    }
@Test
    public void testDeleteTopic() throws Exception {
        //test system topic
        for (String topic : systemTopicSet) {
            RemotingCommand request = buildDeleteTopicRequest(topic);
            RemotingCommand response = adminBrokerProcessor.processRequest(handlerContext, request);
            assertThat(response.getCode()).isEqualTo(ResponseCode.SYSTEM_ERROR);
            assertThat(response.getRemark()).isEqualTo("The topic[" + topic + "] is conflict with system topic.");
        }

        String topic = "TEST_DELETE_TOPIC";
        RemotingCommand request = buildDeleteTopicRequest(topic);
        RemotingCommand response = adminBrokerProcessor.processRequest(handlerContext, request);
        assertThat(response.getCode()).isEqualTo(ResponseCode.SUCCESS);
    }