private void onFailure(final long currentTimeMs,
                               final Errors responseError) {
            log.debug("Offset fetch failed: {}", responseError.message());
            onFailedAttempt(currentTimeMs);
            if (responseError == COORDINATOR_LOAD_IN_PROGRESS) {
                future.completeExceptionally(responseError.exception());
            } else if (responseError == Errors.UNKNOWN_MEMBER_ID) {
                log.error("OffsetFetch failed with {} because the member is not part of the group" +
                    " anymore.", responseError);
                future.completeExceptionally(responseError.exception());
            } else if (responseError == Errors.STALE_MEMBER_EPOCH) {
                log.error("OffsetFetch failed with {} and the consumer is not part " +
                    "of the group anymore (it probably left the group, got fenced" +
                    " or failed). The request cannot be retried and will fail.", responseError);
                future.completeExceptionally(responseError.exception());
            } else if (responseError == Errors.NOT_COORDINATOR || responseError == Errors.COORDINATOR_NOT_AVAILABLE) {
                // Re-discover the coordinator and retry
                coordinatorRequestManager.markCoordinatorUnknown("error response " + responseError.name(), currentTimeMs);
                future.completeExceptionally(responseError.exception());
            } else if (responseError == Errors.GROUP_AUTHORIZATION_FAILED) {
                future.completeExceptionally(GroupAuthorizationException.forGroupId(groupId));
            } else {
                // Fail with a non-retriable KafkaException for all unexpected errors (even if
                // they are retriable)
                future.completeExceptionally(new KafkaException("Unexpected error in fetch offset response: " + responseError.message()));
            }
        }
Handle failed responses. This will retry if the error is retriable, or complete the
result future exceptionally in the case of non-recoverable or unexpected errors.
clients/src/main/java/org/apache/kafka/clients/consumer/internals/CommitRequestManager.java