def fetch(options = {})
      fetch_max_wait = options.delete(:max_wait_ms) || max_wait_ms
      fetch_max_bytes = options.delete(:max_bytes) || max_bytes
      fetch_min_bytes = options.delete(:min_bytes) || min_bytes

      if options.keys.any?
        raise ArgumentError, "Unknown options: #{options.keys.inspect}"
      end

      topic_fetches = build_topic_fetch_request(fetch_max_bytes)
      fetch_response = @connection.fetch(fetch_max_wait, fetch_min_bytes, topic_fetches)
      topic_response = fetch_response.topic_fetch_responses.first 
      partition_response = topic_response.partition_fetch_responses.first

      unless partition_response.error == Errors::NO_ERROR_CODE
        if @offset < 0 &&
          Errors::ERROR_CODES[partition_response.error] == Errors::OffsetOutOfRange
          @offset = :earliest_offset
          return fetch(options)
        end

        raise Errors::ERROR_CODES[partition_response.error]
      else
        @highwater_mark = partition_response.highwater_mark_offset
        messages = partition_response.message_set.flatten.map do |m|
          FetchedMessage.new(topic_response.topic, m.value, m.key, m.offset)
        end
        if messages.any?
          @offset = messages.last.offset + 1
        end
        messages
      end
    end
def fetch(options = {})
      fetch_max_wait = options.delete(:max_wait_ms) || max_wait_ms
      fetch_max_bytes = options.delete(:max_bytes) || max_bytes
      fetch_min_bytes = options.delete(:min_bytes) || min_bytes

      if options.keys.any?
        raise ArgumentError, "Unknown options: #{options.keys.inspect}"
      end

      topic_fetches = build_topic_fetch_request(fetch_max_bytes)
      fetch_response = @connection.fetch(fetch_max_wait, fetch_min_bytes, topic_fetches)
      topic_response = fetch_response.topic_fetch_responses.first 
      partition_response = topic_response.partition_fetch_responses.first

      unless partition_response.error == Errors::NO_ERROR_CODE
        if @offset < 0 &&
          Errors::ERROR_CODES[partition_response.error] == Errors::OffsetOutOfRange
          @offset = :earliest_offset
          return fetch(options)
        end

        raise Errors::ERROR_CODES[partition_response.error]
      else
        @highwater_mark = partition_response.highwater_mark_offset
        messages = partition_response.message_set.flatten.map do |m|
          FetchedMessage.new(topic_response.topic, m.value, m.key, m.offset)
        end
        if messages.any?
          @offset = messages.last.offset + 1
        end
        messages
      end
    end
Create a new consumer which reads the specified topic and partition from
 the host.

 @param [String] client_id  Used to identify this client should be unique.
 @param [String] host
 @param [Integer] port
 @param [String] topic Topic to read from
 @param [Integer] partition Partitions are zero indexed.
 @param [Integer,Symbol] offset
   Offset to start reading from. A negative offset can also be passed.
   There are a couple special offsets which can be passed as symbols:
     :earliest_offset       Start reading from the first offset the server has.
     :latest_offset         Start reading from the latest offset the server has.

 @param [Hash] options
   Theses options can all be overridden in each individual fetch command.

 @option options [Integer] :max_bytes
   Maximum number of bytes to fetch
   Default: 1048576 (1MB)

 @option options [Integer] :max_wait_ms
   How long to block until the server sends us data.
   NOTE: This is only enforced if min_bytes is > 0.
   Default: 100 (100ms)

 @option options [Integer] :min_bytes
   Smallest amount of data the server should send us.
   Default: 1 (Send us data as soon as it is ready)

 @option options [Integer] :socket_timeout_ms
   How long to wait for reply from server. Should be higher than max_wait_ms.
   Default: 10000 (10s)

 @api public
 Fetch messages from the broker.

 @param [Hash] options

 @option options [Integer] :max_bytes
   Maximum number of bytes to fetch

 @option options [Integer] :max_wait_ms
   How long to block until the server sends us data.

 @option options [Integer] :min_bytes
   Smallest amount of data the server should send us.

 @api public