public static void createSubscription(RestTemplate restTemplate, PulsarClusterInfo clusterInfo, String topicPath,
            String subscription) throws Exception {
        JsonObject jsonObject = new JsonObject();
        jsonObject.addProperty("entryId", Long.MAX_VALUE);
        jsonObject.addProperty("ledgerId", Long.MAX_VALUE);
        jsonObject.addProperty("partitionIndex", -1);
        HttpHeaders headers = getHttpHeaders(clusterInfo.getToken());
        MediaType type = MediaType.parseMediaType("application/json; charset=UTF-8");
        headers.setContentType(type);
        headers.add("Accept", MediaType.APPLICATION_JSON.toString());
        HttpUtils.request(restTemplate,
                clusterInfo.getAdminUrls(QUERY_PERSISTENT_PATH + "/" + topicPath + "/subscription/"
                        + subscription),
                HttpMethod.PUT, jsonObject.toString(), headers);
    }
Create a subscription on the topic.

@param restTemplate spring framework RestTemplate
@param clusterInfo pulsar cluster info
@param topicPath pulsar topic path
@param subscription pulsar topic subscription info
@throws Exception any exception if occurred
inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/resource/queue/pulsar/PulsarUtils.java