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

def ensure_active_group(self):

    if (not self.need_rejoin()):
        return
    if (not self.rejoining):
        self._on_join_prepare(self.generation, self.member_id)
        self.rejoining = True
    while self.need_rejoin():
        self.ensure_coordinator_known()
        while (not self.coordinator_unknown()):
            if (not self._client.in_flight_request_count(self.coordinator_id)):
                break
            self._client.poll(delayed_tasks=False)
        else:
            continue
        future = self._send_join_group_request()
        self._client.poll(future=future)
        if future.succeeded():
            member_assignment_bytes = future.value
            self._on_join_complete(self.generation, self.member_id, self.protocol, member_assignment_bytes)
            self.rejoining = False
            self.heartbeat_task.reset()
        else:
            assert future.failed()
            exception = future.exception
            if isinstance(exception, (Errors.UnknownMemberIdError, Errors.RebalanceInProgressError, Errors.IllegalGenerationError)):
                continue
            elif (not future.retriable()):
                raise exception
            time.sleep((self.config['retry_backoff_ms'] / 1000))
