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

Code before:
    private boolean canElectLastKnownLeader() {
        if (!eligibleLeaderReplicasEnabled || !useLastKnownLeaderInBalancedRecovery) {
            log.trace("Try to elect last known leader for {}-{} but elrEnabled={}, useLastKnownLeaderInBalancedRecovery={}",
                    topicId, partitionId, eligibleLeaderReplicasEnabled, useLastKnownLeaderInBalancedRecovery);
            return false;
        }
        if (!targetElr.isEmpty() || !targetIsr.isEmpty()) {
            log.trace("Try to elect last known leader for {}-{} but ELR/ISR is not empty. ISR={}, ELR={}",
                    topicId, partitionId, targetIsr, targetElr);
            return false;
        }

        // When the last known leader is enabled:
        // 1. The targetLastKnownElr will only be used to store the last known leader, and it is updated after the
        //    leader election. So we can only refer to the lastKnownElr in the existing partition registration.
        // 2. When useLastKnownLeaderInBalancedRecovery=false, it intends to use other type of unclean leader election
        //    and the lastKnownElr is populated with the real last known ELR members. Then it may have multiple members
        //    in the field even if useLastKnownLeaderInBalancedRecovery is set to true again. In this case, we can't
        //    refer to the lastKnownElr.
        if (partition.lastKnownElr.length != 1) {
            log.trace("Try to elect last known leader for {}-{} but lastKnownElr does not only have 1 member. lastKnownElr={}",
                    topicId, partitionId, Arrays.toString(partition.lastKnownElr));
            return false;
        }
        if (isAcceptableLeader.test(partition.lastKnownElr[0])) {
            log.trace("Try to elect last known leader for {}-{} but last known leader is not alive. last known leader={}",
                    topicId, partitionId, partition.lastKnownElr[0]);
        }
        return true;
    }

    private boolean isValidNewLeader(int replica) {
        // The valid new leader should be in either ISR or in ELR when ISR is empty.
        return (targetIsr.contains(replica) || (targetIsr.isEmpty() && targetElr.contains(replica))) &&
            isAcceptableLeader.test(replica);
    }

    private void tryElection(PartitionChangeRecord record) {
        ElectionResult electionResult = electLeader();
        if (electionResult.node != partition.leader) {
            // generating log messages for partition elections can get expensive on large clusters,
            // so only log clean elections at TRACE level; log unclean elections at INFO level
            // to ensure the message is emitted since an unclean election can lead to data loss;
            if (targetElr.contains(electionResult.node)) {
                targetIsr = List.of(electionResult.node);
                targetElr = targetElr.stream().filter(replica -> replica != electionResult.node)
                    .collect(Collectors.toList());
                log.info("Setting new leader for topicId {}, partition {} to {} using ELR. Previous partition: {}, change record: {}",
                        topicId, partitionId, electionResult.node, partition, record);
            } else if (electionResult.unclean) {
                log.info("Setting new leader for topicId {}, partition {} to {} using an unclean election. Previous partition: {}, change record: {}",

Review comment (suggestion):
Sorry for late reviews. 

It seems this comment is not totally accurate. The `isAcceptableLeader` could reject the leader if the leader, while still alive, has no available directory for the partition.

Code after:
    private boolean canElectLastKnownLeader() {
        if (!eligibleLeaderReplicasEnabled || !useLastKnownLeaderInBalancedRecovery) {
            log.trace("Try to elect last known leader for {}-{} but elrEnabled={}, useLastKnownLeaderInBalancedRecovery={}",
                    topicId, partitionId, eligibleLeaderReplicasEnabled, useLastKnownLeaderInBalancedRecovery);
            return false;
        }
        if (!targetElr.isEmpty() || !targetIsr.isEmpty()) {
            log.trace("Try to elect last known leader for {}-{} but ELR/ISR is not empty. ISR={}, ELR={}",
                    topicId, partitionId, targetIsr, targetElr);
            return false;
        }

        // When the last known leader is enabled:
        // 1. The targetLastKnownElr will only be used to store the last known leader, and it is updated after the
        //    leader election. So we can only refer to the lastKnownElr in the existing partition registration.
        // 2. When useLastKnownLeaderInBalancedRecovery=false, it intends to use other type of unclean leader election
        //    and the lastKnownElr is populated with the real last known ELR members. Then it may have multiple members
        //    in the field even if useLastKnownLeaderInBalancedRecovery is set to true again. In this case, we can't
        //    refer to the lastKnownElr.
        if (partition.lastKnownElr.length != 1) {
            log.trace("Try to elect last known leader for {}-{} but lastKnownElr does not only have 1 member. lastKnownElr={}",
                    topicId, partitionId, Arrays.toString(partition.lastKnownElr));
            return false;
        }
        if (!isAcceptableLeader.test(partition.lastKnownElr[0])) {
            log.trace("Try to elect last known leader for {}-{} but last known leader is not alive. last known leader={}",
                    topicId, partitionId, partition.lastKnownElr[0]);
            return false;
        }
        return true;
    }

    private boolean isValidNewLeader(int replica) {
        // The valid new leader should be in either ISR or in ELR when ISR is empty.
        return (targetIsr.contains(replica) || (targetIsr.isEmpty() && targetElr.contains(replica))) &&
            isAcceptableLeader.test(replica);
    }

    private void tryElection(PartitionChangeRecord record) {
        ElectionResult electionResult = electLeader();
        if (electionResult.node != partition.leader) {
            // generating log messages for partition elections can get expensive on large clusters,
            // so only log clean elections at TRACE level; log unclean elections at INFO level
            // to ensure the message is emitted since an unclean election can lead to data loss;
            if (targetElr.contains(electionResult.node)) {
                targetIsr = List.of(electionResult.node);
                targetElr = targetElr.stream().filter(replica -> replica != electionResult.node)
                    .collect(Collectors.toList());
                log.info("Setting new leader for topicId {}, partition {} to {} using ELR. Previous partition: {}, change record: {}",
                        topicId, partitionId, electionResult.node, partition, record);
            } else if (electionResult.unclean) {