support producers tagging

CHANGELOG.md

@@ -1,5 +1,8 @@
 # WaterDrop changelog
 
+## 2.8.2 (Unreleased)
+- [Feature] Allow for tagging of producer instances similar to how consumers can be tagged.
+
 ## 2.8.1 (2024-12-26)
 - [Enhancement] Raise `WaterDrop::ProducerNotTransactionalError` when attempting to use transactions on a non-transactional producer.
 - [Fix] Disallow closing a producer from within a transaction.

Gemfile.lock

@@ -1,7 +1,7 @@
 PATH
   remote: .
   specs:
-    waterdrop (2.8.1)
+    waterdrop (2.8.2)
       karafka-core (>= 2.4.3, < 3.0.0)
       karafka-rdkafka (>= 0.17.5)
       zeitwerk (~> 2.3)

lib/waterdrop/producer.rb

@@ -9,6 +9,7 @@ class Producer
     include Buffer
     include Transactions
     include ::Karafka::Core::Helpers::Time
+    include ::Karafka::Core::Taggable
 
     # Local storage for given thread waterdrop client references for variants
     ::Fiber.send(:attr_accessor, :waterdrop_clients)

lib/waterdrop/version.rb

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

spec/lib/waterdrop/producer_spec.rb

@@ -322,6 +322,19 @@ def on_oauthbearer_token_refresh(_); end
     end
   end
 
+  describe '#tags' do
+    let(:producer1) { build(:producer) }
+    let(:producer2) { build(:producer) }
+
+    before do
+      producer1.tags.add(:type, 'transactional')
+      producer2.tags.add(:type, 'regular')
+    end
+
+    it { expect(producer1.tags.to_a).to eq(%w[transactional]) }
+    it { expect(producer2.tags.to_a).to eq(%w[regular]) }
+  end
+
   describe 'statistics callback hook' do
     let(:message) { build(:valid_message) }