public void replay(TopicRecord record) {
        Uuid existingUuid = topicsByName.put(record.name(), record.topicId());
        if (existingUuid != null) {
            // We don't currently support sending a second TopicRecord for the same topic name...
            // unless, of course, there is a RemoveTopicRecord in between.
            if (existingUuid.equals(record.topicId())) {
                throw new RuntimeException("Found duplicate TopicRecord for " + record.name() +
                        " with topic ID " + record.topicId());
            } else {
                throw new RuntimeException("Found duplicate TopicRecord for " + record.name() +
                        " with a different ID than before. Previous ID was " + existingUuid +
                        " and new ID is " + record.topicId());
            }
        }
        if (Topic.hasCollisionChars(record.name())) {
            String normalizedName = Topic.unifyCollisionChars(record.name());
            TimelineHashSet<String> topicNames = topicsWithCollisionChars.get(normalizedName);
            if (topicNames == null) {
                topicNames = new TimelineHashSet<>(snapshotRegistry, 1);
                topicsWithCollisionChars.put(normalizedName, topicNames);
            }
            topicNames.add(record.name());
        }
        topics.put(record.topicId(),
            new TopicControlInfo(record.name(), snapshotRegistry, record.topicId()));
        log.info("Replayed TopicRecord for topic {} with topic ID {}.", record.name(), record.topicId());
    }
Timer for partition reassignment elect timeout.
metadata/src/main/java/org/apache/kafka/controller/ReplicationControlManager.java