@VisibleForTesting
    public List<BriefMQMessage> getLatestMessage(Consumer<byte[], byte[]> consumer, String topicName,
            InlongStreamInfo streamInfo, QueryMessageRequest request) {
        List<BriefMQMessage> messageList = new ArrayList<>();
        Integer messageCount = request.getMessageCount();
        try {
            List<PartitionInfo> partitionInfoList = consumer.partitionsFor(topicName);
            List<TopicPartition> topicPartitionList = partitionInfoList.stream()
                    .map(topicPartition -> new TopicPartition(topicPartition.topic(), topicPartition.partition()))
                    .collect(Collectors.toList());

            Map<TopicPartition, Long> beginningTopicPartitionList = consumer.beginningOffsets(topicPartitionList);
            Map<TopicPartition, Long> endTopicPartitionList = consumer.endOffsets(topicPartitionList);

            int count = (int) Math.ceil((double) messageCount / topicPartitionList.size());
            Map<TopicPartition, Long> expectedOffsetMap = beginningTopicPartitionList.entrySet()
                    .stream()
                    .map(entry -> {
                        long beginningOffset = entry.getValue();
                        long endOffset = endTopicPartitionList.getOrDefault(entry.getKey(), beginningOffset);
                        Long offset = (endOffset - beginningOffset) >= count ? (endOffset - count) : beginningOffset;
                        return Pair.of(entry.getKey(), offset);
                    }).collect(Collectors.toMap(Pair::getKey, Pair::getValue));

            consumer.assign(topicPartitionList);
            expectedOffsetMap.forEach(consumer::seek);

            int index = 0;
            ConsumerRecords<byte[], byte[]> records = consumer.poll(Duration.ofMillis(100));
            for (ConsumerRecord<byte[], byte[]> record : records) {
                Map<String, String> headers = new HashMap<>();
                for (Header header : record.headers()) {
                    headers.put(header.key(), new String(header.value(), StandardCharsets.UTF_8));
                }

                MessageWrapType messageWrapType = MessageWrapType.forType(streamInfo.getWrapType());
                if (headers.get(InlongConstants.MSG_ENCODE_VER) != null) {
                    messageWrapType =
                            MessageWrapType.valueOf(Integer.parseInt(headers.get(InlongConstants.MSG_ENCODE_VER)));
                }
                DeserializeOperator deserializeOperator = deserializeOperatorFactory.getInstance(messageWrapType);
                deserializeOperator.decodeMsg(streamInfo, messageList, record.value(), headers, index, request);
                if (messageList.size() >= messageCount) {
                    break;
                }
            }
        } catch (Exception e) {
            String errMsg = "decode msg error: ";
            LOGGER.error(errMsg, e);
            throw new BusinessException(errMsg + e.getMessage());
        } finally {
            consumer.close();
        }

        LOGGER.debug("success query messages for topic={}, size={}, returned size={}",
                topicName, messageList.size(), messageCount);
        // only return a list of messages of the specified count
        int fromIndex = (messageList.size() > messageCount) ? (messageList.size() - messageCount) : 0;
        List<BriefMQMessage> resultList = messageList.subList(fromIndex, messageList.size());
        for (int i = 0; i < resultList.size(); i++) {
            BriefMQMessage message = resultList.get(i);
            message.setId(i + 1);
        }

        return resultList;
    }
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