public List<BriefMQMessage> queryLatestMessage(KafkaClusterInfo clusterInfo, String topicName,
            String consumeGroup, InlongStreamInfo streamInfo, QueryMessageRequest request) {
        LOGGER.debug("begin to query message for topic {} in cluster: {}", topicName, clusterInfo);

        Properties properties = getProperties(clusterInfo.getUrl(), consumeGroup);
        KafkaConsumer<byte[], byte[]> consumer = new KafkaConsumer<>(properties);
        return getLatestMessage(consumer, topicName, streamInfo, request);
    }
Query topic message for the given Kafka cluster.
inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/resource/queue/kafka/KafkaOperator.java