Only consume the last message once

The simple consumer repeatedly consumed the last message in a partition.

Author: dasch

examples/simple-consumer.rb

@@ -27,22 +27,6 @@
   logger: logger,
 )
 
-begin
-  offset = :latest
-  partition = 0
-
-  loop do
-    messages = kafka.fetch_messages(
-      topic: topic,
-      partition: partition,
-      offset: offset
-    )
-
-    messages.each do |message|
-      puts message.value
-      offset = message.offset + 1
-    end
-  end
-ensure
-  kafka.close
+kafka.each_message(topic: topic) do |message|
+  puts message.value
 end

lib/kafka/client.rb

@@ -357,7 +357,7 @@ def each_message(topic:, start_from_beginning: true, max_wait_time: 5, min_bytes
 
         batches.each do |batch|
           batch.messages.each(&block)
-          offsets[batch.partition] = batch.last_offset
+          offsets[batch.partition] = batch.last_offset + 1
         end
       end
     end