Class: Kafka::AsyncProducer::Worker

Inherits:

Object

• Object
• Kafka::AsyncProducer::Worker

Defined in:

lib/kafka/async_producer.rb

Instance Method Summary

• #initialize(queue:, producer:, delivery_threshold:, max_retries: -1,, retry_backoff: 0, instrumenter:, logger:) ⇒ Worker constructor

  A new instance of Worker.

• #run ⇒ Object

Constructor Details

#initialize(queue:, producer:, delivery_threshold:, max_retries: -1,, retry_backoff: 0, instrumenter:, logger:) ⇒ Worker

Returns a new instance of Worker.

# File 'lib/kafka/async_producer.rb', line 201

def initialize(queue:, producer:, delivery_threshold:, max_retries: -1, retry_backoff: 0, instrumenter:, logger:)
  @queue = queue
  @producer = producer
  @delivery_threshold = delivery_threshold
  @max_retries = max_retries
  @retry_backoff = retry_backoff
  @instrumenter = instrumenter
  @logger = TaggedLogger.new(logger)
end

Instance Method Details

#run ⇒ Object

# File 'lib/kafka/async_producer.rb', line 211

def run
  @logger.push_tags(@producer.to_s)
  @logger.info "Starting async producer in the background..."

  loop do
    operation, payload = @queue.pop

    case operation
    when :produce
      produce(payload[0], **payload[1])
      deliver_messages if threshold_reached?
    when :deliver_messages
      deliver_messages
    when :shutdown
      begin
        # Deliver any pending messages first.
        @producer.deliver_messages
      rescue Error => e
        @logger.error("Failed to deliver messages during shutdown: #{e.message}")

        @instrumenter.instrument("drop_messages.async_producer", {
          message_count: @producer.buffer_size + @queue.size,
        })
      end

      # Stop the run loop.
      break
    else
      raise "Unknown operation #{operation.inspect}"
    end
  end
rescue Kafka::Error => e
  @logger.error "Unexpected Kafka error #{e.class}: #{e.message}\n#{e.backtrace.join("\n")}"
  @logger.info "Restarting in 10 seconds..."

  sleep 10
  retry
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