Handle SSL HANDSHAKE state in KafkaClient state change handler

kafka/client_async.py

@@ -168,6 +168,8 @@ class KafkaClient(object):
 
     def _conn_state_change(self, node_id, conn):
         if conn.connecting():
+            # 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)