private void loadMessageStores(final BrokerConfig tubeConfig)
            throws IOException, InterruptedException {
        StringBuilder sBuilder = new StringBuilder(512);
        logger.info(sBuilder.append("[Store Manager] Begin to load message stores from path ")
                .append(tubeConfig.getPrimaryPath()).toString());
        sBuilder.delete(0, sBuilder.length());
        final long start = System.currentTimeMillis();
        final AtomicInteger errCnt = new AtomicInteger(0);
        final AtomicInteger finishCnt = new AtomicInteger(0);
        List<Callable<MessageStore>> tasks = new ArrayList<>();
        for (final File dir : this.getLogDirSet(tubeConfig)) {
            if (dir == null) {
                continue;
            }
            final File[] ls = dir.listFiles();
            if (ls == null) {
                continue;
            }
            for (final File subDir : ls) {
                if (subDir == null) {
                    continue;
                }
                if (!subDir.isDirectory()) {
                    continue;
                }
                final String name = subDir.getName();
                final int index = name.lastIndexOf('-');
                if (index < 0) {
                    logger.warn(sBuilder.append("[Store Manager] Ignore invalid directory:")
                            .append(subDir.getAbsolutePath()).toString());
                    sBuilder.delete(0, sBuilder.length());
                    continue;
                }
                final String topic = name.substring(0, index);
                final TopicMetadata topicMetadata = metadataManager.getTopicMetadata(topic);
                if (topicMetadata == null) {
                    logger.warn(sBuilder
                            .append("[Store Manager] No valid topic config for topic data directories:")
                            .append(topic).toString());
                    sBuilder.delete(0, sBuilder.length());
                    continue;
                }
                final int storeId = Integer.parseInt(name.substring(index + 1));
                final MessageStoreManager messageStoreManager = this;
                tasks.add(new Callable<MessageStore>() {

                    @Override
                    public MessageStore call() throws Exception {
                        MessageStore msgStore = null;
                        try {
                            msgStore = new MessageStore(messageStoreManager,
                                    topicMetadata, storeId, tubeConfig, maxMsgTransferSize);
                            ConcurrentHashMap<Integer, MessageStore> map =
                                    dataStores.get(msgStore.getTopic());
                            if (map == null) {
                                map = new ConcurrentHashMap<>();
                                ConcurrentHashMap<Integer, MessageStore> oldmap =
                                        dataStores.putIfAbsent(msgStore.getTopic(), map);
                                if (oldmap != null) {
                                    map = oldmap;
                                }
                            }
                            MessageStore oldMsgStore = map.putIfAbsent(msgStore.getStoreId(), msgStore);
                            if (oldMsgStore != null) {
                                try {
                                    msgStore.close();
                                    logger.info(new StringBuilder(512)
                                            .append("[Store Manager] Close duplicated messageStore ")
                                            .append(msgStore.getStoreKey()).toString());
                                } catch (Throwable e2) {
                                    //
                                    logger.info("[Store Manager] Close duplicated messageStore failure", e2);
                                }
                            }
                        } catch (Throwable e2) {
                            errCnt.incrementAndGet();
                            logger.error(new StringBuilder(512).append("[Store Manager] Loaded ")
                                    .append(subDir.getAbsolutePath())
                                    .append("message store failure:").toString(), e2);
                        } finally {
                            finishCnt.incrementAndGet();
                        }
                        return null;
                    }
                });
            }
        }
        this.loadStoresInParallel(tasks);
        tasks.clear();
        if (errCnt.get() > 0) {
            throw new RuntimeException(
                    "[Store Manager] failure to load message stores, please check load logger and fix first!");
        }
        logger.info(sBuilder.append("[Store Manager] End to load message stores in ")
                .append((System.currentTimeMillis() - start) / 1000).append(" secs").toString());
    }
Load stores sequential.

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