Add topic and partition to acknowledged_message event

lib/racecar/datadog.rb

@@ -256,7 +256,11 @@ def deliver_messages(event)
       end
 
       def acknowledged_message(event)
-        tags = { client: event.payload.fetch(:client_id) }
+        tags = {
+          client: event.payload.fetch(:client_id),
+          topic: event.payload.fetch(:topic),
+          partition: event.payload.fetch(:partition)
+        }
 
         # Number of messages ACK'd for the topic.
         increment("producer.ack.messages", tags: tags)

lib/racecar/delivery_callback.rb

@@ -12,13 +12,15 @@ def call(delivery_report)
       if delivery_report.error.to_i.zero?
         payload = {
           offset: delivery_report.offset,
-          partition: delivery_report.partition
+          partition: delivery_report.partition,
+          topic: delivery_report.topic_name
         }
         instrumenter.instrument("acknowledged_message", payload)
       else
         payload = {
           partition: delivery_report.partition,
-          exception: delivery_report.error
+          exception: delivery_report.error,
+          topic: delivery_report.topic_name
         }
         instrumenter.instrument("produce_delivery_error", payload)
       end

spec/datadog_spec.rb

@@ -427,14 +427,18 @@ def create_event(name, payload = {})
   describe '#acknowledged_message' do
     let(:event) do
       create_event(
-        'deliver_messages',
+        'acknowledged_message',
         client_id: 'racecar',
-        delivered_message_count: 10
+        offset: 123,
+        partition: 1,
+        topic: 'test_topic'
       )
     end
     let(:metric_tags) do
       %w[
           client:racecar
+          topic:test_topic
+          partition:1
         ]
     end
 

spec/runner_spec.rb

@@ -204,7 +204,7 @@ def wait(max_wait_timeout: 60, wait_timeout: 0.1)
   end
 
   def create_result
-    Rdkafka::Producer::DeliveryReport.new(partition, offset)
+    Rdkafka::Producer::DeliveryReport.new(partition, offset, topic)
   end
 
   def offset
@@ -214,6 +214,10 @@ def offset
   def partition
     0
   end
+
+  def topic
+    "test"
+  end
 end
 
 class FakeRdkafka
@@ -719,7 +723,7 @@ def process_batch(batch, hello); end
       runner.run
 
       expect(instrumenter).to have_received(:instrument)
-        .with("acknowledged_message", {partition: 0, offset: 0})
+        .with("acknowledged_message", {partition: 0, offset: 0, topic: "test"})
     end
   end