public void createSubscription(PulsarClusterInfo pulsarClusterInfo, String fullTopicName, String queueModule,
            String subscription) throws Exception {
        LOGGER.info("begin to create pulsar subscription={} for topic={}", subscription, fullTopicName);
        try {
            boolean isExists = this.subscriptionExists(pulsarClusterInfo, fullTopicName, subscription,
                    InlongConstants.PULSAR_QUEUE_TYPE_PARALLEL.equals(queueModule));
            if (isExists) {
                LOGGER.warn("pulsar subscription={} already exists, skip to create", subscription);
                return;
            }

            PulsarUtils.createSubscription(restTemplate, pulsarClusterInfo, fullTopicName, subscription);
            LOGGER.info("success to create subscription={}", subscription);
        } catch (Exception e) {
            LOGGER.error("failed to create pulsar subscription=" + subscription, e);
            throw e;
        }
    }
Create a Pulsar subscription for the given topic
inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/resource/queue/pulsar/PulsarOperator.java