Repository: apache/kafka
File: metadata/src/main/java/org/apache/kafka/controller/ReplicationControlManager.java (Java)

Code before:
     *
     * @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
     */
    private Errors validateAlterPartitionData(
        int brokerId,
        TopicControlInfo topic,
        int partitionId,
        PartitionRegistration partition,
        AlterPartitionRequestData.PartitionData partitionData
    ) {
        if (partition == null) {
            log.info("Rejecting AlterPartition request for unknown partition {}-{}.",
                    topic.name, partitionId);

            return UNKNOWN_TOPIC_OR_PARTITION;
        }

        String partitionChangeInfo = String.format("Proposed ISR was %s and current ISR is %s. " +
                "Current replica set is %s. Proposed partitionEpoch was %d and current partitionEpoch is %d. " +
                "Proposed leaderEpoch was %d and current leaderEpoch is %d.",
            partitionData.newIsrWithEpochs(), Arrays.toString(partition.isr), Arrays.toString(partition.replicas),
            partitionData.partitionEpoch(), partition.partitionEpoch, partitionData.leaderEpoch(), partition.leaderEpoch);

        // 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(),
                partitionChangeInfo);
            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(),
                partitionChangeInfo);
            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(), partitionChangeInfo);


Review comment (refactor):
Senders of the ALTER_PARTITION request do not propose leader epoch or partition epoch. The active controller instead check that this value match to make sure that the changes are based on the same state. This is how Kafka implement optimistic locking for partition changes.

Every usage of `partitionChangeInfo` is followed by a `return`. So you can move this to a helper as it will only be called once.

Code after:
    /**
     * 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
     */
    private Errors validateAlterPartitionData(
        int brokerId,
        TopicControlInfo topic,
        int partitionId,
        PartitionRegistration partition,
        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(),
                logPartitionChangeInfo(partition, partitionData.newIsrWithEpochs()));
            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(),
                logPartitionChangeInfo(partition, partitionData.newIsrWithEpochs()));
            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(),
                    logPartitionChangeInfo(partition, partitionData.newIsrWithEpochs()));

            return FENCED_LEADER_EPOCH;
        }
        if (brokerId != partition.leader) {