private long pollFollowerAsObserver(FollowerState state, long currentTimeMs) {
        if (state.hasFetchTimeoutExpired(currentTimeMs)) {
            return maybeSendAnyVoterFetch(currentTimeMs);
        } else {
            final long backoffMs;

            // If the current leader is backing off due to some failure or if the
            // request has timed out, then we attempt to send the Fetch to another
            // voter in order to discover if there has been a leader change.
            Node leaderNode = state.leaderNode(channel.listenerName());
            if (requestManager.hasRequestTimedOut(leaderNode, currentTimeMs)) {
                // Once the request has timed out backoff the connection
                requestManager.reset(leaderNode);
                backoffMs = maybeSendAnyVoterFetch(currentTimeMs);
            } else if (requestManager.isBackingOff(leaderNode, currentTimeMs)) {
                backoffMs = maybeSendAnyVoterFetch(currentTimeMs);
            } else if (!requestManager.hasAnyInflightRequest(currentTimeMs)) {
                backoffMs = maybeSendFetchOrFetchSnapshot(state, currentTimeMs);
            } else {
                backoffMs = requestManager.backoffBeforeAvailableBootstrapServer(currentTimeMs);
            }

            return Math.min(backoffMs, state.remainingFetchTimeMs(currentTimeMs));
        }
    }
Attempt to send a request. Return the time to wait before the request can be retried.
raft/src/main/java/org/apache/kafka/raft/KafkaRaftClient.java