Merge pull request #443 from jturkel/feature/offset-retention

Commit offsets even if we don't read any new messages

Author: dasch

lib/kafka/consumer.rb

@@ -216,6 +216,8 @@ def each_message(min_bytes: 1, max_wait_time: 1, automatically_mark_as_processed
 
         # We may not have received any messages, but it's still a good idea to
         # commit offsets if we've processed messages in the last set of batches.
+        # This also ensures the offsets are retained if we haven't read any messages
+        # since the offset retention period has elapsed.
         @offset_manager.commit_offsets_if_necessary
       end
     end
@@ -279,6 +281,12 @@ def each_batch(min_bytes: 1, max_wait_time: 1, automatically_mark_as_processed:
 
           return if !@running
         end
+
+        # We may not have received any messages, but it's still a good idea to
+        # commit offsets if we've processed messages in the last set of batches.
+        # This also ensures the offsets are retained if we haven't read any messages
+        # since the offset retention period has elapsed.
+        @offset_manager.commit_offsets_if_necessary
       end
     end