def await_ready(self, node_id, timeout_ms=30000):
        """
        Invokes `poll` to discard pending disconnects, followed by `client.ready` and 0 or more `client.poll`
        invocations until the connection to `node` is ready, the timeoutMs expires or the connection fails.

        It returns `true` if the call completes normally or `false` if the timeoutMs expires. If the connection fails,
        an `IOException` is thrown instead. Note that if the `NetworkClient` has been configured with a positive
        connection timeoutMs, it is possible for this method to raise an `IOException` for a previous connection which
        has recently disconnected.

        This method is useful for implementing blocking behaviour on top of the non-blocking `NetworkClient`, use it with
        care.
        """
        timer = Timer(timeout_ms)
        self.poll(timeout_ms=0)
        if self.is_ready(node_id):
            return True

        while not self.is_ready(node_id) and not timer.expired:
            if self.connection_failed(node_id):
                raise Errors.KafkaConnectionError("Connection to %s failed." % (node_id,))
            self.maybe_connect(node_id)
            self.poll(timeout_ms=timer.timeout_ms)
        return self.is_ready(node_id)

        Invokes `poll` to discard pending disconnects, followed by `client.ready` and 0 or more `client.poll`
        invocations until the connection to `node` is ready, the timeoutMs expires or the connection fails.

        It returns `true` if the call completes normally or `false` if the timeoutMs expires. If the connection fails,
        an `IOException` is thrown instead. Note that if the `NetworkClient` has been configured with a positive
        connection timeoutMs, it is possible for this method to raise an `IOException` for a previous connection which
        has recently disconnected.

        This method is useful for implementing blocking behaviour on top of the non-blocking `NetworkClient`, use it with
        care.
        
kafka/client_async.py