def send_messages(messages)
      raise Errors::ProducerShutdownError if @shutdown
      if !messages.respond_to?(:each)
        raise ArgumentError, "messages must respond to #each"
      end

      @producer.send_messages(convert_to_messages_objects(messages))
    end
def send_messages(messages)
      raise Errors::ProducerShutdownError if @shutdown
      if !messages.respond_to?(:each)
        raise ArgumentError, "messages must respond to #each"
      end

      @producer.send_messages(convert_to_messages_objects(messages))
    end
Returns a new Producer.

 @param [Array<String>] brokers An array of brokers in the form "host1:port1"

 @param [String] client_id A client_id used to indentify the producer.

 @param [Hash] options

 @option options [:sync / :async] :type (:sync)
   Whether we should send messages right away or queue them and send
   them in the background.

 @option options [:gzip / :snappy / :none] :compression_codec (:none)
   Type of compression to use.

 @option options [Enumberable<String>] :compressed_topics (nil)
   Topics to compress.  If this is not specified we will compress all
   topics provided that +:compression_codec+ is set.

 @option options [Integer: Milliseconds] :metadata_refresh_interval_ms (600_000)
   How frequently we should update the topic metadata in milliseconds.

 @option options [#call, nil] :partitioner
   Object which partitions messages based on key.
   Responds to #call(key, partition_count).

 @option options [Integer] :max_send_retries (3)
   Number of times to retry sending of messages to a leader.

 @option options [Integer] :retry_backoff_ms (100)
   The amount of time (in milliseconds) to wait before refreshing the metadata
   after we are unable to send messages.
   Number of times to retry sending of messages to a leader.

 @option options [Integer] :required_acks (0)
   The number of acks required per request.

 @option options [Integer] :ack_timeout_ms (1500)
   How long the producer waits for acks.

 @option options [Integer] :socket_timeout_ms] (10000)
   How long the producer socket waits for any reply from server.

 @api public
 Send messages to the cluster. Raises an exception if the producer fails to send the messages.

 @param [Enumerable<MessageToSend>] messages
   Messages must have a +topic+ set and may have a +key+ set.

 @return [Boolean]

 @api public