code: def recv(self):
          
          responses = self._recv()
          if not responses and self.requests_timed_out():
              log.warning('%s timed out after %s ms. Closing connection.',
                          self, self.config['request_timeout_ms'])
              self.close(error=Errors.RequestTimedOutError(
                  'Request timed out after %s ms' %
                  self.config['request_timeout_ms']))
              return ()
  
          # augment respones w/ correlation_id, future, and timestamp
          for i, (correlation_id, response) in enumerate(responses):
              try:
                  with self._lock:
                      (future, timestamp) = self.in_flight_requests.pop(correlation_id)
              except KeyError:
                  self.close(Errors.KafkaConnectionError('Received unrecognized correlation id'))
                  return ()
              latency_ms = (time.time() - timestamp) * 1000
              if self._sensors:
                  self._sensors.request_time.record(latency_ms)
  
              log.debug('%s Response %d (%s ms): %s', self, correlation_id, latency_ms, response)
              responses[i] = (response, future)
  
          return responses
docs: Non-blocking network receive.
  
          Return list of (response, future) tuples