summaryrefslogtreecommitdiff
diff options
context:
space:
mode:
authorDana Powers <dana.powers@gmail.com>2019-03-31 19:22:19 -0700
committerJeff Widman <jeff@jeffwidman.com>2019-03-31 19:22:19 -0700
commitc02df4bcc6ee6920db1be259f44a8f958bb36791 (patch)
tree738362abc2665de4806d7b2e8c3b7bda29a57f07
parent3664ae85e5a4c47075489e01688897f8cea8b11d (diff)
downloadkafka-python-c02df4bcc6ee6920db1be259f44a8f958bb36791.tar.gz
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.
-rw-r--r--kafka/client_async.py5
1 files changed, 3 insertions, 2 deletions
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