void handleProducerIdSnapshot(ProducerIdsImage image, KRaftMigrationOperationConsumer operationConsumer) {
        if (image.isEmpty()) {
            // No producer IDs have been allocated, nothing to dual-write
            return;
        }
        Optional<ProducerIdsBlock> zkProducerId = migrationClient.readProducerId();
        if (zkProducerId.isPresent()) {
            if (zkProducerId.get().nextBlockFirstId() != image.nextProducerId()) {
                operationConsumer.accept(UPDATE_PRODUCER_ID, "Set next producer ID", migrationState ->
                    migrationClient.writeProducerId(image.nextProducerId(), migrationState));
            }
        } else {
            operationConsumer.accept(UPDATE_PRODUCER_ID, "Set next producer ID", migrationState ->
                migrationClient.writeProducerId(image.nextProducerId(), 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