void handleTopicsSnapshot(TopicsImage topicsImage, KRaftMigrationOperationConsumer operationConsumer) {
        Map<Uuid, String> deletedTopics = new HashMap<>();
        Set<Uuid> topicsInZk = new HashSet<>();
        Set<Uuid> newTopics = new HashSet<>(topicsImage.topicsById().keySet());
        Set<Uuid> changedTopics = new HashSet<>();
        Map<Uuid, Set<Integer>> partitionsInZk = new HashMap<>();
        Map<String, Set<Integer>> extraneousPartitionsInZk = new HashMap<>();
        Map<Uuid, Map<Integer, PartitionRegistration>> changedPartitions = new HashMap<>();
        Map<Uuid, Map<Integer, PartitionRegistration>> newPartitions = new HashMap<>();

        Set<String> pendingTopicDeletions = migrationClient.topicClient().readPendingTopicDeletions();
        if (!pendingTopicDeletions.isEmpty()) {
            operationConsumer.accept(
                DELETE_PENDING_TOPIC_DELETION,
                "Deleted pending topic deletions",
                migrationState -> migrationClient.topicClient().clearPendingTopicDeletions(pendingTopicDeletions, migrationState)
            );
        }

        migrationClient.topicClient().iterateTopics(
            EnumSet.of(
                TopicMigrationClient.TopicVisitorInterest.TOPICS,
                TopicMigrationClient.TopicVisitorInterest.PARTITIONS),
            new TopicMigrationClient.TopicVisitor() {
                @Override
                public void visitTopic(String topicName, Uuid topicId, Map<Integer, List<Integer>> assignments) {
                    TopicImage topic = topicsImage.getTopic(topicId);
                    if (topic == null) {
                        // If KRaft does not have this topic, it was deleted
                        deletedTopics.put(topicId, topicName);
                    } else {
                        if (!newTopics.remove(topicId)) return;
                        topicsInZk.add(topicId);
                    }
                }

                @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());
                        }
                    }
                }
            });

        // Check for any partition changes in existing topics.
        topicsInZk.forEach(topicId -> {
            TopicImage topic = topicsImage.getTopic(topicId);
            Set<Integer> topicPartitionsInZk = partitionsInZk.computeIfAbsent(topicId, __ -> new HashSet<>());
            if (!topicPartitionsInZk.equals(topic.partitions().keySet())) {
                Map<Integer, PartitionRegistration> newTopicPartitions = new HashMap<>(topic.partitions());
                // Compute KRaft partitions that are not in ZK
                topicPartitionsInZk.forEach(newTopicPartitions::remove);
                newPartitions.put(topicId, newTopicPartitions);

                // Compute ZK partitions that are not in KRaft
                topicPartitionsInZk.removeAll(topic.partitions().keySet());
                if (!topicPartitionsInZk.isEmpty()) {
                    extraneousPartitionsInZk.put(topic.name(), topicPartitionsInZk);
                }
                changedTopics.add(topicId);
            }
        });

        newTopics.forEach(topicId -> {
            TopicImage topic = topicsImage.getTopic(topicId);
            operationConsumer.accept(
                CREATE_TOPIC,
                "Created Topic " + topic.name() + ", ID " + topicId,
                migrationState -> migrationClient.topicClient().createTopic(topic.name(), topicId, topic.partitions(), migrationState)
            );
        });

        changedTopics.forEach(topicId -> {
            TopicImage topic = topicsImage.getTopic(topicId);
            operationConsumer.accept(
                UPDATE_TOPIC,
                "Changed Topic " + topic.name() + ", ID " + topicId,
                migrationState -> migrationClient.topicClient().updateTopic(topic.name(), topicId, topic.partitions(), migrationState)
            );
        });

        deletedTopics.forEach((topicId, topicName) ->
            operationConsumer.accept(
                DELETE_TOPIC,
                "Deleted Topic " + topicName + ", ID " + topicId,
                migrationState -> migrationClient.topicClient().deleteTopic(topicName, migrationState)
            )
        );

        newPartitions.forEach((topicId, partitionMap) -> {
            TopicImage topic = topicsImage.getTopic(topicId);
            operationConsumer.accept(
                UPDATE_PARTITION,
                "Created additional partitions for Topic " + topic.name() + ", ID " + topicId,
                migrationState -> migrationClient.topicClient().updateTopicPartitions(
                    Collections.singletonMap(topic.name(), partitionMap),
                    migrationState));
        });

        changedPartitions.forEach((topicId, partitionMap) -> {
            TopicImage topic = topicsImage.getTopic(topicId);
            operationConsumer.accept(
                UPDATE_PARTITION,
                "Updated Partitions for Topic " + topic.name() + ", ID " + topicId,
                migrationState -> migrationClient.topicClient().updateTopicPartitions(
                    Collections.singletonMap(topic.name(), partitionMap),
                    migrationState));
        });

        extraneousPartitionsInZk.forEach((topicName, partitions) ->
            operationConsumer.accept(
                DELETE_PARTITION,
                "Deleted extraneous Partitions " + partitions + " for Topic " + topicName,
                migrationState -> migrationClient.topicClient().deleteTopicPartitions(
                    Collections.singletonMap(topicName, partitions),
                    migrationState))
        );
    }
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