From c02df4bcc6ee6920db1be259f44a8f958bb36791 Mon Sep 17 00:00:00 2001 From: Dana Powers Date: Sun, 31 Mar 2019 19:22:19 -0700 Subject: 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. --- kafka/client_async.py | 5 +++-- 1 file changed, 3 insertions(+), 2 deletions(-) (limited to 'kafka/client_async.py') diff --git a/kafka/client_async.py b/kafka/client_async.py index ba5c960..dc685f9 100644 --- a/kafka/client_async.py +++ b/kafka/client_async.py @@ -516,14 +516,15 @@ class KafkaClient(object): 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 -- cgit v1.2.1