def ensure_coordinator_ready(self):
        """Block until the coordinator for this group is known
        (and we have an active connection -- java client uses unsent queue).
        """
        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):
                    self.coordinator_id = self._client.least_loaded_node()
                    if self.coordinator_id is not None:
                        self._client.maybe_connect(self.coordinator_id)
                    continue

                future = self.lookup_coordinator()
                self._client.poll(future=future)

                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)
                        else:
                            time.sleep(self.config['retry_backoff_ms'] / 1000)
                    else:
                        raise future.exception