docs: 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