private synchronized boolean ensureCoordinatorReady(final Timer timer, boolean disableWakeup) {
        if (!coordinatorUnknown())
            return true;

        long attempts = 0L;
        do {
            if (fatalFindCoordinatorException != null) {
                final RuntimeException fatalException = fatalFindCoordinatorException;
                fatalFindCoordinatorException = null;
                throw fatalException;
            }
            final RequestFuture<Void> future = lookupCoordinator();
            client.poll(future, timer, disableWakeup);

            if (!future.isDone()) {
                // ran out of time
                break;
            }

            RuntimeException fatalException = null;

            if (future.failed()) {
                if (future.isRetriable()) {
                    log.debug("Coordinator discovery failed, refreshing metadata", future.exception());
                    timer.sleep(retryBackoff.backoff(attempts++));
                    client.awaitMetadataUpdate(timer);
                } else {
                    fatalException = future.exception();
                    log.info("FindCoordinator request hit fatal exception", fatalException);
                }
            } else if (coordinator != null && client.isUnavailable(coordinator)) {
                // we found the coordinator, but the connection has failed, so mark
                // it dead and backoff before retrying discovery
                markCoordinatorUnknown("coordinator unavailable");
                timer.sleep(retryBackoff.backoff(attempts++));
            }

            clearFindCoordinatorFuture();
            if (fatalException != null)
                throw fatalException;
        } while (coordinatorUnknown() && timer.notExpired());

        return !coordinatorUnknown();
    }
Ensure that the coordinator is ready to receive requests. This will return
immediately without blocking. It is intended to be called in an asynchronous
context when wakeups are not expected.

@return true If coordinator discovery and initial connection succeeded, false otherwise
clients/src/main/java/org/apache/kafka/clients/consumer/internals/AbstractCoordinator.java