void createPartitions(ControllerRequestContext context,
                          CreatePartitionsTopic topic,
                          List<ApiMessageAndVersion> records) {
        Uuid topicId = topicsByName.get(topic.name());
        if (topicId == null) {
            throw new UnknownTopicOrPartitionException();
        }
        TopicControlInfo topicInfo = topics.get(topicId);
        if (topicInfo == null) {
            throw new UnknownTopicOrPartitionException();
        }
        if (topic.count() == topicInfo.parts.size()) {
            throw new InvalidPartitionsException("Topic already has " +
                topicInfo.parts.size() + " partition(s).");
        } else if (topic.count() < topicInfo.parts.size()) {
            throw new InvalidPartitionsException("The topic " + topic.name() + " currently " +
                "has " + topicInfo.parts.size() + " partition(s); " + topic.count() +
                " would not be an increase.");
        }
        int additional = topic.count() - topicInfo.parts.size();
        if (topic.assignments() != null) {
            if (topic.assignments().size() != additional) {
                throw new InvalidReplicaAssignmentException("Attempted to add " + additional +
                    " additional partition(s), but only " + topic.assignments().size() +
                    " assignment(s) were specified.");
            }
        }
        try {
            context.applyPartitionChangeQuota(additional); // check controller mutation quota
        } catch (ThrottlingQuotaExceededException e) {
            // log a message and rethrow the exception
            log.debug("Partition creation of {} partitions not allowed because quota is violated. Delay time: {}",
                additional, e.throttleTimeMs());
            throw e;
        }
        Iterator<PartitionRegistration> iterator = topicInfo.parts.values().iterator();
        if (!iterator.hasNext()) {
            throw new UnknownServerException("Invalid state: topic " + topic.name() +
                " appears to have no partitions.");
        }
        PartitionRegistration partitionInfo = iterator.next();
        if (partitionInfo.replicas.length > Short.MAX_VALUE) {
            throw new UnknownServerException("Invalid replication factor " +
                partitionInfo.replicas.length + ": expected a number equal to less than " +
                Short.MAX_VALUE);
        }
        short replicationFactor = (short) partitionInfo.replicas.length;
        int startPartitionId = topicInfo.parts.size();

        List<PartitionAssignment> partitionAssignments;
        List<List<Integer>> isrs;
        if (topic.assignments() != null) {
            partitionAssignments = new ArrayList<>();
            isrs = new ArrayList<>();
            for (int i = 0; i < topic.assignments().size(); i++) {
                List<Integer> replicas = topic.assignments().get(i).brokerIds();
                PartitionAssignment partitionAssignment = new PartitionAssignment(replicas, clusterDescriber);
                validateManualPartitionAssignment(partitionAssignment, OptionalInt.of(replicationFactor));
                partitionAssignments.add(partitionAssignment);
                List<Integer> isr = partitionAssignment.replicas().stream().
                    filter(clusterControl::isActive).collect(Collectors.toList());
                if (isr.isEmpty()) {
                    throw new InvalidReplicaAssignmentException(
                        "All brokers specified in the manual partition assignment for " +
                            "partition " + (startPartitionId + i) + " are fenced or in controlled shutdown.");
                }
                isrs.add(isr);
            }
        } else {
            partitionAssignments = clusterControl.replicaPlacer().place(
                new PlacementSpec(startPartitionId, additional, replicationFactor),
                clusterDescriber
            ).assignments();
            isrs = partitionAssignments.stream().map(PartitionAssignment::replicas).collect(Collectors.toList());
        }
        int partitionId = startPartitionId;
        for (int i = 0; i < partitionAssignments.size(); i++) {
            PartitionAssignment partitionAssignment = partitionAssignments.get(i);
            List<Integer> isr = isrs.get(i).stream().
                filter(clusterControl::isActive).collect(Collectors.toList());
            // If the ISR is empty, it means that all brokers are fenced or
            // in controlled shutdown. To be consistent with the replica placer,
            // we reject the create topic request with INVALID_REPLICATION_FACTOR.
            if (isr.isEmpty()) {
                throw new InvalidReplicationFactorException(
                    "Unable to replicate the partition " + replicationFactor +
                        " time(s): All brokers are currently fenced or in controlled shutdown.");
            }

            // AutoMQ for Kafka inject start
            // Generally, validateManualPartitionAssignment has checked to some extent. We still check here for defensive programming.
            if (ElasticStreamSwitch.isEnabled()) {
                maybeCheckCreatePartitionPolicy(new CreatePartitionPolicy.RequestMetadata(partitionAssignment.replicas(), isr));
            }
            // AutoMQ for Kafka inject end

            records.add(buildPartitionRegistration(partitionAssignment, isr)
                .toRecord(topicId, partitionId, new ImageWriterOptions.Builder().
                        setMetadataVersion(featureControl.metadataVersion()).
                        build()));
            partitionId++;
        }
    }
Attempt to elect a preferred leader for all topic partitions which have a leader that is not the preferred replica.

The response() method in the return object is true if this method returned without electing all possible preferred replicas.
The quorum controller should reschedule this operation immediately if it is true.

@return All of the election records and if there may be more available preferred replicas to elect as leader
metadata/src/main/java/org/apache/kafka/controller/ReplicationControlManager.java