summaryrefslogtreecommitdiff
path: root/kafka/client_async.py
diff options
context:
space:
mode:
authorDana Powers <dana.powers@gmail.com>2019-04-01 19:38:48 -0700
committerGitHub <noreply@github.com>2019-04-01 19:38:48 -0700
commit51313d792a24059d003f5647ec531cfd9d62d7ab (patch)
tree2591e9ebfe8b5c23112074d08da647f2b7dffa0b /kafka/client_async.py
parentc02df4bcc6ee6920db1be259f44a8f958bb36791 (diff)
downloadkafka-python-51313d792a24059d003f5647ec531cfd9d62d7ab.tar.gz
Dont treat popped conn.close() as failure in state change callback (#1773)
Diffstat (limited to 'kafka/client_async.py')
-rw-r--r--kafka/client_async.py13
1 files changed, 10 insertions, 3 deletions
diff --git a/kafka/client_async.py b/kafka/client_async.py
index dc685f9..a86ab55 100644
--- a/kafka/client_async.py
+++ b/kafka/client_async.py
@@ -314,7 +314,12 @@ class KafkaClient(object):
idle_disconnect = True
self._idle_expiry_manager.remove(node_id)
- if self.cluster.is_bootstrap(node_id):
+ # If the connection has already by popped from self._conns,
+ # we can assume the disconnect was intentional and not a failure
+ if node_id not in self._conns:
+ pass
+
+ elif self.cluster.is_bootstrap(node_id):
self._bootstrap_fails += 1
elif self._refresh_on_disconnects and not self._closed and not idle_disconnect:
@@ -419,10 +424,12 @@ class KafkaClient(object):
with self._lock:
if node_id is None:
self._close()
- for conn in self._conns.values():
+ conns = list(self._conns.values())
+ self._conns.clear()
+ for conn in conns:
conn.close()
elif node_id in self._conns:
- self._conns[node_id].close()
+ self._conns.pop(node_id).close()
else:
log.warning("Node %s not found in current connection list; skipping", node_id)
return