lib/kafka/async_producer.rb in ruby-kafka-0.7.5 vs lib/kafka/async_producer.rb in ruby-kafka-0.7.6.beta1

@@ -78,11 +78,11 @@
       raise ArgumentError unless delivery_interval >= 0
 
       @queue = Queue.new
       @max_queue_size = max_queue_size
       @instrumenter = instrumenter
-      @logger = logger
+      @logger = TaggedLogger.new(logger)
 
       @worker = Worker.new(
         queue: @queue,
         producer: sync_producer,
         delivery_threshold: delivery_threshold,
@@ -187,14 +187,15 @@
       def initialize(queue:, producer:, delivery_threshold:, instrumenter:, logger:)
         @queue = queue
         @producer = producer
         @delivery_threshold = delivery_threshold
         @instrumenter = instrumenter
-        @logger = logger
+        @logger = TaggedLogger.new(logger)
       end
 
       def run
+        @logger.push_tags(@producer.to_s)
         @logger.info "Starting async producer in the background..."
 
         loop do
           operation, payload = @queue.pop
 
@@ -231,9 +232,10 @@
       rescue Exception => e
         @logger.error "Unexpected Kafka error #{e.class}: #{e.message}\n#{e.backtrace.join("\n")}"
         @logger.error "Async producer crashed!"
       ensure
         @producer.shutdown
+        @logger.pop_tags
       end
 
       private
 
       def produce(*args)