void handleTopicsDelta(
        Function<Uuid, String> deletedTopicNameResolver,
        TopicsImage topicsImage,
        TopicsDelta topicsDelta,
        KRaftMigrationOperationConsumer operationConsumer
    ) {
        topicsDelta.deletedTopicIds().forEach(topicId -> {
            String name = deletedTopicNameResolver.apply(topicId);
            operationConsumer.accept(DELETE_TOPIC, "Deleted topic " + name + ", ID " + topicId,
                migrationState -> migrationClient.topicClient().deleteTopic(name, migrationState));
        });

        topicsDelta.changedTopics().forEach((topicId, topicDelta) -> {
            if (topicsDelta.createdTopicIds().contains(topicId)) {
                operationConsumer.accept(
                    CREATE_TOPIC,
                    "Created Topic " + topicDelta.name() + ", ID " + topicId,
                    migrationState -> migrationClient.topicClient().createTopic(
                        topicDelta.name(),
                        topicId,
                        topicDelta.partitionChanges(),
                        migrationState));
            } else {
                if (topicDelta.hasPartitionsWithAssignmentChanges())
                    operationConsumer.accept(
                        UPDATE_TOPIC,
                        "Updated Topic " + topicDelta.name() + ", ID " + topicId,
                        migrationState -> migrationClient.topicClient().updateTopic(
                            topicDelta.name(),
                            topicId,
                            topicsImage.getTopic(topicId).partitions(),
                            migrationState));
                Map<Integer, PartitionRegistration> newPartitions = new HashMap<>(topicDelta.newPartitions());
                Map<Integer, PartitionRegistration> changedPartitions = new HashMap<>(topicDelta.partitionChanges());
                if (!newPartitions.isEmpty()) {
                    operationConsumer.accept(
                        UPDATE_PARTITION,
                        "Created new partitions for Topic " + topicDelta.name() + ", ID " + topicId,
                        migrationState -> migrationClient.topicClient().createTopicPartitions(
                            Collections.singletonMap(topicDelta.name(), newPartitions),
                            migrationState));
                    newPartitions.keySet().forEach(changedPartitions::remove);
                }
                if (!changedPartitions.isEmpty()) {
                    // Need a final for the lambda
                    final Map<Integer, PartitionRegistration> finalChangedPartitions = changedPartitions;
                    operationConsumer.accept(
                        UPDATE_PARTITION,
                        "Updated Partitions for Topic " + topicDelta.name() + ", ID " + topicId,
                        migrationState -> migrationClient.topicClient().updateTopicPartitions(
                            Collections.singletonMap(topicDelta.name(), finalChangedPartitions),
                            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