void handleDirectoriesOffline(
        int brokerId,
        long brokerEpoch,
        List<Uuid> offlineDirs,
        List<ApiMessageAndVersion> records
    ) {
        BrokerRegistration registration = clusterControl.registration(brokerId);
        List<Uuid> newOfflineDirs = registration.directoryIntersection(offlineDirs);
        if (!newOfflineDirs.isEmpty()) {
            for (Uuid newOfflineDir : newOfflineDirs) {
                TimelineHashSet<TopicIdPartition> parts = directoriesToPartitions.get(newOfflineDir);
                Iterator<TopicIdPartition> iterator = (parts == null) ?
                        Collections.emptyIterator() : parts.iterator();
                generateLeaderAndIsrUpdates(
                        "handleDirectoriesOffline[" + brokerId + ":" + newOfflineDir + "]",
                        brokerId, NO_LEADER, NO_LEADER, records, iterator);
            }
            List<Uuid> newOnlineDirs = registration.directoryDifference(offlineDirs);
            records.add(new ApiMessageAndVersion(new BrokerRegistrationChangeRecord().
                    setBrokerId(brokerId).setBrokerEpoch(brokerEpoch).
                    setLogDirs(newOnlineDirs),
                    (short) 2));
            log.warn("Directories {} in broker {} marked offline, remaining directories: {}",
                    newOfflineDirs, brokerId, newOnlineDirs);
        }
    }
@Test
    void testHandleDirectoriesOffline() {
        ReplicationControlTestContext ctx = new ReplicationControlTestContext.Builder().build();
        int b1 = 101, b2 = 102;
        Uuid dir1b1 = Uuid.fromString("suitdzfTTdqoWcy8VqmkUg");
        Uuid dir2b1 = Uuid.fromString("yh3acnzGSeurSTj8aIhOjw");
        Uuid dir1b2 = Uuid.fromString("OmpmJ8RjQliQlEFht56DwQ");
        Uuid dir2b2 = Uuid.fromString("w05baLpsT5Oz0LvKTKXoDw");
        ctx.registerBrokersWithDirs(b1, asList(dir1b1, dir2b1), b2, asList(dir1b2, dir2b2));
        ctx.unfenceBrokers(b1, b2);
        Uuid topicA = ctx.createTestTopic("a", new int[][]{new int[]{b1, b2}, new int[]{b1, b2}}).topicId();
        Uuid topicB = ctx.createTestTopic("b", new int[][]{new int[]{b1, b2}, new int[]{b1, b2}}).topicId();
        ctx.assignReplicasToDirs(b1, new HashMap<TopicIdPartition, Uuid>() {{
                put(new TopicIdPartition(topicA, 0), dir1b1);
                put(new TopicIdPartition(topicA, 1), dir2b1);
                put(new TopicIdPartition(topicB, 0), dir1b1);
                put(new TopicIdPartition(topicB, 1), dir2b1);
            }});
        ctx.assignReplicasToDirs(b2, new HashMap<TopicIdPartition, Uuid>() {{
                put(new TopicIdPartition(topicA, 0), dir1b2);
                put(new TopicIdPartition(topicA, 1), dir2b2);
                put(new TopicIdPartition(topicB, 0), dir1b2);
                put(new TopicIdPartition(topicB, 1), dir2b2);
            }});
        List<ApiMessageAndVersion> records = new ArrayList<>();
        ctx.replicationControl.handleDirectoriesOffline(b1, defaultBrokerEpoch(b1), asList(
                dir1b1,
                dir1b2 // should not cause update to dir1b2 as it's not registered to b1
        ), records);
        assertEquals(
            singletonList(new ApiMessageAndVersion(new BrokerRegistrationChangeRecord()
                    .setBrokerId(b1).setBrokerEpoch(defaultBrokerEpoch(b1))
                    .setLogDirs(singletonList(dir2b1)), (short) 2)),
            filter(records, BrokerRegistrationChangeRecord.class)
        );
        short partitionChangeRecordVersion = ctx.featureControl.metadataVersion().partitionChangeRecordVersion();
        assertEquals(
            sortPartitionChangeRecords(asList(
                new ApiMessageAndVersion(new PartitionChangeRecord().setTopicId(topicA).setPartitionId(0)
                        .setLeader(b2).setIsr(singletonList(b2)), partitionChangeRecordVersion),
                new ApiMessageAndVersion(new PartitionChangeRecord().setTopicId(topicB).setPartitionId(0)
                        .setLeader(b2).setIsr(singletonList(b2)), partitionChangeRecordVersion)
            )),
            sortPartitionChangeRecords(filter(records, PartitionChangeRecord.class))
        );
        assertEquals(3, records.size());
        ctx.replay(records);
        assertEquals(Collections.singletonList(dir2b1), ctx.clusterControl.registration(b1).directories());
    }