public void ensureActiveGroup() {
    // always ensure that the coordinator is ready because we may have been disconnected
    // when sending heartbeats and does not necessarily require us to rejoin the group.
    ensureCoordinatorReady();
    startHeartbeatThreadIfNeeded();
    joinGroupIfNeeded();
}

private synchronized void startHeartbeatThreadIfNeeded() {
    if (heartbeatThread == null) {
        heartbeatThread = new HeartbeatThread();
        heartbeatThread.start();
    }
}

// visible for testing
synchronized RequestFuture<Void> sendHeartbeatRequest() {
    log.debug("Sending Heartbeat request to coordinator {}", coordinator);
    HeartbeatRequest.Builder requestBuilder =
            new HeartbeatRequest.Builder(this.groupId, this.generation.generationId, this.generation.memberId);
    return client.send(coordinator, requestBuilder)
            .compose(new HeartbeatResponseHandler());
}

else if (!heartbeat.shouldHeartbeat(now)) {
    // poll again after waiting for the retry backoff in case the heartbeat failed or the
    // coordinator disconnected
    AbstractCoordinator.this.wait(retryBackoffMs);
}