boolean joinGroupIfNeeded(final Timer timer) {
        while (rejoinNeededOrPending()) {
            if (!ensureCoordinatorReady(timer)) {
                return false;
            }

            // call onJoinPrepare 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 (needsJoinPrepare) {
                // need to set the flag before calling onJoinPrepare since the user callback may throw
                // exception, in which case upon retry we should not retry onJoinPrepare either.
                needsJoinPrepare = false;
                // return false when onJoinPrepare is waiting for committing offset
                if (!onJoinPrepare(timer, generation.generationId, generation.memberId)) {
                    needsJoinPrepare = true;
                    //should not initiateJoinGroup if needsJoinPrepare still is true
                    return false;
                }
            }

            final RequestFuture<ByteBuffer> future = initiateJoinGroup();
            client.poll(future, timer);
            if (!future.isDone()) {
                // we ran out of time
                return false;
            }

            if (future.succeeded()) {
                Generation generationSnapshot;
                MemberState stateSnapshot;

                // Generation data maybe concurrently cleared by Heartbeat thread.
                // Can't use synchronized for {@code onJoinComplete}, because it can be long enough
                // and shouldn't block heartbeat thread.
                // See {@link PlaintextConsumerTest#testMaxPollIntervalMsDelayInAssignment}
                synchronized (AbstractCoordinator.this) {
                    generationSnapshot = this.generation;
                    stateSnapshot = this.state;
                }

                if (!hasGenerationReset(generationSnapshot) && stateSnapshot == MemberState.STABLE) {
                    // Duplicate the buffer in case `onJoinComplete` does not complete and needs to be retried.
                    ByteBuffer memberAssignment = future.value().duplicate();

                    onJoinComplete(generationSnapshot.generationId, generationSnapshot.memberId, generationSnapshot.protocolName, memberAssignment);

                    // Generally speaking we should always resetJoinGroupFuture once the future is done, but here
                    // we can only reset the join group future after the completion callback returns. This ensures
                    // that if the callback is woken up, we will retry it on the next joinGroupIfNeeded.
                    // And because of that we should explicitly trigger resetJoinGroupFuture in other conditions below.
                    resetJoinGroupFuture();
                    needsJoinPrepare = true;
                } else {
                    final String reason = String.format("rebalance failed since the generation/state was " +
                            "modified by heartbeat thread to %s/%s before the rebalance callback triggered",
                            generationSnapshot, stateSnapshot);

                    resetStateAndRejoin(reason, true);
                    resetJoinGroupFuture();
                }
            } else {
                final RuntimeException exception = future.exception();

                resetJoinGroupFuture();
                synchronized (AbstractCoordinator.this) {
                    final String simpleName = exception.getClass().getSimpleName();
                    final String shortReason = String.format("rebalance failed due to %s", simpleName);
                    final String fullReason = String.format("rebalance failed due to '%s' (%s)",
                        exception.getMessage(),
                        simpleName);
                    // Don't need to request rejoin again for MemberIdRequiredException since we've done that in JoinGroupResponseHandler
                    if (!(exception instanceof MemberIdRequiredException)) {
                        requestRejoin(shortReason, fullReason);
                    }
                }

                if (exception instanceof UnknownMemberIdException ||
                        exception instanceof IllegalGenerationException ||
                        exception instanceof RebalanceInProgressException ||
                        exception instanceof MemberIdRequiredException)
                    continue;
                else if (!future.isRetriable())
                    throw exception;

                // We need to return upon expired timer, in case if the client.poll returns immediately and the time
                // has elapsed.
                if (timer.isExpired()) {
                    return false;
                }

                timer.sleep(rebalanceConfig.retryBackoffMs);
            }
        }
        return true;
    }
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