private long pollLeader(long currentTimeMs) {
        LeaderState<T> state = quorum.leaderStateOrThrow();
        maybeFireLeaderChange(state);

        long timeUntilCheckQuorumExpires = state.timeUntilCheckQuorumExpires(currentTimeMs);
        if (shutdown.get() != null || state.isResignRequested() || timeUntilCheckQuorumExpires == 0) {
            transitionToResigned(state.nonLeaderVotersByDescendingFetchOffset());
            return 0L;
        }

        long timeUtilVoterChangeExpires = state.maybeExpirePendingOperation(currentTimeMs);

        long timeUntilFlush = maybeAppendBatches(
            state,
            currentTimeMs
        );

        long timeUntilNextBeginQuorumSend = maybeSendBeginQuorumEpochRequests(
            state,
            currentTimeMs
        );

        return Math.min(
            timeUntilFlush,
            Math.min(
                timeUntilNextBeginQuorumSend,
                Math.min(
                    timeUntilCheckQuorumExpires,
                    timeUtilVoterChangeExpires
                )
            )
        );
    }
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