private void closeHeartbeatThread() {
        HeartbeatThread thread;
        synchronized (this) {
            if (heartbeatThread == null)
                return;
            heartbeatThread.close();
            thread = heartbeatThread;
            heartbeatThread = null;
        }
        try {
            thread.join();
        } catch (InterruptedException e) {
            log.warn("Interrupted while waiting for consumer heartbeat thread to close");
            throw new InterruptException(e);
        }
    }
Ensure the group is active (i.e., joined and synced)

@param timer Timer bounding how long this method can block
@throws KafkaException if the callback throws exception
@return true iff the group is active
clients/src/main/java/org/apache/kafka/clients/consumer/internals/AbstractCoordinator.java