docs: def ensure_active_group(self):
          """Ensure that the group is active (i.e. joined and synced)"""
          with self._client._lock, self._lock:
              if self._heartbeat_thread is None:
                  self._start_heartbeat_thread()
  
              while self.need_rejoin() or self._rejoin_incomplete():
                  self.ensure_coordinator_ready()
  
                  # call on_join_prepare if needed. We set a flag
                  # to make sure that we do not call it a second
                  # time if the client is woken up before a pending
                  # rebalance completes. This must be called on each
                  # iteration of the loop because an event requiring
                  # a rebalance (such as a metadata refresh which
                  # changes the matched subscription set) can occur
                  # while another rebalance is still in progress.
                  if not self.rejoining:
                      self._on_join_prepare(self._generation.generation_id,
                                            self._generation.member_id)
                      self.rejoining = True
  
                  # ensure that there are no pending requests to the coordinator.
                  # This is important in particular to avoid resending a pending
                  # JoinGroup request.
                  while not self.coordinator_unknown():
                      if not self._client.in_flight_request_count(self.coordinator_id):
                          break
                      self._client.poll()
                  else:
                      continue
  
                  # we store the join future in case we are woken up by the user
                  # after beginning the rebalance in the call to poll below.
                  # This ensures that we do not mistakenly attempt to rejoin
                  # before the pending rebalance has completed.
                  if self.join_future is None:
                      # Fence off the heartbeat thread explicitly so that it cannot
                      # interfere with the join group. Note that this must come after
                      # the call to _on_join_prepare since we must be able to continue
                      # sending heartbeats if that callback takes some time.
                      self._heartbeat_thread.disable()
  
                      self.state = MemberState.REBALANCING
                      future = self._send_join_group_request()
  
                      self.join_future = future  # this should happen before adding callbacks
  
                      # handle join completion in the callback so that the
                      # callback will be invoked even if the consumer is woken up
                      # before finishing the rebalance
                      future.add_callback(self._handle_join_success)
  
                      # we handle failures below after the request finishes.
                      # If the join completes after having been woken up, the
                      # exception is ignored and we will rejoin
                      future.add_errback(self._handle_join_failure)
  
                  else:
                      future = self.join_future
  
                  self._client.poll(future=future)
  
                  if future.succeeded():
                      self._on_join_complete(self._generation.generation_id,
                                             self._generation.member_id,
                                             self._generation.protocol,
                                             future.value)
                      self.join_future = None
                      self.rejoining = False
  
                  else:
                      self.join_future = None
                      exception = future.exception
                      if isinstance(exception, (Errors.UnknownMemberIdError,
                                                Errors.RebalanceInProgressError,
                                                Errors.IllegalGenerationError)):
                          continue
                      elif not future.retriable():
                          raise exception  # pylint: disable-msg=raising-bad-type
                      time.sleep(self.config['retry_backoff_ms'] / 1000)