Always check for request timeouts (dpkp#887)

dpkp · web-flow · commit f71cfc4607c0 · 2016-11-18T09:07:53.000-08:00
* Check for requests that timeout without causing a socket read/write event
diff --git a/kafka/client_async.py b/kafka/client_async.py
@@ -578,6 +578,14 @@ def _poll(self, timeout, sleep=True):
                     if response:
                         responses.append(response)
 
+        for conn in six.itervalues(self._conns):
+            if conn.requests_timed_out():
+                log.warning('%s timed out after %s ms. Closing connection.',
+                            conn, conn.config['request_timeout_ms'])
+                conn.close(error=Errors.RequestTimedOutError(
+                    'Request timed out after %s ms' %
+                    conn.config['request_timeout_ms']))
+
         if self._sensors:
             self._sensors.io_time.record((time.time() - end_select) * 1000000000)
         return responses
diff --git a/kafka/conn.py b/kafka/conn.py
@@ -575,15 +575,15 @@ def recv(self):
             log.warning('%s: No in-flight-requests to recv', self)
             return None
 
-        elif self._requests_timed_out():
+        response = self._recv()
+        if not response 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 None
-
-        return self._recv()
+        return response
 
     def _recv(self):
         # Not receiving is the state of reading the payload header
@@ -719,7 +719,7 @@ def _process_response(self, read_buffer):
         self._processing = False
         return response
 
-    def _requests_timed_out(self):
+    def requests_timed_out(self):
         if self.in_flight_requests:
             oldest_at = self.in_flight_requests[0].timestamp
             timeout = self.config['request_timeout_ms'] / 1000.0
