private synchronized RequestFuture<ByteBuffer> initiateJoinGroup() {
        // 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 (joinFuture == null) {
            state = MemberState.PREPARING_REBALANCE;
            // a rebalance can be triggered consecutively if the previous one failed,
            // in this case we would not update the start time.
            if (lastRebalanceStartMs == -1L)
                lastRebalanceStartMs = time.milliseconds();
            joinFuture = sendJoinGroupRequest();
            joinFuture.addListener(new RequestFutureListener<ByteBuffer>() {
                @Override
                public void onSuccess(ByteBuffer value) {
                    // do nothing since all the handler logic are in SyncGroupResponseHandler already
                }

                @Override
                public void onFailure(RuntimeException e) {
                    // 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;
                    // this can be triggered when either join or sync request failed
                    synchronized (AbstractCoordinator.this) {
                        sensors.failedRebalanceSensor.record();
                    }
                }
            });
        }
        return joinFuture;
    }
Joins the group without starting the heartbeat thread.

If this function returns true, the state must always be in STABLE and heartbeat enabled.
If this function returns false, the state can be in one of the following:
 * UNJOINED: got error response but times out before being able to re-join, heartbeat disabled
 * PREPARING_REBALANCE: not yet received join-group response before timeout, heartbeat disabled
 * COMPLETING_REBALANCE: not yet received sync-group response before timeout, heartbeat enabled

Visible for testing.

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