def ensure_coordinator_ready(self, timeout_ms=None):
        """Block until the coordinator for this group is known.

        Keyword Arguments:
            timeout_ms (numeric, optional): Maximum number of milliseconds to
                block waiting to find coordinator. Default: None.

        Returns: True is coordinator found before timeout_ms, else False
        """
        timer = Timer(timeout_ms)
        with self._client._lock, self._lock:
            while self.coordinator_unknown():

                # Prior to 0.8.2 there was no group coordinator
                # so we will just pick a node at random and treat
                # it as the "coordinator"
                if self.config['api_version'] < (0, 8, 2):
                    maybe_coordinator_id = self._client.least_loaded_node()
                    if maybe_coordinator_id is None or self._client.cluster.is_bootstrap(maybe_coordinator_id):
                        future = Future().failure(Errors.NoBrokersAvailable())
                    else:
                        self.coordinator_id = maybe_coordinator_id
                        self._client.maybe_connect(self.coordinator_id)
                        if timer.expired:
                            return False
                        else:
                            continue
                else:
                    future = self.lookup_coordinator()

                self._client.poll(future=future, timeout_ms=timer.timeout_ms)

                if not future.is_done:
                    return False

                if future.failed():
                    if future.retriable():
                        if getattr(future.exception, 'invalid_metadata', False):
                            log.debug('Requesting metadata for group coordinator request: %s', future.exception)
                            metadata_update = self._client.cluster.request_update()
                            self._client.poll(future=metadata_update, timeout_ms=timer.timeout_ms)
                            if not metadata_update.is_done:
                                return False
                        else:
                            if timeout_ms is None or timer.timeout_ms > self.config['retry_backoff_ms']:
                                time.sleep(self.config['retry_backoff_ms'] / 1000)
                            else:
                                time.sleep(timer.timeout_ms / 1000)
                    else:
                        raise future.exception  # pylint: disable-msg=raising-bad-type
                if timer.expired:
                    return False
            else:
                return True
Block until the coordinator for this group is known.

        Keyword Arguments:
            timeout_ms (numeric, optional): Maximum number of milliseconds to
                block waiting to find coordinator. Default: None.

        Returns: True is coordinator found before timeout_ms, else False
        
kafka/coordinator/base.py