void deleteTopic(ControllerRequestContext context, Uuid id, List<ApiMessageAndVersion> records) {
        TopicControlInfo topic = topics.get(id);
        if (topic == null) {
            throw new UnknownTopicIdException(UNKNOWN_TOPIC_ID.message());
        }
        int numPartitions = topic.parts.size();
        try {
            context.applyPartitionChangeQuota(numPartitions); // check controller mutation quota
        } catch (ThrottlingQuotaExceededException e) {
            // log a message and rethrow the exception
            log.debug("Topic deletion of {} partitions not allowed because quota is violated. Delay time: {}",
                numPartitions, e.throttleTimeMs());
            throw e;
        }
        records.add(new ApiMessageAndVersion(new RemoveTopicRecord().
            setTopicId(id), (short) 0));

        // AutoMQ inject start
        if (featureControl.autoMQVersion().isTopicCleanupByControllerSupported()) {
            records.add(new ApiMessageAndVersion(
                new KVRecord().setKeyValues(
                    List.of(new KVRecord.KeyValue()
                        .setKey(TopicDeletion.TOPIC_DELETION_PREFIX + id.toString())
                        .setValue(Integer.toString(TopicDeletion.Status.PENDING.value()).getBytes(StandardCharsets.UTF_8))
                    )
                ),
                (short) 0));
        }
        // AutoMQ inject end

    }
Timer for partition reassignment elect timeout.
metadata/src/main/java/org/apache/kafka/controller/ReplicationControlManager.java