lib/kafka/fetcher.rb in ruby-kafka-0.6.8 vs lib/kafka/fetcher.rb in ruby-kafka-0.7.0.alpha1
- old
+ new
@@ -136,12 +136,12 @@
partition: batch.partition,
offset_lag: batch.offset_lag,
highwater_mark_offset: batch.highwater_mark_offset,
message_count: batch.messages.count,
})
-
- @next_offsets[batch.topic][batch.partition] = batch.last_offset + 1
end
+
+ @next_offsets[batch.topic][batch.partition] = batch.last_offset + 1
end
@queue << [:batches, batches]
rescue Kafka::NoPartitionsToFetchFrom
@logger.warn "No partitions to fetch from, sleeping for 1s"