remarks

Author: mensfeld

CHANGELOG.md

@@ -3,6 +3,7 @@
 ## 2.8.13 (Unreleased)
 - [Enhancement] Make `fenced` error skip-reload behavior configurable via new `non_reloadable_errors` setting (defaults to `[:fenced]` for backward compatibility).
 - [Enhancement] Add `producer.reload` event allowing config modification before reload to escape fencing loops (#706).
+- [Enhancement] Do not early initialize the new instance on reload.
 
 ## 2.8.12 (2025-10-10)
 - [Enhancement] Introduce `reload_on_idempotent_fatal_error` to automatically reload librdkafka producer after fatal errors on idempotent (non-transactional) producers.

lib/waterdrop/producer.rb

@@ -559,5 +559,15 @@ def produce(message)
     ensure
       @operations_in_progress.decrement
     end
+
+    # Reloads the client
+    # @note This should be used only within proper mutexes internally
+    def reload!
+      @client.flush(current_variant.max_wait_timeout)
+      purge
+      @client.close
+      @client = nil
+      @status.configured!
+    end
   end
 end

lib/waterdrop/producer/idempotence.rb

@@ -77,10 +77,7 @@ def idempotent_reload_client_on_fatal_error(attempt, error)
             producer_id: id,
             attempt: attempt
           ) do
-            @client.flush(current_variant.max_wait_timeout)
-            purge
-            @client.close
-            @client = Builder.new.call(self, @config)
+            reload!
           end
         end
       end

lib/waterdrop/producer/transactions.rb

@@ -63,6 +63,8 @@ def transaction
         @transaction_mutex.synchronize do
           ensure_active!
 
+          @reloaded = false
+
           transactional_instrument(:finished) do
             with_transactional_error_handling(:begin) do
               transactional_instrument(:started) { client.begin_transaction }
@@ -285,6 +287,11 @@ def transactional_reload_client_if_needed(error)
 
         # Check if we've exceeded max reload attempts
         return unless transactional_retryable?
+        # We bubble up transactional errors, so there are cases where when fencing is not
+        # considered a non-reloadable, two layers of error handling would attempt to reload the
+        # client causing double reload. This halts reload if we're in a configured state as it
+        # means, we've already reloaded and we are not even yet connected
+        return if @status.configured?
 
         # Increment attempts before reload
         @transaction_fatal_error_attempts += 1
@@ -304,16 +311,14 @@ def transactional_reload_client_if_needed(error)
           # Clear cached state that depends on config
           # We always clear @transactional as it might have been modified via the event
           @transactional = nil
+          @reloaded = true
 
           @monitor.instrument(
             'producer.reloaded',
             producer_id: id,
             attempt: @transaction_fatal_error_attempts
           ) do
-            @client.flush(current_variant.max_wait_timeout)
-            purge
-            @client.close
-            @client = Builder.new.call(self, @config)
+            reload!
           end
         end
 

spec/integrations/fatal_error_recovery/fencing_escape_with_reload_event_spec.rb

@@ -39,17 +39,20 @@
 
 # Subscribe to producer.reload event and modify transactional.id to escape fencing
 producer1.monitor.subscribe('producer.reload') do |event|
-  # When fenced, rotate to a new transactional.id
-  if event[:error].code == :fenced
-    new_id = "#{TRANSACTIONAL_ID}-recovered-#{Time.now.to_i}"
-    event[:caller].config.kafka[:'transactional.id'] = new_id
-    puts "Fencing detected! Rotating transactional.id to: #{new_id}"
-  end
+  config = event[:caller].config
+  config.kafka[:'transactional.id'] = "#{TRANSACTIONAL_ID}-recovered-#{Time.now.to_i}"
 end
 
 producer1.monitor.subscribe('producer.reloaded') { |event| reload_events << event }
 producer1.monitor.subscribe('error.occurred') { |event| error_events << event }
 
+topic_name = "it-fence-escape-#{SecureRandom.hex(6)}"
+
+# First transaction with producer1
+producer1.transaction do
+  producer1.produce_sync(topic: topic_name, payload: 'message1')
+end
+
 # Create second producer with same ID to cause fencing
 producer2 = WaterDrop::Producer.new do |config|
   config.kafka = {
@@ -62,13 +65,6 @@
   config.logger = Logger.new($stdout, level: Logger::INFO)
 end
 
-topic_name = "it-fence-escape-#{SecureRandom.hex(6)}"
-
-# First transaction with producer1
-producer1.transaction do
-  producer1.produce_sync(topic: topic_name, payload: 'message1')
-end
-
 # This transaction will fence producer1
 producer2.transaction do
   producer2.produce_sync(topic: topic_name, payload: 'message2')
@@ -80,8 +76,19 @@
     producer1.produce_sync(topic: topic_name, payload: 'message3-recovered')
   end
 rescue Rdkafka::RdkafkaError => e
-  puts "Failed after reload: #{e.message}"
-  exit(1)
+  # This is expected. User needs to retry transaction if wants
+  # Reloading does not mean, that fencing is not re-raised in the transactional mode
+  exit(1) unless e.code == :fenced
+end
+
+10.times do
+  producer1.transaction do
+    producer1.produce_sync(topic: topic_name, payload: 'message3-recovered')
+  end
+
+  producer2.transaction do
+    producer2.produce_sync(topic: topic_name, payload: 'message2')
+  end
 end
 
 producer1.close
@@ -91,7 +98,4 @@
 # Should have exactly 1 reload (not multiple like in the loop case)
 success = reload_events.size == 1 && reload_events.first[:attempt] == 1
 
-puts "Reload events: #{reload_events.size}"
-puts "Success: #{success}"
-
 exit(success ? 0 : 1)

spec/integrations/fatal_error_recovery/fencing_with_reload_loop_spec.rb

@@ -66,10 +66,14 @@ def configure_producer
     producer2.produce_sync(topic: topic_name, payload: 'message2')
   end
 
-  # This will trigger reload loop: fenced -> reload -> fenced -> reload...
   producer1.transaction do
     producer1.produce_sync(topic: topic_name, payload: 'message3')
   end
+
+  # This will trigger reload loop: fenced -> reload -> fenced -> reload...
+  producer2.transaction do
+    producer2.produce_sync(topic: topic_name, payload: 'message2')
+  end
 rescue Rdkafka::RdkafkaError => e
   exit(1) unless e.code == :fenced