Instrument fetch_batch for each_message

Author: michaelsauter

lib/kafka/consumer.rb

@@ -202,6 +202,15 @@ def each_message(min_bytes: 1, max_bytes: 10485760, max_wait_time: 1, automatica
         )
 
         batches.each do |batch|
+          unless batch.empty?
+            @instrumenter.instrument("fetch_batch.consumer", {
+              topic: batch.topic,
+              partition: batch.partition,
+              offset_lag: batch.offset_lag,
+              highwater_mark_offset: batch.highwater_mark_offset,
+              message_count: batch.messages.count,
+            })
+          end
           batch.messages.each do |message|
             notification = {
               topic: message.topic,

spec/consumer_spec.rb

@@ -75,6 +75,18 @@
       ]
     }
 
+    it "instruments" do
+      expect(instrumenter).to receive(:instrument).once.with('fetch_batch.consumer', anything)
+      expect(instrumenter).to receive(:instrument).once.with('start_process_message.consumer', anything)
+      expect(instrumenter).to receive(:instrument).once.with('process_message.consumer', anything)
+
+      allow(instrumenter).to receive(:instrument).and_call_original
+
+      consumer.each_message do |message|
+        consumer.stop
+      end
+    end
+
     it "raises ProcessingError if the processing code fails" do
       expect {
         consumer.each_message do |message|