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.

def _send_request_to_node(self, node_id, request):
        """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.
        """
        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