public void replay(RemoveKVRecord kvRecord) {
        for (String key : kvRecord.keys()) {
            if (key.startsWith(TopicDeletion.TOPIC_DELETION_PREFIX)) {
                Uuid topicId = Uuid.fromString(key.substring(TopicDeletion.TOPIC_DELETION_PREFIX.length()));
                waitingCleanupTopics.remove(topicId);
            }
        }
    }
Cleanup deleted topic workflow:
1. ReplicationControlManager#deleteTopic generate kv record with key "{@link TopicDeletion#TOPIC_DELETION_PREFIX}/{topicId}"
2. {@link TopicDeletionManager} replay the kv record and put the topicId to {@link TopicDeletionManager#waitingCleanupTopics}
3. {@link TopicDeletionManager} schedule scan the waiting cleanup topics and delete the streams related to topic
4. After the streams related to the topic are deleted, {@link TopicDeletionManager} generate RemoveKvRecord to remove the topicId from {@link TopicDeletionManager#waitingCleanupTopics}
metadata/src/main/java/org/apache/kafka/controller/stream/TopicDeletionManager.java