@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();
        log.info("begin to create pulsar resource for groupId={}", groupId);

        // if the group was successful, no need re-create topic and consumer group
        if (Objects.equal(GroupStatus.CONFIG_SUCCESSFUL.getCode(), groupInfo.getStatus())) {
            log.info("skip to create tubemq resource as the status of groupId={} was successful", groupId);
        }

        try {
            // 1. create tubemq topic
            String clusterTag = groupInfo.getInlongClusterTag();
            TubeClusterInfo tubeCluster = (TubeClusterInfo) clusterService.getOne(clusterTag, null, ClusterType.TUBEMQ);
            String topicName = groupInfo.getMqResource();
            tubeMQOperator.createTopic(tubeCluster, topicName, operator);
            log.info("success to create tubemq topic for groupId={}, cluster={}", groupId, tubeCluster);
        } catch (Exception e) {
            log.error("failed to create tubemq resource for groupId=" + groupId, e);
            throw new WorkflowListenerException("failed to create tubemq resource: " + e.getMessage());
        }
    }
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/tubemq/TubeMQQueueResourceOperator.java