private EndQuorumEpochResponseData handleEndQuorumEpochRequest(
        RaftRequest.Inbound requestMetadata,
        long currentTimeMs
    ) {
        EndQuorumEpochRequestData request = (EndQuorumEpochRequestData) requestMetadata.data();

        if (!hasValidClusterId(request.clusterId())) {
            return new EndQuorumEpochResponseData().setErrorCode(Errors.INCONSISTENT_CLUSTER_ID.code());
        }

        if (!hasValidTopicPartition(request, log.topicPartition())) {
            // Until we support multi-raft, we treat topic partition mismatches as invalid requests
            return new EndQuorumEpochResponseData().setErrorCode(Errors.INVALID_REQUEST.code());
        }

        EndQuorumEpochRequestData.PartitionData partitionRequest =
            request.topics().get(0).partitions().get(0);

        int requestEpoch = partitionRequest.leaderEpoch();
        int requestLeaderId = partitionRequest.leaderId();

        Optional<Errors> errorOpt = validateVoterOnlyRequest(requestLeaderId, requestEpoch);
        if (errorOpt.isPresent()) {
            return buildEndQuorumEpochResponse(
                requestMetadata.listenerName(),
                requestMetadata.apiVersion(),
                errorOpt.get()
            );
        }

        final Endpoints leaderEndpoints;
        if (request.leaderEndpoints().isEmpty()) {
            leaderEndpoints = partitionState.lastVoterSet().listeners(requestLeaderId);
        } else {
            leaderEndpoints = Endpoints.fromEndQuorumEpochRequest(request.leaderEndpoints());
        }

        maybeTransition(
            OptionalInt.of(requestLeaderId),
            requestEpoch,
            leaderEndpoints,
            currentTimeMs
        );

        if (quorum.isFollower()) {
            FollowerState state = quorum.followerStateOrThrow();
            if (state.leaderId() == requestLeaderId) {
                List<ReplicaKey> preferredCandidates = EndQuorumEpochRequest
                    .preferredCandidates(partitionRequest)
                    .stream()
                    .map(replica -> ReplicaKey.of(replica.candidateId(), replica.candidateDirectoryId()))
                    .collect(Collectors.toList());
                long electionBackoffMs = endEpochElectionBackoff(preferredCandidates);
                logger.debug(
                    "Overriding follower fetch timeout to {} after receiving EndQuorumEpoch " +
                    "request from leader {} in epoch {}",
                    electionBackoffMs,
                    requestLeaderId,
                    requestEpoch
                );
                state.overrideFetchTimeout(currentTimeMs, electionBackoffMs);
            }
        }
        return buildEndQuorumEpochResponse(
            requestMetadata.listenerName(),
            requestMetadata.apiVersion(),
            Errors.NONE
        );
    }
Handle an EndEpoch request. This API may return the following errors:

- {@link Errors#INCONSISTENT_CLUSTER_ID} if the cluster id is presented in request
     but different from this node
- {@link Errors#BROKER_NOT_AVAILABLE} if this node is currently shutting down
- {@link Errors#FENCED_LEADER_EPOCH} if the epoch is smaller than this node's epoch
raft/src/main/java/org/apache/kafka/raft/KafkaRaftClient.java