ControllerResult<AssignReplicasToDirsResponseData> handleAssignReplicasToDirs(AssignReplicasToDirsRequestData request) {
        if (!featureControl.metadataVersion().isDirectoryAssignmentSupported()) {
            throw new UnsupportedVersionException("Directory assignment is not supported yet.");
        }
        int brokerId = request.brokerId();
        clusterControl.checkBrokerEpoch(brokerId, request.brokerEpoch());
        BrokerRegistration brokerRegistration = clusterControl.brokerRegistrations().get(brokerId);
        if (brokerRegistration == null) {
            throw new BrokerIdNotRegisteredException("Broker ID " + brokerId + " is not currently registered");
        }
        List<ApiMessageAndVersion> records = new ArrayList<>();
        AssignReplicasToDirsResponseData response = new AssignReplicasToDirsResponseData();
        Set<TopicIdPartition> leaderAndIsrUpdates = new HashSet<>();
        for (AssignReplicasToDirsRequestData.DirectoryData reqDir : request.directories()) {
            Uuid dirId = reqDir.id();
            boolean directoryIsOffline = !brokerRegistration.hasOnlineDir(dirId);
            AssignReplicasToDirsResponseData.DirectoryData resDir = new AssignReplicasToDirsResponseData.DirectoryData().setId(dirId);
            for (AssignReplicasToDirsRequestData.TopicData reqTopic : reqDir.topics()) {
                Uuid topicId = reqTopic.topicId();
                Errors topicError = Errors.NONE;
                TopicControlInfo topicInfo = this.topics.get(topicId);
                if (topicInfo == null) {
                    log.warn("AssignReplicasToDirsRequest from broker {} references unknown topic ID {}", brokerId, topicId);
                    topicError = Errors.UNKNOWN_TOPIC_ID;
                }
                AssignReplicasToDirsResponseData.TopicData resTopic = new AssignReplicasToDirsResponseData.TopicData().setTopicId(topicId);
                for (AssignReplicasToDirsRequestData.PartitionData reqPartition : reqTopic.partitions()) {
                    int partitionIndex = reqPartition.partitionIndex();
                    Errors partitionError = topicError;
                    if (topicError == Errors.NONE) {
                        String topicName = topicInfo.name;
                        PartitionRegistration partitionRegistration = topicInfo.parts.get(partitionIndex);
                        if (partitionRegistration == null) {
                            log.warn("AssignReplicasToDirsRequest from broker {} references unknown partition {}-{}", brokerId, topicName, partitionIndex);
                            partitionError = Errors.UNKNOWN_TOPIC_OR_PARTITION;
                        } else if (!Replicas.contains(partitionRegistration.replicas, brokerId)) {
                            log.warn("AssignReplicasToDirsRequest from broker {} references non assigned partition {}-{}", brokerId, topicName, partitionIndex);
                            partitionError = Errors.NOT_LEADER_OR_FOLLOWER;
                        } else {
                            Optional<ApiMessageAndVersion> partitionChangeRecord = new PartitionChangeBuilder(
                                    partitionRegistration,
                                    topicId,
                                    partitionIndex,
                                    new LeaderAcceptor(clusterControl, partitionRegistration),
                                    featureControl.metadataVersion(),
                                    getTopicEffectiveMinIsr(topicName)
                            )
                                    .setDirectory(brokerId, dirId)
                                    .setDefaultDirProvider(clusterDescriber)
                                    .build();
                            partitionChangeRecord.ifPresent(records::add);
                            if (directoryIsOffline) {
                                leaderAndIsrUpdates.add(new TopicIdPartition(topicId, partitionIndex));
                            }
                            if (log.isDebugEnabled()) {
                                log.debug("Broker {} assigned partition {}:{} to {} dir {}",
                                    brokerId, topics.get(topicId).name(), partitionIndex,
                                    directoryIsOffline ? "OFFLINE" : "ONLINE", dirId);
                            }
                        }
                    }
                    resTopic.partitions().add(new AssignReplicasToDirsResponseData.PartitionData().
                            setPartitionIndex(partitionIndex).
                            setErrorCode(partitionError.code()));
                }
                resDir.topics().add(resTopic);
            }
            response.directories().add(resDir);
        }
        if (!leaderAndIsrUpdates.isEmpty()) {
            generateLeaderAndIsrUpdates("offline-dir-assignment", brokerId, NO_LEADER, NO_LEADER, records, leaderAndIsrUpdates.iterator());
        }
        return ControllerResult.of(records, response);
    }
Apply a given partition reassignment. In general a partition reassignment goes
through several stages:

1. Issue a PartitionChangeRecord adding all the new replicas to the partition's
main replica list, and setting removingReplicas and addingReplicas.

2. Wait for the partition to have an ISR that contains all the new replicas. Or
if there are no new replicas, wait until we have an ISR that contains at least one
replica that we are not removing.

3. Issue a second PartitionChangeRecord removing all removingReplicas from the
partitions' main replica list, and clearing removingReplicas and addingReplicas.

After stage 3, the reassignment is done.

Under some conditions, steps #1 and #2 can be skipped entirely since the ISR is
already suitable to progress to stage #3. For example, a partition reassignment
that merely rearranges existing replicas in the list can bypass step #1 and #2 and
complete immediately.

@param tp                The topic id and partition id.
@param part              The existing partition info.
@param target            The target partition info.

@return                  The ChangePartitionRecord for the new partition assignment,
                         or empty if no change is needed.
metadata/src/main/java/org/apache/kafka/controller/ReplicationControlManager.java