Check for None returned from least_loaded_node when no brokers are availablenone_least_loaded_node

2 files changed, 8 insertions, 0 deletions

diff --git a/kafka/client_async.py b/kafka/client_async.py
index 973ece0..57aea66 100644
--- a/kafka/client_async.py
+++ b/kafka/client_async.py
@@ -532,6 +532,8 @@ class KafkaClient(object):
             return 9999999999
 
         node_id = self.least_loaded_node()
+        if node_id is None:
+            return 0
 
         topics = list(self._topics)
         if self.cluster.need_all_topic_metadata:
@@ -588,6 +590,8 @@ class KafkaClient(object):
         """Attempt to guess the broker version"""
         if node_id is None:
             node_id = self.least_loaded_node()
+            if node_id is None:
+                raise Errors.NoBrokersAvailable()
 
         def connect(node_id):
             timeout_at = time.time() + timeout
diff --git a/kafka/coordinator/consumer.py b/kafka/coordinator/consumer.py
index 0e610c7..a5e3067 100644
--- a/kafka/coordinator/consumer.py
+++ b/kafka/coordinator/consumer.py
@@ -414,6 +414,8 @@ class ConsumerCoordinator(BaseCoordinator):
             node_id = self.coordinator_id
         else:
             node_id = self._client.least_loaded_node()
+            if node_id is None:
+                return Future().failure(Errors.NoBrokersAvailable)
 
         # create the offset commit request
         offset_data = collections.defaultdict(dict)
@@ -560,6 +562,8 @@ class ConsumerCoordinator(BaseCoordinator):
             node_id = self.coordinator_id
         else:
             node_id = self._client.least_loaded_node()
+            if node_id is None:
+                return Future().failure(Errors.NoBrokersAvailable)
 
         # Verify node is ready
         if not self._client.ready(node_id):