Kafka Mongo Consumer

mongo_consumer.rb

#!/usr/bin/env ruby

require 'poseidon'
require 'mongo'

Mongo::Logger.logger.level = Logger::WARN

db = Mongo::Client.new(['127.0.0.1:27017'], database: 'metrics')

consumer = Poseidon::PartitionConsumer.new("mongo_consumer", "localhost", 9092, "metrics", 0, :earliest_offset)

loop do
  messages = consumer.fetch
  messages.each do |m|
    result = db[:metrics].insert_one({ metric: m.value })
    if result.n == 1
      puts "MONGO --> Inserted metric: #{m.value}"
    end
  end
end