public List<BriefMQMessage> queryLastMessage(TubeClusterInfo tubeCluster, String topicName,
            InlongStreamInfo streamInfo, QueryMessageRequest request) {
        LOGGER.info("begin to query message for topic {} in cluster: {}", topicName, tubeCluster);
        String masterUrl = tubeCluster.getMasterWebUrl();
        TubeBrokerInfo brokerView = this.getBrokerInfo(masterUrl);
        String brokerUrl = brokerView.getOnlineBrokerAddress();

        List<BriefMQMessage> messageList = new ArrayList<>();
        try {
            if (StringUtils.isEmpty(brokerUrl) || StringUtils.isEmpty(topicName)) {
                throw new BusinessException("TubeMQ master url or TubeMQ topic cannot be null");
            }

            if (!this.isTopicExist(masterUrl, topicName)) {
                LOGGER.error("TubeMQ topic {} not exists in {}, skip to query", topicName, masterUrl);
                throw new BusinessException("TubeMQ master url or TubeMQ topic cannot be null");
            }

            String url = "http://" + brokerUrl + QUERY_MESSAGE_PATH + TOPIC_NAME + topicName + MSG_COUNT
                    + request.getMessageCount() + FILTER_CONDS + streamInfo.getInlongStreamId();
            TubeMessageResponse response = HttpUtils.request(restTemplate, url, HttpMethod.GET,
                    null, new HttpHeaders(), TubeMessageResponse.class);
            if (response.getErrCode() != SUCCESS_CODE && response.getErrCode() != 200) {
                String msg = String.format("failed to query message for topic %s, error: %s",
                        topicName, response.getErrMsg());
                LOGGER.error(msg + " in {} for broker {}", masterUrl, brokerUrl);
                throw new BusinessException(msg);
            }

            int index = 0;
            for (TubeDataInfo tubeDataInfo : response.getDataSet()) {
                Map<String, String> map = new HashMap<>();
                for (String kv : tubeDataInfo.getAttr().split(InlongConstants.COMMA)) {
                    map.put(kv.split(InlongConstants.EQUAL)[0], kv.split(InlongConstants.EQUAL)[1]);
                }

                MessageWrapType messageWrapType = MessageWrapType.forType(streamInfo.getWrapType());
                if (map.get(InlongConstants.MSG_ENCODE_VER) != null) {
                    messageWrapType =
                            MessageWrapType.valueOf(Integer.parseInt(map.get(InlongConstants.MSG_ENCODE_VER)));
                }
                byte[] messageData = Base64.getDecoder().decode(tubeDataInfo.getData());
                DeserializeOperator deserializeOperator = deserializeOperatorFactory.getInstance(messageWrapType);
                deserializeOperator.decodeMsg(streamInfo, messageList, messageData, map, index, request);
            }

            LOGGER.info("success query messages for topic={}", topicName);
        } catch (Exception e) {
            String errMsg = "failed to query messages: ";
            LOGGER.error(errMsg, e);
            throw new BusinessException(errMsg + e.getMessage());
        }

        return messageList;
    }
Query topic message for the given tubemq cluster.
inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/resource/queue/tubemq/TubeMQOperator.java