persistor.rb

require 'kafka'
kafka = Kafka.new(
  seed_brokers: ["0.0.0.0:9092"],
  client_id: "miq-persistor"
)

consumer = kafka.consumer(group_id: "miq-persisters")
consumer.subscribe("inventory")

consumer.each_message do |message|
  puts "Received inventory"
  persister = ManagerRefresh::Inventory::Persister.from_yaml(message.value)
  
  puts "Saving inventory"
  ManagerRefresh::SaveInventory.save_inventory(persister.manager, persister.inventory_collections)
  puts "Saving inventory...Complete"
end