simple_consumer.rb

# When no group id is passed there's no coordination.
consumer = kafka.consumer
consumer.subscribe("topic1")

# You can seek separately for each partition.
consumer.seek("topic1", partition: 0, offset: 42)
consumer.seek("topic1", partition: 1, offset: 13)

# `poll` returns a list of batches -- useful for e.g. checking against
# the highwater mark.
batches = consumer.poll(timeout: 5, max_bytes_per_partition: 1024 * 5)

batches.each do |batch|
  lag = batch.highwater_mark_offset - batch.messages.first.offset
  logger.info "Consumer lag for #{batch.topic}/#{batch.partition}: #{lag}"
end

# This is simple enough that a separate message list based API may not
# be needed.
messages = batches.flat_map {|batch| batch.messages }