public <T> LeaderState<T> transitionToLeader(long epochStartOffset, BatchAccumulator<T> accumulator) {
        if (isObserver()) {
            throw new IllegalStateException(
                String.format(
                    "Cannot transition to Leader since the local id (%s) and directory id (%s) " +
                    "is not one of the voters %s",
                    localId,
                    localDirectoryId,
                    partitionState.lastVoterSet()
                )
            );
        } else if (!isCandidate()) {
            throw new IllegalStateException("Cannot transition to Leader from current state " + state);
        }

        CandidateState candidateState = candidateStateOrThrow();
        if (!candidateState.isVoteGranted())
            throw new IllegalStateException("Cannot become leader without majority votes granted");

        // Note that the leader does not retain the high watermark that was known
        // in the previous state. The reason for this is to protect the monotonicity
        // of the global high watermark, which is exposed through the leader. The
        // only way a new leader can be sure that the high watermark is increasing
        // monotonically is to wait until a majority of the voters have reached the
        // starting offset of the new epoch. The downside of this is that the local
        // state machine is temporarily stalled by the advancement of the global
        // high watermark even though it only depends on local monotonicity. We
        // could address this problem by decoupling the local high watermark, but
        // we typically expect the state machine to be caught up anyway.

        LeaderState<T> state = new LeaderState<>(
            time,
            ReplicaKey.of(localIdOrThrow(), localDirectoryId),
            epoch(),
            epochStartOffset,
            partitionState.lastVoterSet(),
            partitionState.lastVoterSetOffset(),
            partitionState.lastKraftVersion(),
            candidateState.grantingVoters(),
            accumulator,
            localListeners,
            fetchTimeoutMs,
            logContext
        );
        durableTransitionTo(state);
        return state;
    }
Become a follower of an elected leader so that we can begin fetching.
raft/src/main/java/org/apache/kafka/raft/QuorumState.java