public void replay(PartitionRecord record) {
        TopicControlInfo topicInfo = topics.get(record.topicId());
        if (topicInfo == null) {
            throw new RuntimeException("Tried to create partition " + record.topicId() +
                ":" + record.partitionId() + ", but no topic with that ID was found.");
        }
        PartitionRegistration newPartInfo = new PartitionRegistration(record);
        PartitionRegistration prevPartInfo = topicInfo.parts.get(record.partitionId());
        String description = topicInfo.name + "-" + record.partitionId() +
            " with topic ID " + record.topicId();
        if (prevPartInfo == null) {
            log.info("Replayed PartitionRecord for new partition {} and {}.", description,
                    newPartInfo);
            topicInfo.parts.put(record.partitionId(), newPartInfo);
            updatePartitionInfo(record.topicId(), record.partitionId(), null, newPartInfo);
            updatePartitionDirectories(record.topicId(), record.partitionId(), null, newPartInfo.directories);
            updateReassigningTopicsIfNeeded(record.topicId(), record.partitionId(),
                    false,  isReassignmentInProgress(newPartInfo));
        } else if (!newPartInfo.equals(prevPartInfo)) {
            log.info("Replayed PartitionRecord for existing partition {} and {}.", description,
                    newPartInfo);
            newPartInfo.maybeLogPartitionChange(log, description, prevPartInfo);
            topicInfo.parts.put(record.partitionId(), newPartInfo);
            updatePartitionInfo(record.topicId(), record.partitionId(), prevPartInfo, newPartInfo);
            updatePartitionDirectories(record.topicId(), record.partitionId(), prevPartInfo.directories, newPartInfo.directories);
            updateReassigningTopicsIfNeeded(record.topicId(), record.partitionId(),
                    isReassignmentInProgress(prevPartInfo), isReassignmentInProgress(newPartInfo));
        }

        if (newPartInfo.hasPreferredLeader()) {
            imbalancedPartitions.remove(new TopicIdPartition(record.topicId(), record.partitionId()));
        } else {
            imbalancedPartitions.add(new TopicIdPartition(record.topicId(), record.partitionId()));
        }
    }
Timer for partition reassignment elect timeout.
metadata/src/main/java/org/apache/kafka/controller/ReplicationControlManager.java