public ControllerResult<Void> cleanupDeletedTopics0() {
        long lastStableOffset = quorumController.lastStableOffset();
        if (waitingCleanupTopics.isEmpty(lastStableOffset) || !lastDeletion.isDone()) {
            return ControllerResult.of(Collections.emptyList(), null);
        }
        String topicId = waitingCleanupTopics.entrySet(lastStableOffset).iterator().next().getKey().toString();
        TimelineHashMap<Long, StreamRuntimeMetadata> streams = streamControlManager.streamsMetadata();
        List<Long> streamsToDelete = new LinkedList<>();
        // check all streams are closed
        for (Map.Entry<Long, StreamRuntimeMetadata> entry : streams.entrySet(lastStableOffset)) {
            Long streamId = entry.getKey();
            StreamRuntimeMetadata metadata = entry.getValue();
            if (topicId.equals(metadata.tags().get(StreamTags.Topic.KEY))) {
                if (metadata.currentState() == StreamState.CLOSED) {
                    streamsToDelete.add(streamId);
                } else {
                    return ControllerResult.of(Collections.emptyList(), null);
                }
            }
        }
        if (streamsToDelete.isEmpty()) {
            List<ApiMessageAndVersion> records = new ArrayList<>();
            List<String> metaStreamKvList = new ArrayList<>();
            String metaStreamKvPrefix = ObjectUtils.genMetaStreamKvPrefix(topicId);
            kvControlManager.kv().forEach((k, v) -> {
                if (k.startsWith(metaStreamKvPrefix)) {
                    metaStreamKvList.add(k);
                }
            });
            records.add(new ApiMessageAndVersion(new RemoveKVRecord().setKeys(metaStreamKvList), (short) 0));
            records.add(new ApiMessageAndVersion(new RemoveKVRecord().setKeys(List.of(TopicDeletion.TOPIC_DELETION_PREFIX + topicId)), (short) 0));
            LOGGER.info("Topic clean up completed for topic {}", topicId);
            if (waitingCleanupTopics.size(lastStableOffset) > 1) {
                // there are more topics to clean up, fast trigger next round cleanup
                cleanupDeletedTopics();
            }
            return ControllerResult.of(records, null);
        }
        lastDeletion = CompletableFuture.allOf(streamsToDelete.stream().map(streamId -> {
            ControllerRequestContext ctx = new ControllerRequestContext(null, null, OptionalLong.empty());
            DeleteStreamsRequestData req = new DeleteStreamsRequestData()
                .setDeleteStreamRequests(List.of(new DeleteStreamsRequestData.DeleteStreamRequest().setStreamId(streamId).setStreamEpoch(Long.MAX_VALUE)));
            return quorumController.deleteStreams(ctx, req).exceptionally(ex -> {
                LOGGER.error("Failed to delete stream {}", streamId, ex);
                return null;
            });
        }).toArray(CompletableFuture[]::new));
        // fast trigger next round cleanup to remove the topic
        lastDeletion.thenAccept(nil -> cleanupDeletedTopics());
        return ControllerResult.of(Collections.emptyList(), null);
    }
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