@Override
    public int deleteTopics(final Set<String> deleteTopics) {
        if (deleteTopics == null || deleteTopics.isEmpty()) {
            return 0;
        }

        int deleteCount = 0;
        for (String topic : deleteTopics) {
            ConcurrentMap<Integer, ConsumeQueueInterface> queueTable = this.consumeQueueStore.findConsumeQueueMap(topic);

            if (queueTable == null || queueTable.isEmpty()) {
                continue;
            }

            for (ConsumeQueueInterface cq : queueTable.values()) {
                try {
                    this.consumeQueueStore.destroy(cq);
                } catch (RocksDBException e) {
                    LOGGER.error("DeleteTopic: ConsumeQueue cleans error!, topic={}, queueId={}", cq.getTopic(), cq.getQueueId(), e);
                }
                LOGGER.info("DeleteTopic: ConsumeQueue has been cleaned, topic={}, queueId={}", cq.getTopic(), cq.getQueueId());
                this.consumeQueueStore.removeTopicQueueTable(cq.getTopic(), cq.getQueueId());
            }

            // remove topic from cq table
            this.consumeQueueStore.getConsumeQueueTable().remove(topic);

            if (this.brokerConfig.isAutoDeleteUnusedStats()) {
                this.brokerStatsManager.onTopicDeleted(topic);
            }

            // destroy consume queue dir
            String consumeQueueDir = StorePathConfigHelper.getStorePathConsumeQueue(
                this.messageStoreConfig.getStorePathRootDir()) + File.separator + topic;
            String consumeQueueExtDir = StorePathConfigHelper.getStorePathConsumeQueueExt(
                this.messageStoreConfig.getStorePathRootDir()) + File.separator + topic;
            String batchConsumeQueueDir = StorePathConfigHelper.getStorePathBatchConsumeQueue(
                this.messageStoreConfig.getStorePathRootDir()) + File.separator + topic;

            UtilAll.deleteEmptyDirectory(new File(consumeQueueDir));
            UtilAll.deleteEmptyDirectory(new File(consumeQueueExtDir));
            UtilAll.deleteEmptyDirectory(new File(batchConsumeQueueDir));

            LOGGER.info("DeleteTopic: Topic has been destroyed, topic={}", topic);
            deleteCount++;
        }
        return deleteCount;
    }
@Test
    public void testDeleteTopics() {
        MessageStoreConfig messageStoreConfig = messageStore.getMessageStoreConfig();
        ConcurrentMap<String, ConcurrentMap<Integer, ConsumeQueueInterface>> consumeQueueTable =
            ((DefaultMessageStore) messageStore).getConsumeQueueTable();
        for (int i = 0; i < 10; i++) {
            ConcurrentMap<Integer, ConsumeQueueInterface> cqTable = new ConcurrentHashMap<>();
            String topicName = "topic-" + i;
            for (int j = 0; j < 4; j++) {
                ConsumeQueue consumeQueue = new ConsumeQueue(topicName, j, messageStoreConfig.getStorePathRootDir(),
                    messageStoreConfig.getMappedFileSizeConsumeQueue(), messageStore);
                cqTable.put(j, consumeQueue);
            }
            consumeQueueTable.put(topicName, cqTable);
        }
        Assert.assertEquals(consumeQueueTable.size(), 10);
        HashSet<String> resultSet = Sets.newHashSet("topic-3", "topic-5");
        messageStore.deleteTopics(Sets.difference(consumeQueueTable.keySet(), resultSet));
        Assert.assertEquals(consumeQueueTable.size(), 2);
        Assert.assertEquals(resultSet, consumeQueueTable.keySet());
    }