@Override
    public String getSortConsumeGroup(InlongGroupInfo groupInfo, InlongStreamEntity streamEntity,
            StreamSinkEntity sinkEntity) {
        InlongKafkaInfo kafkaInfo = (InlongKafkaInfo) groupInfo;
        String topicName = streamEntity.getMqResource();
        return String.format(KAFKA_CONSUMER_GROUP, kafkaInfo.getInlongClusterTag(), topicName);
    }
Delete Kafka Topic and Subscription, and delete the consumer group info.
inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/resource/queue/kafka/KafkaQueueResourceOperator.java