def recv(self):
        """Non-blocking network receive.

        Return list of (response, future) tuples
        """
        responses = self._recv()
        if not responses and self.requests_timed_out():
            timed_out = self.timed_out_ifrs()
            timeout_ms = (timed_out[0][2] - timed_out[0][1]) * 1000
            log.warning('%s: timed out after %s ms. Closing connection.',
                        self, timeout_ms)
            self.close(error=Errors.RequestTimedOutError(
                'Request timed out after %s ms' %
                timeout_ms))
            return ()

        # augment responses w/ correlation_id, future, and timestamp
        for i, (correlation_id, response) in enumerate(responses):
            try:
                with self._lock:
                    (future, timestamp, _timeout) = 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)
            self._maybe_throttle(response)
            responses[i] = (response, future)

        return responses
Non-blocking network receive.

        Return list of (response, future) tuples
        
kafka/conn.py