Move bootstrap_brokers fallback from brokers() to least_loaded_node() (#2809)

Author: dpkp

kafka/client_async.py

@@ -815,7 +815,7 @@ def least_loaded_node(self):
         Returns:
             node_id or None if no suitable node was found
         """
-        nodes = [broker.node_id for broker in self.cluster.brokers()]
+        nodes = [broker.node_id for broker in self.cluster.brokers() or self.cluster.bootstrap_brokers()]
         random.shuffle(nodes)
 
         inflight = float('inf')
@@ -851,7 +851,7 @@ def least_loaded_node_refresh_ms(self):
         Returns:
            float: delay_ms
         """
-        return min([self._refresh_delay_ms(broker.node_id) for broker in self.cluster.brokers()])
+        return min([self._refresh_delay_ms(broker.node_id) for broker in self.cluster.brokers() or self.cluster.bootstrap_brokers()])
 
     def _next_ifr_request_timeout_ms(self):
         if self._conns:

kafka/cluster.py

@@ -133,7 +133,7 @@ def brokers(self):
         Returns:
             set: {MetadataResponseBroker, ...}
         """
-        return list(self._brokers.values()) or list(self._bootstrap_brokers.values())
+        return list(self._brokers.values())
 
     def bootstrap_brokers(self):
         return list(self._bootstrap_brokers.values())

test/integration/test_ssl_integration.py

@@ -45,6 +45,5 @@ def test_kafka_client_ssl(self, ssl_kafka):
             ssl_cafile=os.path.join(ssl_kafka.ssl_dir, 'ca-cert'),
             ssl_check_hostname=False,
         )
-        client.poll(timeout_ms=5000)
-        assert client.cluster.brokers()
+        assert client.broker_version_data
         client.close()