public boolean topicExists(PulsarClusterInfo pulsarClusterInfo, String tenant, String namespace, String topicName,
            boolean isPartitioned) {
        if (StringUtils.isBlank(topicName)) {
            return true;
        }

        // persistent://tenant/namespace/topic
        List<String> topics;
        boolean topicExists = false;
        try {
            if (isPartitioned) {
                topics = PulsarUtils.getPartitionedTopics(restTemplate, pulsarClusterInfo, tenant,
                        namespace);
            } else {
                topics = PulsarUtils.getTopics(restTemplate, pulsarClusterInfo, tenant, namespace);
            }
            for (String t : topics) {
                t = t.substring(t.lastIndexOf("/") + 1); // not contains /
                if (!isPartitioned) {
                    int suffixIndex = t.lastIndexOf("-partition-");
                    if (suffixIndex > 0) {
                        t = t.substring(0, suffixIndex);
                    }
                }
                if (topicName.equals(t)) {
                    topicExists = true;
                    break;
                }
            }
        } catch (Exception pe) {
            LOGGER.error("check if the pulsar topic={} exists error, begin retry", topicName, pe);
            int count = 0;
            try {
                while (!topicExists && ++count <= RETRY_TIMES) {
                    LOGGER.info("check whether the pulsar topic={} exists error, try count={}", topicName, count);
                    Thread.sleep(DELAY_SECONDS);

                    topics = PulsarUtils.getPartitionedTopics(restTemplate, pulsarClusterInfo,
                            tenant, namespace);
                    for (String t : topics) {
                        t = t.substring(t.lastIndexOf("/") + 1);
                        if (topicName.equals(t)) {
                            topicExists = true;
                            break;
                        }
                    }
                }
            } catch (Exception e) {
                LOGGER.error("after retry, check if the pulsar topic={} exists still error", topicName, pe);
            }
        }
        return topicExists;
    }
Verify whether the specified Topic exists under the specified Tenant/Namespace

@apiNote cannot compare whether the string contains, otherwise it may be misjudged, such as:
        Topic "ab" does not exist, but if "abc" exists, "ab" will be mistakenly judged to exist
inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/resource/queue/pulsar/PulsarOperator.java