public GetMessageResult getMessages(final MessageStore msgStore,
            final String topic,
            final int partitionId,
            final int msgCount,
            final Set<String> filterCondSet) throws IOException {
        long requestOffset = 0L;
        try {
            final long maxOffset = msgStore.getIndexMaxOffset();
            ConsumerNodeInfo consumerNodeInfo =
                    new ConsumerNodeInfo(tubeBroker.getStoreManager(), "visit",
                            "visit", filterCondSet, "", System.currentTimeMillis(), "", "");
            int maxIndexReadSize = (msgCount + 1)
                    * DataStoreUtils.STORE_INDEX_HEAD_LEN * msgStore.getPartitionNum();
            if (filterCondSet != null && !filterCondSet.isEmpty()) {
                maxIndexReadSize *= 5;
            }
            requestOffset = maxOffset - maxIndexReadSize < 0 ? 0L : maxOffset - maxIndexReadSize;
            return msgStore.getMessages(303, requestOffset, partitionId,
                    consumerNodeInfo, topic, this.maxMsgTransferSize, 0);
        } catch (Throwable e1) {
            return new GetMessageResult(false, TErrCodeConstants.INTERNAL_SERVER_ERROR,
                    requestOffset, 0, "Get message failure, errMsg=" + e1.getMessage());
        }
    }
Get message from store.

@param msgStore        the message-store
@param topic           the topic name
@param partitionId     the partition id
@param msgCount        the message count to read
@param filterCondSet   the filter condition set
@return                the query result
@throws IOException    the exception during processing
inlong-tubemq/tubemq-server/src/main/java/org/apache/inlong/tubemq/server/broker/msgstore/MessageStoreManager.java