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

Code before:
                 * RECOVERED which is ignored when serializing to version 0. For any other version, the
                 * LeaderRecoveryState field is supported.
                 */
                responseTopicData.partitions().add(new AlterPartitionResponseData.PartitionData().
                    setPartitionIndex(partitionId).
                    setErrorCode(Errors.NONE.code()).
                    setLeaderId(partition.leader).
                    setIsr(Replicas.toList(partition.isr)).
                    setLeaderRecoveryState(partition.leaderRecoveryState.value()).
                    setLeaderEpoch(partition.leaderEpoch).
                    setPartitionEpoch(partition.partitionEpoch));
            }
        }

        return ControllerResult.of(records, response);
    }

    /**
     * Validates that a batch of topics will create less than {@value MAX_PARTITIONS_PER_BATCH}. Exceeding this number of topics per batch
     * has led to out-of-memory exceptions. We use this validation to fail earlier to avoid allocating the memory.
     * Validates an upper bound number of partitions. The actual number may be smaller if some topics are misconfigured.
     *
     * @param request a batch of topics to create.
     * @param defaultNumPartitions default number of partitions to assign if unspecified.
     * @throws PolicyViolationException if total number of partitions exceeds {@value MAX_PARTITIONS_PER_BATCH}.
     */
    static void validateTotalNumberOfPartitions(CreateTopicsRequestData request, int defaultNumPartitions) {
        int totalPartitions = 0;
        for (CreatableTopic topic: request.topics()) {
            if (topic.assignments().isEmpty()) {
                if (topic.numPartitions() == -1) {
                    totalPartitions += defaultNumPartitions;
                } else if (topic.numPartitions() > 0) {
                    totalPartitions += topic.numPartitions();
                }
            } else {
                totalPartitions += topic.assignments().size();
            }

        }
        if (totalPartitions > MAX_PARTITIONS_PER_BATCH) {
            throw new PolicyViolationException("Excessively large number of partitions per request.");
        }
    }

    /**
     * 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

Review comment (suggestion):
Okay but you can use `%s` for all of these formats. `%d` is different from `%s` only if you use `%d`'s special formatting rules like `%09d`.

Code after:
                            "the ongoing partition reassignment. {}",
                            request.brokerId(), topic.name, partitionId, logPartitionChangeInfo(originalPartition, partition));
                    }
                }

                /* Setting the LeaderRecoveryState field is always safe because it will always be the
                 * same as the value set in the request. For version 0, that is always the default
                 * RECOVERED which is ignored when serializing to version 0. For any other version, the
                 * LeaderRecoveryState field is supported.
                 */
                responseTopicData.partitions().add(new AlterPartitionResponseData.PartitionData().
                    setPartitionIndex(partitionId).
                    setErrorCode(Errors.NONE.code()).
                    setLeaderId(partition.leader).
                    setIsr(Replicas.toList(partition.isr)).
                    setLeaderRecoveryState(partition.leaderRecoveryState.value()).
                    setLeaderEpoch(partition.leaderEpoch).
                    setPartitionEpoch(partition.partitionEpoch));
            }
        }

        return ControllerResult.of(records, response);
    }

    private static String logPartitionChangeInfo(PartitionRegistration oldRegistration, PartitionRegistration newRegistration) {
        return String.format("isr: %s -> %s, replicaSet: %s -> %s, partitionEpoch: %d -> %d, leaderEpoch: %d -> %d",
            Arrays.toString(oldRegistration.isr), Arrays.toString(newRegistration.isr),
            Arrays.toString(oldRegistration.replicas), Arrays.toString(newRegistration.replicas),
            oldRegistration.partitionEpoch, newRegistration.partitionEpoch,
            oldRegistration.leaderEpoch, newRegistration.leaderEpoch);
    }

    /**
     * Validates that a batch of topics will create less than {@value MAX_PARTITIONS_PER_BATCH}. Exceeding this number of topics per batch
     * has led to out-of-memory exceptions. We use this validation to fail earlier to avoid allocating the memory.
     * Validates an upper bound number of partitions. The actual number may be smaller if some topics are misconfigured.
     *
     * @param request a batch of topics to create.
     * @param defaultNumPartitions default number of partitions to assign if unspecified.
     * @throws PolicyViolationException if total number of partitions exceeds {@value MAX_PARTITIONS_PER_BATCH}.
     */
    static void validateTotalNumberOfPartitions(CreateTopicsRequestData request, int defaultNumPartitions) {
        int totalPartitions = 0;
        for (CreatableTopic topic: request.topics()) {
            if (topic.assignments().isEmpty()) {
                if (topic.numPartitions() == -1) {
                    totalPartitions += defaultNumPartitions;
                } else if (topic.numPartitions() > 0) {
                    totalPartitions += topic.numPartitions();
                }
            } else {