private boolean isTopicExists(Admin admin, String topicName, Integer partitionNum) throws Exception {
        ListTopicsResult listResult = admin.listTopics();
        if (!listResult.namesToListings().get().containsKey(topicName)) {
            LOGGER.info("kafka topic {} not existed", topicName);
            return false;
        }

        DescribeTopicsResult result = admin.describeTopics(Collections.singletonList(topicName));
        TopicDescription desc = result.values().get(topicName).get();
        String info = "kafka topic=%s already exist with partition num=%s";
        if (desc.partitions().size() != partitionNum) {
            String errMsg = String.format(info + ", but the requested partition num=%s", topicName,
                    desc.partitions().size(), partitionNum);
            LOGGER.error(errMsg);
            throw new IllegalArgumentException(errMsg);
        } else {
            LOGGER.info(String.format(info + ", no need to create", topicName, partitionNum));
            return true;
        }
    }
Check whether the topic exists in the Kafka MQ
inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/resource/sink/kafka/KafkaResourceOperator.java