@Override
    public void close() {
        if (this.stopped.get()) {
            return;
        }
        if (this.stopped.compareAndSet(false, true)) {
            logger.info("[Store Manager] begin close store manager......");
            this.logClearScheduler.shutdownNow();
            this.unFlushDiskScheduler.shutdownNow();
            this.unFlushMemScheduler.shutdownNow();
            for (Map.Entry<String, ConcurrentHashMap<Integer, MessageStore>> entry : this.dataStores.entrySet()) {
                if (entry.getValue() != null) {
                    ConcurrentHashMap<Integer, MessageStore> subMap = entry.getValue();
                    for (Map.Entry<Integer, MessageStore> subEntry : subMap.entrySet()) {
                        if (subEntry.getValue() != null) {
                            try {
                                subEntry.getValue().close();
                            } catch (final Throwable e) {
                                logger.error(new StringBuilder(512)
                                        .append("[Store Manager] Try to run close  ")
                                        .append(subEntry.getValue().getStoreKey()).append(" failed").toString(), e);
                            }
                        }
                    }
                }
            }
            this.dataStores.clear();
            logger.info("[Store Manager] Store Manager stopped!");
        }
    }
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