public void initialize(OffsetAndEpoch logEndOffsetAndEpoch) throws IllegalStateException {
        // We initialize in whatever state we were in on shutdown. If we were a leader
        // or candidate, probably an election was held, but we will find out about it
        // when we send Vote or BeginEpoch requests.
        ElectionState election = readElectionState();

        final EpochState initialState;
        if (election.hasVoted() && !localId.isPresent()) {
            throw new IllegalStateException(
                String.format(
                    "Initialized quorum state (%s) with a voted candidate but without a local id",
                    election
                )
            );
        } else if (election.epoch() < logEndOffsetAndEpoch.epoch()) {
            log.warn(
                "Epoch from quorum store file ({}) is {}, which is smaller than last written " +
                "epoch {} in the log",
                store.path(),
                election.epoch(),
                logEndOffsetAndEpoch.epoch()
            );
            initialState = new UnattachedState(
                time,
                logEndOffsetAndEpoch.epoch(),
                OptionalInt.empty(),
                partitionState.lastVoterSet().voterIds(),
                Optional.empty(),
                randomElectionTimeoutMs(),
                logContext
            );
        } else if (localId.isPresent() && election.isLeader(localId.getAsInt())) {
            // If we were previously a leader, then we will start out as resigned
            // in the same epoch. This serves two purposes:
            // 1. It ensures that we cannot vote for another leader in the same epoch.
            // 2. It protects the invariant that each record is uniquely identified by
            //    offset and epoch, which might otherwise be violated if unflushed data
            //    is lost after restarting.
            initialState = new ResignedState(
                time,
                localId.getAsInt(),
                election.epoch(),
                partitionState.lastVoterSet().voterIds(),
                randomElectionTimeoutMs(),
                Collections.emptyList(),
                localListeners,
                logContext
            );
        } else if (
            localId.isPresent() &&
            election.isVotedCandidate(ReplicaKey.of(localId.getAsInt(), localDirectoryId))
        ) {
            initialState = new CandidateState(
                time,
                localId.getAsInt(),
                localDirectoryId,
                election.epoch(),
                partitionState.lastVoterSet(),
                Optional.empty(),
                1,
                randomElectionTimeoutMs(),
                logContext
            );
        } else if (election.hasVoted()) {
            initialState = new VotedState(
                time,
                election.epoch(),
                election.votedKey(),
                partitionState.lastVoterSet().voterIds(),
                Optional.empty(),
                randomElectionTimeoutMs(),
                logContext
            );
        } else if (election.hasLeader()) {
            VoterSet voters = partitionState.lastVoterSet();
            Endpoints leaderEndpoints = voters.listeners(election.leaderId());
            if (leaderEndpoints.isEmpty()) {
                // Since the leader's endpoints are not known, it cannot send Fetch or
                // FetchSnapshot requests to the leader.
                //
                // Transition to unattached instead and discover the leader's endpoint through
                // Fetch requests to the bootstrap servers or from a BeginQuorumEpoch request from
                // the leader.
                log.info(
                    "The leader in election state {} is not a member of the latest voter set {}; " +
                    "transitioning to unattached instead of follower because the leader's " +
                    "endpoints are not known",
                    election,
                    voters
                );

                initialState = new UnattachedState(
                    time,
                    election.epoch(),
                    OptionalInt.of(election.leaderId()),
                    partitionState.lastVoterSet().voterIds(),
                    Optional.empty(),
                    randomElectionTimeoutMs(),
                    logContext
                );
            } else {
                initialState = new FollowerState(
                    time,
                    election.epoch(),
                    election.leaderId(),
                    leaderEndpoints,
                    voters.voterIds(),
                    Optional.empty(),
                    fetchTimeoutMs,
                    logContext
                );
            }
        } else {
            initialState = new UnattachedState(
                time,
                election.epoch(),
                OptionalInt.empty(),
                partitionState.lastVoterSet().voterIds(),
                Optional.empty(),
                randomElectionTimeoutMs(),
                logContext
            );
        }

        durableTransitionTo(initialState);
    }
This class is responsible for managing the current state of this node and ensuring
only valid state transitions. Below we define the possible state transitions and
how they are triggered:

Unattached|Resigned transitions to:
   Unattached: After learning of a new election with a higher epoch
   Voted: After granting a vote to a candidate
   Candidate: After expiration of the election timeout
   Follower: After discovering a leader with an equal or larger epoch

Voted transitions to:
   Unattached: After learning of a new election with a higher epoch
   Candidate: After expiration of the election timeout

Candidate transitions to:
   Unattached: After learning of a new election with a higher epoch
   Candidate: After expiration of the election timeout
   Leader: After receiving a majority of votes

Leader transitions to:
   Unattached: After learning of a new election with a higher epoch
   Resigned: When shutting down gracefully

Follower transitions to:
   Unattached: After learning of a new election with a higher epoch
   Candidate: After expiration of the fetch timeout
   Follower: After discovering a leader with a larger epoch

Observers follow a simpler state machine. The Voted/Candidate/Leader/Resigned
states are not possible for observers, so the only transitions that are possible
are between Unattached and Follower.

Unattached transitions to:
   Unattached: After learning of a new election with a higher epoch
   Follower: After discovering a leader with an equal or larger epoch

Follower transitions to:
   Unattached: After learning of a new election with a higher epoch
   Follower: After discovering a leader with a larger epoch
raft/src/main/java/org/apache/kafka/raft/QuorumState.java