code: def _send_request_to_node(self, node_id, request):
          
          while not self._client.ready(node_id):
              # poll until the connection to broker is ready, otherwise send()
              # will fail with NodeNotReadyError
              self._client.poll()
          future = self._client.send(node_id, request)
          self._client.poll(future=future)
          if future.succeeded():
              return future.value
          else:
              raise future.exception
docs: Send a Kafka protocol message to a specific broker.
  
          Will block until the message result is received.
  
          :param node_id: The broker id to which to send the message.
          :param request: The message to send.
          :return: The Kafka protocol response for the message.
          :exception: The exception if the message could not be sent.