private void onResponse(
        final long currentTimeMs,
        final FindCoordinatorResponse response
    ) {
        // handles Runtime exception
        Optional<FindCoordinatorResponseData.Coordinator> coordinator = response.coordinatorByKey(this.groupId);
        if (!coordinator.isPresent()) {
            String msg = String.format("Response did not contain expected coordinator section for groupId: %s", this.groupId);
            onFailedResponse(currentTimeMs, new IllegalStateException(msg));
            return;
        }

        FindCoordinatorResponseData.Coordinator node = coordinator.get();
        if (node.errorCode() != Errors.NONE.code()) {
            onFailedResponse(currentTimeMs, Errors.forCode(node.errorCode()).exception());
            return;
        }
        onSuccessfulResponse(currentTimeMs, node);
    }
Handles the response upon completing the {@link FindCoordinatorRequest} if the
future returned successfully. This method must still unwrap the response object
to check for protocol errors.

@param currentTimeMs current time in ms.
@param response      the response for finding the coordinator. null if an exception is thrown.
clients/src/main/java/org/apache/kafka/clients/consumer/internals/CoordinatorRequestManager.java