bugfix: fix infinite loop on KafkaAdminClient (dpkp#2194)

An infinite loop may happen with the following pattern:

    self._send_request_to_node(self._client.least_loaded_node(), request)

The problem happens when `self._client`'s cluster metadata is out-of-date, and the
result of `least_loaded_node()` is a node that has been removed from the cluster but
the client is unware of it. When this happens `_send_request_to_node` will enter an
infinite loop waiting for the chosen node to become available, which won't happen,
resulting in an infinite loop.

This commit introduces a new method named `_send_request_to_least_loaded_node` which
handles the case above. This is done by regularly checking if the target node is
available in the cluster metadata, and if not, a new node is chosen.

Notes:

- This does not yet cover every call site to `_send_request_to_node`, there are some
  other places were similar race conditions may happen.
- The code above does not guarantee that the request itself will be sucessful, since
  it is still possible for the target node to exit, however, it does remove the
  infinite loop which can render client code unusable.

Author: hackaugusto

kafka/admin/client.py

@@ -274,7 +274,7 @@ def _refresh_controller_id(self):
         version = self._matching_api_version(MetadataRequest)
         if 1 <= version <= 6:
             request = MetadataRequest[version]()
-            future = self._send_request_to_node(self._client.least_loaded_node(), request)
+            future = self._send_request_to_least_loaded_node(request)
 
             self._wait_for_futures([future])
 
@@ -312,7 +312,7 @@ def _find_coordinator_id_send_request(self, group_id):
             raise NotImplementedError(
                 "Support for GroupCoordinatorRequest_v{} has not yet been added to KafkaAdminClient."
                 .format(version))
-        return self._send_request_to_node(self._client.least_loaded_node(), request)
+        return self._send_request_to_least_loaded_node(request)
 
     def _find_coordinator_id_process_response(self, response):
         """Process a FindCoordinatorResponse.
@@ -357,9 +357,36 @@ def _find_coordinator_ids(self, group_ids):
         }
         return groups_coordinators
 
-    def _send_request_to_node(self, node_id, request, wakeup=True):
+    def _send_request_to_least_loaded_node(self, request):
+        """Send a Kafka protocol message to the least loaded broker.
+
+        Returns a future that may be polled for status and results.
+
+        :param request: The message to send.
+        :return: A future object that may be polled for status and results.
+        :exception: The exception if the message could not be sent.
+        """
+        node_id = self._client.least_loaded_node()
+        while not self._client.ready(node_id):
+            # poll until the connection to broker is ready, otherwise send()
+            # will fail with NodeNotReadyError
+            self._client.poll()
+
+            # node_id is not part of the cluster anymore, choose a new broker
+            # to connect to
+            if self._client.cluster.broker_metadata(node_id) is None:
+                node_id = self._client.least_loaded_node()
+
+        return self._client.send(node_id, request)
+
+    def _send_request_to_node(self, node_id, request):
         """Send a Kafka protocol message to a specific broker.
 
+        .. note::
+
+            This function will enter in an infinite loop if `node_id` is
+            removed from the cluster.
+
         Returns a future that may be polled for status and results.
 
         :param node_id: The broker id to which to send the message.
@@ -509,10 +536,7 @@ def _get_cluster_metadata(self, topics=None, auto_topic_creation=False):
                 allow_auto_topic_creation=auto_topic_creation
             )
 
-        future = self._send_request_to_node(
-            self._client.least_loaded_node(),
-            request
-        )
+        future = self._send_request_to_least_loaded_node(request)
         self._wait_for_futures([future])
         return future.value
 
@@ -604,7 +628,7 @@ def describe_acls(self, acl_filter):
                     .format(version)
             )
 
-        future = self._send_request_to_node(self._client.least_loaded_node(), request)
+        future = self._send_request_to_least_loaded_node(request)
         self._wait_for_futures([future])
         response = future.value
 
@@ -695,7 +719,7 @@ def create_acls(self, acls):
                     .format(version)
             )
 
-        future = self._send_request_to_node(self._client.least_loaded_node(), request)
+        future = self._send_request_to_least_loaded_node(request)
         self._wait_for_futures([future])
         response = future.value
 
@@ -789,7 +813,7 @@ def delete_acls(self, acl_filters):
                     .format(version)
             )
 
-        future = self._send_request_to_node(self._client.least_loaded_node(), request)
+        future = self._send_request_to_least_loaded_node(request)
         self._wait_for_futures([future])
         response = future.value
 
@@ -849,8 +873,7 @@ def describe_configs(self, config_resources, include_synonyms=False):
                     ))
 
             if len(topic_resources) > 0:
-                futures.append(self._send_request_to_node(
-                    self._client.least_loaded_node(),
+                futures.append(self._send_request_to_least_loaded_node(
                     DescribeConfigsRequest[version](resources=topic_resources)
                 ))
 
@@ -870,8 +893,7 @@ def describe_configs(self, config_resources, include_synonyms=False):
                     ))
 
             if len(topic_resources) > 0:
-                futures.append(self._send_request_to_node(
-                    self._client.least_loaded_node(),
+                futures.append(self._send_request_to_least_loaded_node(
                     DescribeConfigsRequest[version](resources=topic_resources, include_synonyms=include_synonyms)
                 ))
         else:
@@ -918,7 +940,7 @@ def alter_configs(self, config_resources):
         # // a single request that may be sent to any broker.
         #
         # So this is currently broken as it always sends to the least_loaded_node()
-        future = self._send_request_to_node(self._client.least_loaded_node(), request)
+        future = self._send_request_to_least_loaded_node(request)
 
         self._wait_for_futures([future])
         response = future.value