@Override
    public void createQueueForGroup(InlongGroupInfo groupInfo, String operator) {
        Preconditions.expectNotNull(groupInfo, "inlong group info cannot be null");
        Preconditions.expectNotBlank(operator, ErrorCodeEnum.INVALID_PARAMETER, "operator cannot be null");

        String groupId = groupInfo.getInlongGroupId();
        String clusterTag = groupInfo.getInlongClusterTag();
        log.info("begin to create pulsar resource for groupId={}, clusterTag={}", groupId, clusterTag);

        if (!(groupInfo instanceof InlongPulsarInfo)) {
            throw new BusinessException("the mqType must be PULSAR for inlongGroupId=" + groupId);
        }

        InlongPulsarInfo pulsarInfo = (InlongPulsarInfo) groupInfo;
        String tenant = pulsarInfo.getPulsarTenant();
        // get pulsar cluster via the inlong cluster tag from the inlong group
        List<ClusterInfo> clusterInfos = clusterService.listByTagAndType(clusterTag, ClusterType.PULSAR);
        for (ClusterInfo clusterInfo : clusterInfos) {
            PulsarClusterInfo pulsarCluster = (PulsarClusterInfo) clusterInfo;
            try {
                // create pulsar tenant and namespace
                if (StringUtils.isBlank(tenant)) {
                    tenant = pulsarCluster.getPulsarTenant();
                }

                // if the group was not successful, need create tenant and namespace
                if (!Objects.equal(GroupStatus.CONFIG_SUCCESSFUL.getCode(), groupInfo.getStatus())) {
                    pulsarOperator.createTenant(pulsarCluster, tenant);
                    String namespace = groupInfo.getMqResource();
                    pulsarOperator.createNamespace(pulsarCluster, pulsarInfo, tenant, namespace);

                    log.info("success to create pulsar resource for groupId={}, tenant={}, namespace={}, cluster={}",
                            groupId, tenant, namespace, pulsarCluster);
                }
            } catch (Exception e) {
                String msg = "failed to create pulsar resource for groupId=" + groupId;
                log.error(msg + ", cluster=" + pulsarCluster, e);
                throw new WorkflowListenerException(msg + ": " + e.getMessage());
            }
        }

        log.info("success to create pulsar resource for groupId={}, clusterTag={}", groupId, clusterTag);
    }
The name rule for Pulsar subscription: clusterTag_topicName_sinkId_consumer_group
inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/resource/queue/pulsar/PulsarQueueResourceOperator.java