Avoid race condition on client._conns in send() (#1772)

There was a very small possibility that between checking `self._can_send_request(node_id)` and grabbing the connection object via `self._conns[node_id]` that the connection could get closed / recycled / removed from _conns and cause a KeyError. This PR should prevent such a KeyError. In the case where the connection is disconnected by the time we call send(), we should expect conn.send() simply to fail the request.

Author: dpkp

kafka/client_async.py

@@ -516,14 +516,15 @@ def send(self, node_id, request, wakeup=True):
         Returns:
             Future: resolves to Response struct or Error
         """
-        if not self._can_send_request(node_id):
+        conn = self._conns.get(node_id)
+        if not conn or not self._can_send_request(node_id):
             self.maybe_connect(node_id, wakeup=wakeup)
             return Future().failure(Errors.NodeNotReadyError(node_id))
 
         # conn.send will queue the request internally
         # we will need to call send_pending_requests()
         # to trigger network I/O
-        future = self._conns[node_id].send(request, blocking=False)
+        future = conn.send(request, blocking=False)
 
         # Wakeup signal is useful in case another thread is
         # blocked waiting for incoming network traffic while holding