Always check for request timeouts (#887)

* Check for requests that timeout without causing a socket read/write event

1 files changed, 8 insertions, 0 deletions

diff --git a/kafka/client_async.py b/kafka/client_async.py
index 03a2f00..bd9bf2e 100644
--- a/kafka/client_async.py
+++ b/kafka/client_async.py
@@ -578,6 +578,14 @@ class KafkaClient(object):
                     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