raise on transactional attempt when non-transactional (#546)

CHANGELOG.md

@@ -1,5 +1,8 @@
 # WaterDrop changelog
 
+## 2.8.1 (Unreleased)
+- [Enhancement] Raise `WaterDrop::ProducerNotTransactionalError` when attempting to use transactions on a non-transactional producer.
+
 ## 2.8.0 (2024-09-16)
 
 This release contains **BREAKING** changes. Make sure to read and apply upgrade notes.

Gemfile.lock

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

lib/waterdrop/errors.rb

@@ -31,6 +31,9 @@ module Errors
     # Raised when we want to commit transactional offset and the input is invalid
     TransactionalOffsetInvalidError = Class.new(BaseError)
 
+    # Raised when transaction attempt happens on a non-transactional producer
+    ProducerNotTransactionalError = Class.new(BaseError)
+
     # Raised when we've got an unexpected status. This should never happen. If it does, please
     # contact us as it is an error.
     StatusInvalidError = Class.new(BaseError)

lib/waterdrop/producer/transactions.rb

@@ -55,6 +55,13 @@ module Transactions
       #
       #   handler.wait
       def transaction
+        unless transactional?
+          raise(
+            Errors::ProducerNotTransactionalError,
+            "#{id} is not transactional"
+          )
+        end
+
         # This will safely allow us to support one operation transactions so a transactional
         # producer can work without the transactional block if needed
         return yield if @transaction_mutex.owned?

lib/waterdrop/version.rb

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

spec/lib/waterdrop/clients/buffered_spec.rb

@@ -66,6 +66,16 @@
   end
 
   describe '#transaction' do
+    let(:producer) do
+      WaterDrop::Producer.new do |config|
+        config.deliver = false
+        config.kafka = {
+          'bootstrap.servers': 'localhost:9092',
+          'transactional.id': SecureRandom.uuid
+        }
+      end
+    end
+
     context 'when no error and no abort' do
       it 'expect to return the block value' do
         expect(producer.transaction { 1 }).to eq(1)

spec/lib/waterdrop/clients/dummy_spec.rb

@@ -105,6 +105,16 @@
   end
 
   describe '#transaction' do
+    let(:producer) do
+      WaterDrop::Producer.new do |config|
+        config.deliver = false
+        config.kafka = {
+          'bootstrap.servers': 'localhost:9092',
+          'transactional.id': SecureRandom.uuid
+        }
+      end
+    end
+
     context 'when no error and no abort' do
       it 'expect to return the block value' do
         expect(producer.transaction { 1 }).to eq(1)

spec/lib/waterdrop/producer/transactions_spec.rb

@@ -35,7 +35,7 @@
 
     it 'expect to raise with info that this functionality is not configured' do
       expect { producer.transaction {} }
-        .to raise_error(::Rdkafka::RdkafkaError, /Local: Functionality not configured/)
+        .to raise_error(::WaterDrop::Errors::ProducerNotTransactionalError)
     end
 
     it { expect(producer.transactional?).to eq(false) }