private void createSubscription(InlongPulsarInfo pulsarInfo, PulsarClusterInfo pulsarCluster, String topicName,
            String streamId) throws Exception {
        String tenant = pulsarInfo.getPulsarTenant();
        if (StringUtils.isBlank(tenant)) {
            tenant = pulsarCluster.getPulsarTenant();
        }
        String namespace = pulsarInfo.getMqResource();
        String fullTopicName = tenant + "/" + namespace + "/" + topicName;
        boolean exist = pulsarOperator.topicExists(pulsarCluster, tenant, namespace, topicName,
                InlongConstants.PULSAR_QUEUE_TYPE_PARALLEL.equals(pulsarInfo.getQueueModule()));
        if (!exist) {
            String serviceUrl = pulsarCluster.getAdminUrl();
            log.error("topic={} not exists in {}", fullTopicName, serviceUrl);
            throw new WorkflowListenerException("topic=" + fullTopicName + " not exists in " + serviceUrl);
        }

        // create subscription for all sinks
        String groupId = pulsarInfo.getInlongGroupId();
        List<StreamSink> streamSinks = sinkService.listSink(groupId, streamId);
        if (CollectionUtils.isEmpty(streamSinks)) {
            log.warn("no need to create subs, as no sink exists for groupId={}, streamId={}", groupId, streamId);
            return;
        }

        // subscription naming rules: clusterTag_topicName_sinkId_consumer_group
        String clusterTag = pulsarInfo.getInlongClusterTag();
        for (StreamSink sink : streamSinks) {
            String subs = String.format(PULSAR_SUBSCRIPTION, clusterTag, topicName, sink.getId());
            pulsarOperator.createSubscription(pulsarCluster, fullTopicName, pulsarInfo.getQueueModule(), subs);
            log.info("success to create subs={} for groupId={}, topic={}", subs, groupId, fullTopicName);

            // insert the consumer group info into the inlong_consume table
            Integer id = consumeService.saveBySystem(pulsarInfo, topicName, subs);
            log.info("success to save inlong consume [{}] for subs={}, groupId={}, topic={}",
                    id, subs, groupId, topicName);
        }
    }
Create Pulsar Subscription, and save the consumer group info.
inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/resource/queue/pulsar/PulsarQueueResourceOperator.java