@Override
    public MessageStore getOrCreateMessageStore(final String topic,
            final int partition) throws Throwable {
        StringBuilder sBuilder = new StringBuilder(512);
        final int storeId = partition < TBaseConstants.META_STORE_INS_BASE
                ? 0
                : partition / TBaseConstants.META_STORE_INS_BASE;
        int realPartition = partition < TBaseConstants.META_STORE_INS_BASE
                ? partition
                : partition % TBaseConstants.META_STORE_INS_BASE;
        final String dataStoreToken = sBuilder.append("tube_store_manager_").append(topic).toString();
        sBuilder.delete(0, sBuilder.length());
        if (realPartition < 0 || realPartition >= this.metadataManager.getNumPartitions(topic)) {
            throw new IllegalArgumentException(sBuilder.append("Wrong partition value ")
                    .append(partition).append(",valid partitions in (0,")
                    .append(this.metadataManager.getNumPartitions(topic) - 1)
                    .append(")").toString());
        }
        ConcurrentHashMap<Integer, MessageStore> dataMap = dataStores.get(topic);
        if (dataMap == null) {
            ConcurrentHashMap<Integer, MessageStore> tmpTopicMap =
                    new ConcurrentHashMap<>();
            dataMap = this.dataStores.putIfAbsent(topic, tmpTopicMap);
            if (dataMap == null) {
                dataMap = tmpTopicMap;
            }
        }
        MessageStore messageStore = dataMap.get(storeId);
        if (messageStore == null) {
            synchronized (dataStoreToken.intern()) {
                messageStore = dataMap.get(storeId);
                if (messageStore == null) {
                    TopicMetadata topicMetadata =
                            metadataManager.getTopicMetadata(topic);
                    MessageStore tmpMessageStore =
                            new MessageStore(this, topicMetadata, storeId,
                                    tubeConfig, 0, maxMsgTransferSize);
                    messageStore = dataMap.putIfAbsent(storeId, tmpMessageStore);
                    if (messageStore == null) {
                        messageStore = tmpMessageStore;
                        logger.info(sBuilder
                                .append("[Store Manager] Created a new message storage, storeKey=")
                                .append(topic).append("-").append(storeId).toString());
                    } else {
                        tmpMessageStore.close();
                    }
                }
            }
        }
        return messageStore;
    }
Get or create message store.

@param topic           the topic name
@param partition       the partition id
@return                the message-store instance
@throws IOException    the exception during processing
inlong-tubemq/tubemq-server/src/main/java/org/apache/inlong/tubemq/server/broker/msgstore/MessageStoreManager.java