Timeout all unconnected conns (incl SSL) after request_timeout_ms

Author: dpkp

kafka/conn.py

@@ -351,7 +351,6 @@ def connect(self):
         if self.state is ConnectionStates.CONNECTING:
             # in non-blocking mode, use repeated calls to socket.connect_ex
             # to check connection status
-            request_timeout = self.config['request_timeout_ms'] / 1000.0
             ret = None
             try:
                 ret = self._sock.connect_ex(self._sock_addr)
@@ -389,11 +388,6 @@ def connect(self):
                 errstr = errno.errorcode.get(ret, 'UNKNOWN')
                 self.close(Errors.KafkaConnectionError('{} {}'.format(ret, errstr)))
 
-            # Connection timed out
-            elif time.time() > request_timeout + self.last_attempt:
-                log.error('Connection attempt to %s timed out', self)
-                self.close(Errors.KafkaConnectionError('timeout'))
-
             # Needs retry
             else:
                 pass
@@ -419,6 +413,14 @@ def connect(self):
                     self._reset_reconnect_backoff()
                     self.config['state_change_callback'](self)
 
+        if self.state not in (ConnectionStates.CONNECTED,
+                              ConnectionStates.DISCONNECTED):
+            # Connection timed out
+            request_timeout = self.config['request_timeout_ms'] / 1000.0
+            if time.time() > request_timeout + self.last_attempt:
+                log.error('Connection attempt to %s timed out', self)
+                self.close(Errors.KafkaConnectionError('timeout'))
+
         return self.state
 
     def _wrap_ssl(self):