public void createConsumerGroup(TubeClusterInfo tubeCluster, String topic, String consumerGroup, String operator) {
        String masterUrl = tubeCluster.getMasterWebUrl();
        LOGGER.info("begin to create consumer group {} for topic {} in master {}", consumerGroup, topic, masterUrl);
        if (StringUtils.isEmpty(masterUrl) || StringUtils.isEmpty(consumerGroup) || StringUtils.isEmpty(topic)) {
            throw new BusinessException("TubeMQ master url, consumer group, or TubeMQ topic cannot be null");
        }

        if (!this.isTopicExist(masterUrl, topic)) {
            LOGGER.warn("cannot create TubeMQ consumer group {}, as the topic {} not exists in master {}",
                    consumerGroup, topic, masterUrl);
            return;
        }

        if (this.isConsumerGroupExist(masterUrl, topic, consumerGroup)) {
            LOGGER.warn("TubeMQ consumer group {} already exists for topic {} in master {}, skip to create",
                    consumerGroup, topic, masterUrl);
            return;
        }

        this.createConsumerGroupOpt(masterUrl, topic, consumerGroup, tubeCluster.getToken(), operator);
        LOGGER.info("success to create TubeMQ consumer group {} for topic {} in {}", consumerGroup, topic, masterUrl);
    }
Create consumer group for the given tubemq topic and cluster.
inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/resource/queue/tubemq/TubeMQOperator.java