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);
    }
@Test
    void testHandleAssignReplicasToDirsFailsOnOlderMv() {
        ReplicationControlTestContext ctx = new ReplicationControlTestContext.Builder().
            setMetadataVersion(MetadataVersion.IBP_3_7_IV1).
            build();
        assertThrows(UnsupportedVersionException.class,
            () -> ctx.replicationControl.handleAssignReplicasToDirs(new AssignReplicasToDirsRequestData()));
    }