public static ResponseEntity<byte[]> examineMessage(RestTemplate restTemplate, PulsarClusterInfo clusterInfo,
            String topicPartition, String messageType, int messagePosition) throws Exception {
        String adminUrl = clusterInfo.getAdminUrl();
        String[] adminUrls = adminUrl.replace(HTTP_PREFIX, InlongConstants.EMPTY).split(InlongConstants.COMMA);
        for (int i = 0; i < adminUrls.length; i++) {
            try {
                StringBuilder urlBuilder = new StringBuilder().append(HTTP_PREFIX + adminUrls[i])
                        .append(QUERY_PERSISTENT_PATH)
                        .append("/")
                        .append(topicPartition)
                        .append("/examinemessage")
                        .append("?initialPosition=")
                        .append(messageType)
                        .append("&messagePosition=")
                        .append(messagePosition);
                ResponseEntity<byte[]> response = restTemplate.exchange(urlBuilder.toString(), HttpMethod.GET,
                        new HttpEntity<>(getHttpHeaders(clusterInfo.getToken())), byte[].class);
                if (!response.getStatusCode().is2xxSuccessful()) {
                    log.error("request error for {}, status code {}, body {}", urlBuilder.toString(),
                            response.getStatusCode(),
                            response.getBody());
                }
                return response;
            } catch (Exception e) {
                log.error("examine message for topic partition={} error, begin retry", topicPartition, e);
                if (i >= (adminUrls.length - 1)) {
                    log.error("after retry, examine message for topic partition={} still error", topicPartition, e);
                    throw e;
                }
            }
        }
        throw new Exception(String.format("examine message failed for topic partition=%s", topicPartition));
    }
Examine a specific message on a topic by position relative to the earliest or the latest message.

@param restTemplate spring framework RestTemplate
@param clusterInfo pulsar cluster info
@param topicPartition pulsar topic partition info
@param messageType pulsar message type info
@param messagePosition pulsar message position info
@return spring framework HttpEntity
@throws Exception any exception if occurred
inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/resource/queue/pulsar/PulsarUtils.java