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

Code before:
                    partitionId,
                    new LeaderAcceptor(clusterControl, partition),
                    featureControl.metadataVersionOrThrow(),
                    getTopicEffectiveMinIsr(topic.name)
                )
                    .setEligibleLeaderReplicasEnabled(featureControl.isElrFeatureEnabled());
                if (configurationControl.uncleanLeaderElectionEnabledForTopic(topic.name())) {
                    builder.setElection(PartitionChangeBuilder.Election.UNCLEAN);
                }
                Optional<ApiMessageAndVersion> record = builder
                    .setTargetIsrWithBrokerStates(partitionData.newIsrWithEpochs())
                    .setTargetLeaderRecoveryState(LeaderRecoveryState.of(partitionData.leaderRecoveryState()))
                    .setDefaultDirProvider(clusterDescriber)
                    .build();
                if (record.isPresent()) {
                    records.add(record.get());
                    PartitionChangeRecord change = (PartitionChangeRecord) record.get().message();
                    String priorIsr = Arrays.toString(partition.isr);
                    String priorReplicaSet = Arrays.toString(partition.replicas);
                    int priorPartitionEpoch = partition.partitionEpoch;
                    int priorLeaderEpoch = partition.leaderEpoch;
                    partition = partition.merge(change);
                    String partitionChangeInfo = String.format("isr: %s -> %s, replicaSet: %s -> %s, " +
                            "partitionEpoch: %d -> %d, leaderEpoch: %d -> %d",
                        priorIsr, Arrays.toString(partition.isr), priorReplicaSet, Arrays.toString(partition.replicas),
                        priorPartitionEpoch, partition.partitionEpoch, priorLeaderEpoch, partition.leaderEpoch);
                    if (log.isDebugEnabled()) {
                        log.debug("Node {} has altered ISR for {}-{}. {}",
                            request.brokerId(), topic.name, partitionId, partitionChangeInfo);
                    }
                    if (change.leader() != request.brokerId() &&
                            change.leader() != NO_LEADER_CHANGE) {
                        // Normally, an AlterPartition request, which is made by the partition
                        // leader itself, is not allowed to modify the partition leader.
                        // However, if there is an ongoing partition reassignment and the
                        // ISR change completes it, then the leader may change as part of
                        // the changes made during reassignment cleanup.
                        //
                        // In this case, we report back NEW_LEADER_ELECTED to the leader
                        // which made the AlterPartition request. This lets it know that it must
                        // fetch new metadata before trying again. This return code is
                        // unusual because we both return an error and generate a new
                        // metadata record. We usually only do one or the other.
                        Errors error = NEW_LEADER_ELECTED;
                        log.info("AlterPartition request from node {} for {}-{} completed " +
                            "the ongoing partition reassignment and triggered a leadership change {}. Returning {}.",
                            request.brokerId(), topic.name, partitionId, partitionChangeInfo, error);
                        responseTopicData.partitions().add(new AlterPartitionResponseData.PartitionData().
                            setPartitionIndex(partitionId).
                            setErrorCode(error.code()));
                        continue;

Review comment (bug):
All of this work should only be done if debug logging is enabled. You can take advantage of the fact that PartitionRegistration are immutable and that PartitionRegistration#merge is side effect free to make sure that all of this formatting is only performed if debug logging is enabled.

Can you move this to a private static method? This a lot of code to generate a string which distracts from the core functionality of the method.

Code after:
                    partitionId,
                    new LeaderAcceptor(clusterControl, partition),
                    featureControl.metadataVersionOrThrow(),
                    getTopicEffectiveMinIsr(topic.name)
                )
                    .setEligibleLeaderReplicasEnabled(featureControl.isElrFeatureEnabled());
                if (configurationControl.uncleanLeaderElectionEnabledForTopic(topic.name())) {
                    builder.setElection(PartitionChangeBuilder.Election.UNCLEAN);
                }
                Optional<ApiMessageAndVersion> record = builder
                    .setTargetIsrWithBrokerStates(partitionData.newIsrWithEpochs())
                    .setTargetLeaderRecoveryState(LeaderRecoveryState.of(partitionData.leaderRecoveryState()))
                    .setDefaultDirProvider(clusterDescriber)
                    .build();
                if (record.isPresent()) {
                    records.add(record.get());
                    PartitionChangeRecord change = (PartitionChangeRecord) record.get().message();
                    PartitionRegistration originalPartition = partition;
                    partition = originalPartition.merge(change);
                    if (log.isDebugEnabled()) {
                        log.debug("Node {} has altered ISR for {}-{}. {}",
                            request.brokerId(), topic.name, partitionId, logPartitionChangeInfo(originalPartition, partition));
                    }
                    if (change.leader() != request.brokerId() &&
                            change.leader() != NO_LEADER_CHANGE) {
                        // Normally, an AlterPartition request, which is made by the partition
                        // leader itself, is not allowed to modify the partition leader.
                        // However, if there is an ongoing partition reassignment and the
                        // ISR change completes it, then the leader may change as part of
                        // the changes made during reassignment cleanup.
                        //
                        // In this case, we report back NEW_LEADER_ELECTED to the leader
                        // which made the AlterPartition request. This lets it know that it must
                        // fetch new metadata before trying again. This return code is
                        // unusual because we both return an error and generate a new
                        // metadata record. We usually only do one or the other.
                        Errors error = NEW_LEADER_ELECTED;
                        log.info("AlterPartition request from node {} for {}-{} completed " +
                            "the ongoing partition reassignment and triggered a leadership change {}. Returning {}.",
                            request.brokerId(), topic.name, partitionId,
                            logPartitionChangeInfo(originalPartition, partition), error);
                        responseTopicData.partitions().add(new AlterPartitionResponseData.PartitionData().
                            setPartitionIndex(partitionId).
                            setErrorCode(error.code()));
                        continue;
                    } else if (isReassignmentInProgress(partition)) {
                        log.info("AlterPartition request from node {} for {}-{} completed " +
                            "the ongoing partition reassignment. {}",
                            request.brokerId(), topic.name, partitionId, logPartitionChangeInfo(originalPartition, partition));
                    }
                }