Optional<ApiMessageAndVersion> changePartitionReassignmentV2(TopicIdPartition tp,
        PartitionRegistration part,
        ReassignablePartition target) {

        if (target.replicas().size() != 1) {
            throw new InvalidReplicaAssignmentException("elastic stream only support replicas = 1, partition[" + tp + "] replicas[" + target.replicas() + "]");
        }
        // Check that the requested partition assignment is valid.
        validateManualPartitionAssignment(target.replicas(), OptionalInt.empty());

        List<Integer> currentReplicas = Replicas.toList(part.replicas);
        // No need to change the assignment if it is already the same.
        if (Objects.equals(currentReplicas, target.replicas())) {
            // The requested assignment is the same as the current assignment.
            return Optional.empty();
        }

        PartitionChangeBuilder builder = new PartitionChangeBuilder(part,
            tp.topicId(),
            tp.partitionId(),
            // no leader election
            brokerId -> false,
            featureControl.metadataVersion(),
            getTopicEffectiveMinIsr(topics.get(tp.topicId()).name.toString())
        );
        builder.setZkMigrationEnabled(clusterControl.zkRegistrationAllowed());
        builder.setEligibleLeaderReplicasEnabled(isElrEnabled());
        builder.setTargetNode(target.replicas().get(0));
        TopicControlInfo topicControlInfo = topics.get(tp.topicId());
        if (topicControlInfo == null) {
            log.warn("unknown topicId[{}]", tp.topicId());
        } else {
            TopicPartition topicPartition = new TopicPartition(topicControlInfo.name, tp.partitionId());
            addPartitionToReElectTimeouts(topicPartition);
        }
        return builder.setDefaultDirProvider(clusterDescriber).build();
    }
Updates the directory to partition mapping for a single partition.
Assignments to reserved directory IDs are ignored, since they cannot
be used for directories, there's no use in maintaining a set of
partitions assigned to them.
metadata/src/main/java/org/apache/kafka/controller/ReplicationControlManager.java