diff options
author | Dana Powers <dana.powers@gmail.com> | 2019-03-31 10:12:05 -0700 |
---|---|---|
committer | Dana Powers <dana.powers@gmail.com> | 2019-03-31 16:35:23 -0700 |
commit | e331261585bce06aa1c979d33f3932572115ba01 (patch) | |
tree | c715e6b6d0432c8864e60ab369c4ab632d399357 | |
parent | b1effa24aca3a6bcf2268354caae12ee82d6b36d (diff) | |
download | kafka-python-client_send_race.tar.gz |
Avoid race condition on client._conns in send()client_send_race
-rw-r--r-- | kafka/client_async.py | 5 |
1 files changed, 3 insertions, 2 deletions
diff --git a/kafka/client_async.py b/kafka/client_async.py index b6adb77..3b5ca46 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 |