Commit offset only if records were consumed

Currently offsets are committed after each poll, even if no records were
consumed.

Author: makubi

lib/logstash/inputs/kafka.rb

@@ -250,6 +250,7 @@ def thread_runner(logstash_queue, consumer)
         codec_instance = @codec.clone
         while !stop?
           records = consumer.poll(poll_timeout_ms)
+          next unless records.count > 0
           for record in records do
             codec_instance.decode(record.value.to_s) do |event|
               decorate(event)