@Override
    public Set<String> removeTopicStore() {
        if (isRemovingTopic.get()) {
            return null;
        }
        if (!isRemovingTopic.compareAndSet(false, true)) {
            return null;
        }
        try {
            Set<String> removedTopics = new HashSet<>();
            Map<String, TopicMetadata> removedTopicMap =
                    this.metadataManager.getRemovedTopicConfigMap();
            if (removedTopicMap.isEmpty()) {
                return removedTopics;
            }
            Set<String> targetTopics = new HashSet<>();
            for (Map.Entry<String, TopicMetadata> entry : removedTopicMap.entrySet()) {
                if (entry.getKey() == null || entry.getValue() == null) {
                    continue;
                }
                if (entry.getValue().getStatusId() == TStatusConstants.STATUS_TOPIC_SOFT_REMOVE) {
                    targetTopics.add(entry.getKey());
                }
            }
            if (targetTopics.isEmpty()) {
                return removedTopics;
            }
            logger.info("[Remove Topic] start remove topics : {}", targetTopics);
            for (String tmpTopic : targetTopics) {
                ConcurrentHashMap<Integer, MessageStore> topicStores =
                        dataStores.get(tmpTopic);
                if (topicStores != null) {
                    Set<Integer> storeIds = topicStores.keySet();
                    for (Integer storeId : storeIds) {
                        try {
                            MessageStore tmpStore = topicStores.remove(storeId);
                            tmpStore.close();
                            if (topicStores.isEmpty()) {
                                this.dataStores.remove(tmpTopic);
                            }
                        } catch (Throwable ee) {
                            logger.error(new StringBuilder(512)
                                    .append("[Remove Topic] Close removed store failure, storeKey=")
                                    .append(tmpTopic).append("-").append(storeId).toString(), ee);
                        }
                    }
                }
                TopicMetadata tmpTopicConf = removedTopicMap.get(tmpTopic);
                if (tmpTopicConf != null) {
                    StringBuilder sBuilder = new StringBuilder(512);
                    for (int storeId = 0; storeId < tmpTopicConf.getNumTopicStores(); storeId++) {
                        String storeDir = sBuilder.append(tubeConfig.getPrimaryPath())
                                .append(File.separator).append(tmpTopic).append("-")
                                .append(storeId).toString();
                        sBuilder.delete(0, sBuilder.length());
                        logger.info("[Remove Topic] remove topic files : {}", storeDir);
                        try {
                            delTopicFiles(storeDir);
                        } catch (Throwable e) {
                            logger.error("[Remove Topic] remove topic files error : ", e);
                        }
                        ThreadUtils.sleep(50);
                    }
                    tmpTopicConf.setStatusId(TStatusConstants.STATUS_TOPIC_HARD_REMOVE);
                    removedTopics.add(tmpTopic);
                }
                ThreadUtils.sleep(100);
            }
            logger.info("[Remove Topic] finished remove topics : {}", removedTopics);
            return removedTopics;
        } finally {
            this.isRemovingTopic.set(false);
        }
    }
Initial the message-store manager.

@param tubeBroker      the broker instance
@param tubeConfig      the initial configure
@throws IOException    the exception during processing
inlong-tubemq/tubemq-server/src/main/java/org/apache/inlong/tubemq/server/broker/msgstore/MessageStoreManager.java