private boolean subscriptionExists(PulsarClusterInfo pulsarClusterInfo, String topic, String subscription,
            boolean isPartitioned) {
        int count = 0;
        while (++count <= RETRY_TIMES) {
            try {
                LOGGER.info("check whether the subscription exists for topic={}, try count={}", topic, count);
                Thread.sleep(DELAY_SECONDS);

                // first lookup to load the topic, and then query whether the subscription exists
                if (isPartitioned) {
                    Map<String, String> topicMap = PulsarUtils.lookupPartitionedTopic(restTemplate,
                            pulsarClusterInfo, topic);
                    if (topicMap.isEmpty()) {
                        LOGGER.error("result of lookups topic={} is empty, continue retry", topic);
                        continue;
                    }
                } else {
                    String lookupTopic = PulsarUtils.lookupTopic(restTemplate, pulsarClusterInfo, topic);
                    if (StringUtils.isBlank(lookupTopic)) {
                        LOGGER.error("result of lookups topic={} is empty, continue retry", topic);
                        continue;
                    }
                }

                List<String> subscriptionList = PulsarUtils.getSubscriptions(restTemplate,
                        pulsarClusterInfo, topic);
                return subscriptionList.contains(subscription);
            } catch (Exception e) {
                LOGGER.error("check if the subscription exists for topic={} error, continue retry", topic, e);
                if (count == RETRY_TIMES) {
                    LOGGER.error("after {} times retry, still check subscription exception for topic {}", count, topic);
                    throw new BusinessException("check if the subscription exists error: " + e.getMessage());
                }
            }
        }
        return false;
    }
Check whether the Pulsar topic exists.
inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/resource/queue/pulsar/PulsarOperator.java