@Override
                public void visitPartition(TopicIdPartition topicIdPartition, PartitionRegistration partitionRegistration) {
                    TopicImage topic = topicsImage.getTopic(topicIdPartition.topicId());
                    if (topic == null) {
                        return; // The topic was deleted in KRaft. Handled by deletedTopics
                    }

                    // If there is failure in previous Zk writes, We could end up with Zookeeper
                    // containing with partial or without any partitions for existing topics. So
                    // accumulate the partition ids to check for any missing partitions in Zk.
                    partitionsInZk
                        .computeIfAbsent(topic.id(), __ -> new HashSet<>())
                        .add(topicIdPartition.partition());

                    // Check if the KRaft partition state changed
                    PartitionRegistration kraftPartition = topic.partitions().get(topicIdPartition.partition());
                    if (kraftPartition != null) {
                        if (!kraftPartition.equals(partitionRegistration)) {
                            changedPartitions.computeIfAbsent(topicIdPartition.topicId(), __ -> new HashMap<>())
                                .put(topicIdPartition.partition(), kraftPartition);
                        }

                        // Check if partition assignment has changed. This will need topic update.
                        if (!kraftPartition.hasSameAssignment(partitionRegistration)) {
                            changedTopics.add(topic.id());
                        }
                    }
                }
Handle a snapshot of the topic metadata. This requires scanning through all the topics and partitions
in ZooKeeper to determine what has changed. Topic configs are not handled here since they exist in the
ConfigurationsImage.
metadata/src/main/java/org/apache/kafka/metadata/migration/KRaftMigrationZkWriter.java