add runner and datadog specs for process_message and process_batch ex…

…ception instrumentation

spec/datadog_spec.rb

@@ -87,6 +87,15 @@ def create_event(name, payload = {})
 
       subscriber.process_message(event)
     end
+
+    it 'publishes errors' do
+      event.payload[:exception] = StandardError.new("surprise")
+      expect(statsd).
+        to receive(:increment).
+        with("consumer.process_message.errors", tags: metric_tags)
+
+      subscriber.process_message(event)
+    end
   end
 
   describe '#process_batch' do
@@ -137,6 +146,15 @@ def create_event(name, payload = {})
 
       subscriber.process_batch(event)
     end
+
+    it 'publishes errors' do
+      event.payload[:exception] = StandardError.new("surprise")
+      expect(statsd).
+        to receive(:increment).
+        with("consumer.process_batch.errors", tags: metric_tags)
+
+      subscriber.process_batch(event)
+    end
   end
 
   describe '#join_group' do

spec/runner_spec.rb

@@ -248,6 +248,18 @@ def consumer_rebalance_listener=(_listenr)
 class FakeInstrumenter < Racecar::Instrumenter
   def initialize(*)
     super(backend: Racecar::NullInstrumenter)
+    @errored_events = []
+  end
+
+  def instrument(event_name, payload = {}, &block)
+    super
+  rescue => e
+    @errored_events << event_name unless @errored_events.include?(event_name)
+    raise
+  end
+
+  def event_raised_errors?(event_name)
+    @errored_events.include?(event_name)
   end
 end
 
@@ -497,6 +509,11 @@ def initialize(*)
 
         runner.run
       end
+
+      specify 'message processing errors are propagated to the instrumenter' do
+        kafka.deliver_message(StandardError.new("surprise"), topic: "greetings")
+        expect { runner.run }.to change { instrumenter.event_raised_errors?("process_message") }.to(true)
+      end
     end
   end
 
@@ -624,6 +641,11 @@ def initialize(*)
 
         runner.run
       end
+
+      specify 'batch processing errors are propagated to the instrumenter' do
+        kafka.deliver_message(StandardError.new("surprise"), topic: "greetings")
+        expect { runner.run }.to change { instrumenter.event_raised_errors?("process_batch") }.to(true)
+      end
     end
   end