private long pollResigned(long currentTimeMs) {
        ResignedState state = quorum.resignedStateOrThrow();
        long endQuorumBackoffMs = maybeSendRequests(
            currentTimeMs,
            partitionState
                .lastVoterSet()
                .voterNodes(state.unackedVoters().stream(), channel.listenerName()),
            () -> buildEndQuorumEpochRequest(state)
        );

        GracefulShutdown shutdown = this.shutdown.get();
        final long stateTimeoutMs;
        if (shutdown != null) {
            // If we are shutting down, then we will remain in the resigned state
            // until either the shutdown expires or an election bumps the epoch
            stateTimeoutMs = shutdown.remainingTimeMs();
        } else if (state.hasElectionTimeoutExpired(currentTimeMs)) {
            if (quorum.isVoter()) {
                transitionToCandidate(currentTimeMs);
            } else {
                // It is posible that the old leader is not a voter in the new voter set.
                // In that case increase the epoch and transition to unattached. The epoch needs
                // to be increased to avoid FETCH responses with the leader being this replica.
                transitionToUnattached(quorum.epoch() + 1);
            }
            stateTimeoutMs = 0L;
        } else {
            stateTimeoutMs = state.remainingElectionTimeMs(currentTimeMs);
        }

        return Math.min(stateTimeoutMs, endQuorumBackoffMs);
    }
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