@Override
    public void createQueueForStream(InlongGroupInfo groupInfo, InlongStreamInfo streamInfo, String operator) {
        String groupId = groupInfo.getInlongGroupId();
        String streamId = streamInfo.getInlongStreamId();
        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;
        }
        for (StreamSink sink : streamSinks) {
            // 1. create tubemq consumer group
            // consumer naming rules: clusterTag_topicName_sinkId_consumer_group
            String clusterTag = groupInfo.getInlongClusterTag();
            TubeClusterInfo tubeCluster = (TubeClusterInfo) clusterService.getOne(clusterTag, null, ClusterType.TUBEMQ);
            String topicName = groupInfo.getMqResource();
            String consumeGroup = String.format(TUBE_CONSUMER_GROUP, clusterTag, topicName, sink.getId());
            tubeMQOperator.createConsumerGroup(tubeCluster, topicName, consumeGroup, operator);
            log.info("success to create tubemq consumer group for groupId={}", groupId);

            // insert the consumer group info
            Integer id = consumeService.saveBySystem(groupInfo, topicName, consumeGroup);
            log.info("success to save inlong consume [{}] for consumerGroup={}, groupId={}, topic={}",
                    id, consumeGroup, groupId, topicName);
        }
    }
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