private Errors validateAlterPartitionData(
        int brokerId,
        TopicControlInfo topic,
        int partitionId,
        PartitionRegistration partition,
        short requestApiVersion,
        AlterPartitionRequestData.PartitionData partitionData
    ) {
        if (partition == null) {
            log.info("Rejecting AlterPartition request for unknown partition {}-{}.",
                    topic.name, partitionId);

            return UNKNOWN_TOPIC_OR_PARTITION;
        }

        // If the partition leader has a higher leader/partition epoch, then it is likely
        // that this node is no longer the active controller. We return NOT_CONTROLLER in
        // this case to give the leader an opportunity to find the new controller.
        if (partitionData.leaderEpoch() > partition.leaderEpoch) {
            log.debug("Rejecting AlterPartition request from node {} for {}-{} because " +
                    "the current leader epoch is {}, which is greater than the local value {}.",
                brokerId, topic.name, partitionId, partition.leaderEpoch, partitionData.leaderEpoch());
            return NOT_CONTROLLER;
        }
        if (partitionData.partitionEpoch() > partition.partitionEpoch) {
            log.debug("Rejecting AlterPartition request from node {} for {}-{} because " +
                    "the current partition epoch is {}, which is greater than the local value {}.",
                brokerId, topic.name, partitionId, partition.partitionEpoch, partitionData.partitionEpoch());
            return NOT_CONTROLLER;
        }
        if (partitionData.leaderEpoch() < partition.leaderEpoch) {
            log.debug("Rejecting AlterPartition request from node {} for {}-{} because " +
                    "the current leader epoch is {}, not {}.", brokerId, topic.name,
                    partitionId, partition.leaderEpoch, partitionData.leaderEpoch());

            return FENCED_LEADER_EPOCH;
        }
        if (brokerId != partition.leader) {
            log.info("Rejecting AlterPartition request from node {} for {}-{} because " +
                    "the current leader is {}.", brokerId, topic.name,
                    partitionId, partition.leader);

            return INVALID_REQUEST;
        }
        if (partitionData.partitionEpoch() < partition.partitionEpoch) {
            log.info("Rejecting AlterPartition request from node {} for {}-{} because " +
                    "the current partition epoch is {}, not {}.", brokerId,
                    topic.name, partitionId, partition.partitionEpoch,
                    partitionData.partitionEpoch());

            return INVALID_UPDATE_VERSION;
        }

        int[] newIsr = partitionData.newIsrWithEpochs().stream()
            .mapToInt(BrokerState::brokerId).toArray();

        if (!Replicas.validateIsr(partition.replicas, newIsr)) {
            log.error("Rejecting AlterPartition request from node {} for {}-{} because " +
                    "it specified an invalid ISR {}.", brokerId,
                    topic.name, partitionId, partitionData.newIsrWithEpochs());

            return INVALID_REQUEST;
        }
        if (!Replicas.contains(newIsr, partition.leader)) {
            // The ISR must always include the current leader.
            log.error("Rejecting AlterPartition request from node {} for {}-{} because " +
                    "it specified an invalid ISR {} that doesn't include itself.",
                    brokerId, topic.name, partitionId, partitionData.newIsrWithEpochs());

            return INVALID_REQUEST;
        }
        LeaderRecoveryState leaderRecoveryState = LeaderRecoveryState.of(partitionData.leaderRecoveryState());
        if (leaderRecoveryState == LeaderRecoveryState.RECOVERING && newIsr.length > 1) {
            log.info("Rejecting AlterPartition request from node {} for {}-{} because " +
                    "the ISR {} had more than one replica while the leader was still " +
                    "recovering from an unclean leader election {}.",
                    brokerId, topic.name, partitionId, partitionData.newIsrWithEpochs(),
                    leaderRecoveryState);

            return INVALID_REQUEST;
        }
        if (partition.leaderRecoveryState == LeaderRecoveryState.RECOVERED &&
                leaderRecoveryState == LeaderRecoveryState.RECOVERING) {
            log.info("Rejecting AlterPartition request from node {} for {}-{} because " +
                    "the leader recovery state cannot change from RECOVERED to RECOVERING.",
                    brokerId, topic.name, partitionId);

            return INVALID_REQUEST;
        }

        List<IneligibleReplica> ineligibleReplicas = ineligibleReplicasForIsr(partitionData.newIsrWithEpochs());
        if (!ineligibleReplicas.isEmpty()) {
            log.info("Rejecting AlterPartition request from node {} for {}-{} because " +
                    "it specified ineligible replicas {} in the new ISR {}.",
                    brokerId, topic.name, partitionId, ineligibleReplicas, partitionData.newIsrWithEpochs());

            if (requestApiVersion > 1) {
                return INELIGIBLE_REPLICA;
            } else {
                return OPERATION_NOT_ATTEMPTED;
            }
        }

        return Errors.NONE;
    }
Validate the partition information included in the alter partition request.

@param brokerId id of the broker requesting the alter partition
@param topic current topic information store by the replication manager
@param partitionId partition id being altered
@param partition current partition registration for the partition being altered
@param partitionData partition data from the alter partition request

@return Errors.NONE for valid alter partition data; otherwise the validation error
metadata/src/main/java/org/apache/kafka/controller/ReplicationControlManager.java