@Override
    public List<BriefMQMessage> queryLatestMessages(InlongGroupInfo groupInfo, InlongStreamInfo streamInfo,
            QueryMessageRequest request) {
        ClusterInfo clusterInfo = clusterService.getOne(groupInfo.getInlongClusterTag(), null, ClusterType.KAFKA);

        String topicName = streamInfo.getMqResource();
        if (topicName.equals(streamInfo.getInlongStreamId())) {
            // 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());
        }

        String consumeGroup =
                String.format(KAFKA_CONSUMER_GROUP_REALTIME_REVIEW, groupInfo.getInlongClusterTag(), topicName);
        return kafkaOperator.queryLatestMessage((KafkaClusterInfo) clusterInfo, topicName, consumeGroup, streamInfo,
                request);
    }
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