code: def send_fetch_request(self, payloads=None, fail_on_error=True,
                             callback=None,
                             max_wait_time=DEFAULT_FETCH_SERVER_WAIT_MSECS,
                             min_bytes=DEFAULT_FETCH_MIN_BYTES):
          
          if (max_wait_time / 1000) > (self.timeout - 0.1):
              raise ValueError(
                  "%r: max_wait_time: %d must be less than client.timeout by "
                  "at least 100 milliseconds.", self, max_wait_time)
  
          encoder = partial(KafkaCodec.encode_fetch_request,
                            max_wait_time=max_wait_time,
                            min_bytes=min_bytes)
  
          # resps is a list of FetchResponse() objects, each of which can hold
          # 1-n messages.
          resps = yield self._send_broker_aware_request(
              payloads, encoder,
              KafkaCodec.decode_fetch_response)
  
          returnValue(self._handle_responses(resps, fail_on_error, callback))
docs: Encode and send a FetchRequest
  
          Payloads are grouped by topic and partition so they can be pipelined
          to the same brokers.
  
          Raises
          ======
          FailedPayloadsError, LeaderUnavailableError, PartitionUnavailableError