Handle SSL HANDSHAKE state in KafkaClient state change handler

dpkp · dpkp · commit 5b3042cb6d4f · 2016-04-09T09:24:39.000-07:00
diff --git a/kafka/client_async.py b/kafka/client_async.py
@@ -168,8 +168,10 @@ def _can_connect(self, node_id):
 
     def _conn_state_change(self, node_id, conn):
         if conn.connecting():
-            self._connecting.add(node_id)
-            self._selector.register(conn._sock, selectors.EVENT_WRITE)
+            # SSL connections can enter this state 2x (second during Handshake)
+            if node_id not in self._connecting:
+                self._connecting.add(node_id)
+                self._selector.register(conn._sock, selectors.EVENT_WRITE)
 
         elif conn.connected():
             log.debug("Node %s connected", node_id)
