private void createTopic(InlongPulsarInfo pulsarInfo, PulsarClusterInfo pulsarCluster, String topicName)
            throws Exception {
        String tenant = pulsarInfo.getPulsarTenant();
        if (StringUtils.isBlank(tenant)) {
            tenant = pulsarCluster.getPulsarTenant();
        }
        String namespace = pulsarInfo.getMqResource();
        PulsarTopicInfo topicInfo = PulsarTopicInfo.builder()
                .pulsarTenant(tenant)
                .namespace(namespace)
                .topicName(topicName)
                .queueModule(pulsarInfo.getQueueModule())
                .numPartitions(pulsarInfo.getPartitionNum())
                .build();
        pulsarOperator.createTopic(pulsarCluster, topicInfo);
    }
Create Pulsar Topic and Subscription, and save the consumer group info.
inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/resource/queue/pulsar/PulsarQueueResourceOperator.java