Ensure that the group is active (i.e. joined and synced)

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)