v2.5.3 (#349)

CHANGELOG.md

@@ -1,5 +1,10 @@
 # WaterDrop changelog
 
+## 2.5.3 (2023-05-26)
+- Require `karafka-core` `2.0.13`
+- Include topic name in the `error.occurred` notification payload.
+- Include topic name in the `message.acknowledged` notification payload.
+
 ## 2.5.2 (2023-04-24)
 - [Fix] Require missing Pathname (#345)
 

Gemfile.lock

@@ -1,8 +1,8 @@
 PATH
   remote: .
   specs:
-    waterdrop (2.5.2)
-      karafka-core (>= 2.0.12, < 3.0.0)
+    waterdrop (2.5.3)
+      karafka-core (>= 2.0.13, < 3.0.0)
       zeitwerk (~> 2.3)
 
 GEM
@@ -22,10 +22,10 @@ GEM
     ffi (1.15.5)
     i18n (1.12.0)
       concurrent-ruby (~> 1.0)
-    karafka-core (2.0.12)
+    karafka-core (2.0.13)
       concurrent-ruby (>= 1.1)
-      karafka-rdkafka (>= 0.12.1)
-    karafka-rdkafka (0.12.1)
+      karafka-rdkafka (>= 0.12.3)
+    karafka-rdkafka (0.12.3)
       ffi (~> 1.15)
       mini_portile2 (~> 2.6)
       rake (> 12)

lib/waterdrop/instrumentation/callbacks/delivery.rb

@@ -35,6 +35,7 @@ def instrument_error(delivery_report)
             producer_id: @producer_id,
             offset: delivery_report.offset,
             partition: delivery_report.partition,
+            topic: delivery_report.topic_name,
             type: 'librdkafka.dispatch_error'
           )
         end
@@ -45,7 +46,8 @@ def instrument_acknowledged(delivery_report)
             'message.acknowledged',
             producer_id: @producer_id,
             offset: delivery_report.offset,
-            partition: delivery_report.partition
+            partition: delivery_report.partition,
+            topic: delivery_report.topic_name
           )
         end
       end

lib/waterdrop/version.rb

@@ -3,5 +3,5 @@
 # WaterDrop library
 module WaterDrop
   # Current WaterDrop version
-  VERSION = '2.5.2'
+  VERSION = '2.5.3'
 end

spec/lib/waterdrop/instrumentation/callbacks/delivery_spec.rb

@@ -6,7 +6,13 @@
   let(:producer) { build(:producer) }
   let(:producer_id) { SecureRandom.uuid }
   let(:monitor) { ::WaterDrop::Instrumentation::Monitor.new }
-  let(:delivery_report) { OpenStruct.new(offset: rand(100), partition: rand(100)) }
+  let(:delivery_report) do
+    OpenStruct.new(
+      offset: rand(100),
+      partition: rand(100),
+      topic_name: rand(100).to_s
+    )
+  end
 
   describe '#call' do
     let(:changed) { [] }
@@ -24,23 +30,26 @@
     it { expect(event[:producer_id]).to eq(producer_id) }
     it { expect(event[:offset]).to eq(delivery_report.offset) }
     it { expect(event[:partition]).to eq(delivery_report.partition) }
+    it { expect(event[:topic]).to eq(delivery_report.topic_name) }
   end
 
   describe '#when we do an end-to-end delivery report check' do
     context 'when there is a message that was successfully delivered' do
       let(:changed) { [] }
       let(:event) { changed.first }
+      let(:message) { build(:valid_message) }
 
       before do
         producer.monitor.subscribe('message.acknowledged') do |event|
           changed << event
         end
 
-        producer.produce_sync(build(:valid_message))
+        producer.produce_sync(message)
       end
 
       it { expect(event.payload[:partition]).to eq(0) }
       it { expect(event.payload[:offset]).to eq(0) }
+      it { expect(event[:topic]).to eq(message[:topic]) }
     end
 
     context 'when there is a message that was not successfully delivered async' do
@@ -62,6 +71,7 @@
       it { expect(event.payload[:error]).to be_a(Rdkafka::RdkafkaError) }
       it { expect(event.payload[:partition]).to eq(-1) }
       it { expect(event.payload[:offset]).to eq(-1001) }
+      it { expect(event.payload[:topic]).to eq('$%^&*') }
     end
 
     context 'when there is a message that was not successfully delivered sync' do

waterdrop.gemspec

@@ -16,7 +16,7 @@ Gem::Specification.new do |spec|
   spec.description   = spec.summary
   spec.license       = 'MIT'
 
-  spec.add_dependency 'karafka-core', '>= 2.0.12', '< 3.0.0'
+  spec.add_dependency 'karafka-core', '>= 2.0.13', '< 3.0.0'
   spec.add_dependency 'zeitwerk', '~> 2.3'
 
   if $PROGRAM_NAME.end_with?('gem')