public List<BriefMQMessage> queryLatestMessage(PulsarClusterInfo pulsarClusterInfo, String topicFullName,
            QueryMessageRequest request, InlongStreamInfo streamInfo, boolean serial) {
        LOGGER.info("begin to query message for topic {}, adminUrl={}", topicFullName, pulsarClusterInfo.getAdminUrl());
        List<BriefMQMessage> messageList = new ArrayList<>();
        int partitionCount = getPartitionCount(pulsarClusterInfo, topicFullName);
        for (int messageIndex = 0; messageIndex < request.getMessageCount(); messageIndex++) {
            int currentPartitionNum = messageIndex % partitionCount;
            int messagePosition = messageIndex / partitionCount + 1;
            String topicNameOfPartition = buildTopicNameOfPartition(topicFullName, currentPartitionNum, serial);
            messageList.addAll(queryMessageFromPulsar(topicNameOfPartition, pulsarClusterInfo, messageIndex, streamInfo,
                    messagePosition, request));
        }
        LOGGER.info("success query message for topic={}", topicFullName);
        return messageList;
    }
Query topic message for the given pulsar cluster.
inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/resource/queue/pulsar/PulsarOperator.java