NetworkClientDelegate.UnsentRequest makeFindCoordinatorRequest(final long currentTimeMs) {
        coordinatorRequestState.onSendAttempt(currentTimeMs);
        FindCoordinatorRequestData data = new FindCoordinatorRequestData()
                .setKeyType(FindCoordinatorRequest.CoordinatorType.GROUP.id())
                .setKey(this.groupId);
        NetworkClientDelegate.UnsentRequest unsentRequest = new NetworkClientDelegate.UnsentRequest(
            new FindCoordinatorRequest.Builder(data),
            Optional.empty()
        );

        return unsentRequest.whenComplete((clientResponse, throwable) -> {
            if (clientResponse != null) {
                FindCoordinatorResponse response = (FindCoordinatorResponse) clientResponse.responseBody();
                onResponse(clientResponse.receivedTimeMs(), response);
            } else {
                onFailedResponse(unsentRequest.handler().completionTimeMs(), throwable);
            }
        });
    }
Poll for the FindCoordinator request.
If we don't need to discover a coordinator, this method will return a PollResult with Long.MAX_VALUE backoff time and an empty list.
If we are still backing off from a previous attempt, this method will return a PollResult with the remaining backoff time and an empty list.
Otherwise, this returns will return a PollResult with a singleton list of UnsentRequest and Long.MAX_VALUE backoff time.
Note that this method does not involve any actual network IO, and it only determines if we need to send a new request or not.

@param currentTimeMs current time in ms.
@return {@link NetworkClientDelegate.PollResult}. This will not be {@code null}.
clients/src/main/java/org/apache/kafka/clients/consumer/internals/CoordinatorRequestManager.java