@Override
    public void deleteQueueForStream(InlongGroupInfo groupInfo, InlongStreamInfo streamInfo, String operator) {
        Preconditions.expectNotNull(groupInfo, "inlong group info cannot be null");
        Preconditions.expectNotNull(streamInfo, "inlong stream info cannot be null");

        String groupId = streamInfo.getInlongGroupId();
        String streamId = streamInfo.getInlongStreamId();
        log.info("begin to delete kafka resource for groupId={} streamId={}", groupId, streamId);

        try {
            String topicName = streamInfo.getMqResource();
            if (StringUtils.isBlank(topicName) || topicName.equals(streamId)) {
                // the default mq resource (stream id) is not sufficient to discriminate different kafka topics
                topicName = String.format(Constants.DEFAULT_KAFKA_TOPIC_FORMAT,
                        groupInfo.getMqResource(), streamInfo.getMqResource());
            }
            this.deleteKafkaTopic(groupInfo, topicName);
            log.info("success to delete kafka topic for groupId={}, streamId={}", groupId, streamId);
        } catch (Exception e) {
            String msg = String.format("failed to delete kafka topic for groupId=%s, streamId=%s", groupId, streamId);
            log.error(msg, e);
            throw new WorkflowListenerException(msg);
        }
        log.info("success to delete kafka resource for groupId={}, streamId={}", groupId, streamId);
    }
The name rule for Kafka consumer group: clusterTag_topicName_consumer_group
inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/resource/queue/kafka/KafkaQueueResourceOperator.java