def produce(value, key: nil, headers: {}, topic:, partition: nil, partition_key: nil, create_time: Time.now)
      message = PendingMessage.new(
        value: value && value.to_s,
        key: key && key.to_s,
        headers: headers,
        topic: topic.to_s,
        partition: partition && Integer(partition),
        partition_key: partition_key && partition_key.to_s,
        create_time: create_time
      )

      if buffer_size >= @max_buffer_size
        buffer_overflow topic,
          "Cannot produce to #{topic}, max buffer size (#{@max_buffer_size} messages) reached"
      end

      if buffer_bytesize + message.bytesize >= @max_buffer_bytesize
        buffer_overflow topic,
          "Cannot produce to #{topic}, max buffer bytesize (#{@max_buffer_bytesize} bytes) reached"
      end

      # If the producer is in transactional mode, all the message production
      # must be used when the producer is currently in transaction
      if @transaction_manager.transactional? && !@transaction_manager.in_transaction?
        raise "Cannot produce to #{topic}: You must trigger begin_transaction before producing messages"
      end

      @target_topics.add(topic)
      @pending_message_queue.write(message)

      @instrumenter.instrument("produce_message.producer", {
        value: value,
        key: key,
        topic: topic,
        create_time: create_time,
        message_size: message.bytesize,
        buffer_size: buffer_size,
        max_buffer_size: @max_buffer_size,
      })

      nil
    end
def produce(value, key: nil, headers: {}, topic:, partition: nil, partition_key: nil, create_time: Time.now)
      message = PendingMessage.new(
        value: value && value.to_s,
        key: key && key.to_s,
        headers: headers,
        topic: topic.to_s,
        partition: partition && Integer(partition),
        partition_key: partition_key && partition_key.to_s,
        create_time: create_time
      )

      if buffer_size >= @max_buffer_size
        buffer_overflow topic,
          "Cannot produce to #{topic}, max buffer size (#{@max_buffer_size} messages) reached"
      end

      if buffer_bytesize + message.bytesize >= @max_buffer_bytesize
        buffer_overflow topic,
          "Cannot produce to #{topic}, max buffer bytesize (#{@max_buffer_bytesize} bytes) reached"
      end

      # If the producer is in transactional mode, all the message production
      # must be used when the producer is currently in transaction
      if @transaction_manager.transactional? && !@transaction_manager.in_transaction?
        raise "Cannot produce to #{topic}: You must trigger begin_transaction before producing messages"
      end

      @target_topics.add(topic)
      @pending_message_queue.write(message)

      @instrumenter.instrument("produce_message.producer", {
        value: value,
        key: key,
        topic: topic,
        create_time: create_time,
        message_size: message.bytesize,
        buffer_size: buffer_size,
        max_buffer_size: @max_buffer_size,
      })

      nil
    end
Produces a message to the specified topic. Note that messages are buffered in
 the producer until {#deliver_messages} is called.

 ## Partitioning

 There are several options for specifying the partition that the message should
 be written to.

 The simplest option is to not specify a message key, partition key, or
 partition number, in which case the message will be assigned a partition at
 random.

 You can also specify the `partition` parameter yourself. This requires you to
 know which partitions are available, however. Oftentimes the best option is
 to specify the `partition_key` parameter: messages with the same partition
 key will always be assigned to the same partition, as long as the number of
 partitions doesn't change. You can also omit the partition key and specify
 a message key instead. The message key is part of the message payload, and
 so can carry semantic value--whether you want to have the message key double
 as a partition key is up to you.

 @param value [String] the message data.
 @param key [String] the message key.
 @param headers [Hash<String, String>] the headers for the message.
 @param topic [String] the topic that the message should be written to.
 @param partition [Integer] the partition that the message should be written to.
 @param partition_key [String] the key that should be used to assign a partition.
 @param create_time [Time] the timestamp that should be set on the message.

 @raise [BufferOverflow] if the maximum buffer size has been reached.
 @return [nil]