private BeginQuorumEpochResponseData handleBeginQuorumEpochRequest(
        RaftRequest.Inbound requestMetadata,
        long currentTimeMs
    ) {
        BeginQuorumEpochRequestData request = (BeginQuorumEpochRequestData) requestMetadata.data();

        if (!hasValidClusterId(request.clusterId())) {
            return new BeginQuorumEpochResponseData().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 BeginQuorumEpochResponseData().setErrorCode(Errors.INVALID_REQUEST.code());
        }

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

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

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

        // Compute the leader's endpoint from the request or the voter set
        final Endpoints leaderEndpoints;
        if (request.leaderEndpoints().isEmpty()) {
            leaderEndpoints = partitionState.lastVoterSet().listeners(requestLeaderId);
        } else {
            leaderEndpoints = Endpoints.fromBeginQuorumEpochRequest(request.leaderEndpoints());
        }

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

        // Check that the request was intended for this replica
        Optional<ReplicaKey> voterKey = RaftUtil.beginQuorumEpochRequestVoterKey(request, partitionRequest);
        if (!isValidVoterKey(voterKey)) {
            logger.info(
                "Leader sent a voter key ({}) in the BEGIN_QUORUM_EPOCH request that doesn't " +
                "match the local key ({}, {}); returning INVALID_VOTER_KEY",
                voterKey,
                nodeId,
                nodeDirectoryId
            );
            // The request is not intended to this replica since the replica keys don't match
            return buildBeginQuorumEpochResponse(
                requestMetadata.listenerName(),
                requestMetadata.apiVersion(),
                Errors.INVALID_VOTER_KEY
            );
        }

        return buildBeginQuorumEpochResponse(
            requestMetadata.listenerName(),
            requestMetadata.apiVersion(),
            Errors.NONE
        );
    }
Handle a BeginEpoch 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