def produce(value, topic:, **options)
      ensure_threads_running!

      if @queue.size >= @max_queue_size
        buffer_overflow topic,
          "Cannot produce to #{topic}, max queue size (#{@max_queue_size} messages) reached"
      end

      args = [value, **options.merge(topic: topic)]
      @queue << [:produce, args]

      @instrumenter.instrument("enqueue_message.async_producer", {
        topic: topic,
        queue_size: @queue.size,
        max_queue_size: @max_queue_size,
      })

      nil
    end
def produce(value, topic:, **options)
      ensure_threads_running!

      if @queue.size >= @max_queue_size
        buffer_overflow topic,
          "Cannot produce to #{topic}, max queue size (#{@max_queue_size} messages) reached"
      end

      args = [value, **options.merge(topic: topic)]
      @queue << [:produce, args]

      @instrumenter.instrument("enqueue_message.async_producer", {
        topic: topic,
        queue_size: @queue.size,
        max_queue_size: @max_queue_size,
      })

      nil
    end
Initializes a new AsyncProducer.

 @param sync_producer [Kafka::Producer] the synchronous producer that should
   be used in the background.
 @param max_queue_size [Integer] the maximum number of messages allowed in
   the queue.
 @param delivery_threshold [Integer] if greater than zero, the number of
   buffered messages that will automatically trigger a delivery.
 @param delivery_interval [Integer] if greater than zero, the number of
   seconds between automatic message deliveries.

 Produces a message to the specified topic.

 @see Kafka::Producer#produce
 @param (see Kafka::Producer#produce)
 @raise [BufferOverflow] if the message queue is full.
 @return [nil]