private LogDirReassignment buildLogDirReassignment(TopicPartition topicPartition,
                                                       int brokerId,
                                                       List<Integer> replicas) throws ExecutionException, InterruptedException {
        try (Admin admin = Admin.create(Collections.singletonMap(CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG, clusterInstance.bootstrapServers()))) {
            DescribeLogDirsResult describeLogDirsResult = admin.describeLogDirs(
                    IntStream.range(0, 4).boxed().collect(Collectors.toList()));

            BrokerDirs logDirInfo = new BrokerDirs(describeLogDirsResult, brokerId);
            assertTrue(logDirInfo.futureLogDirs.isEmpty());

            String currentDir = logDirInfo.curLogDirs.get(topicPartition);
            String newDir = logDirInfo.logDirs.stream().filter(dir -> !dir.equals(currentDir)).findFirst().get();

            List<String> logDirs = replicas.stream().map(replicaId -> {
                if (replicaId == brokerId)
                    return "\"" + newDir + "\"";
                else
                    return "\"any\"";
            }).collect(Collectors.toList());

            String reassignmentJson =
                    " { \"version\": 1," +
                            "  \"partitions\": [" +
                            "    {" +
                            "     \"topic\": \"" + topicPartition.topic() + "\"," +
                            "     \"partition\": " + topicPartition.partition() + "," +
                            "     \"replicas\": [" + replicas.stream().map(Object::toString).collect(Collectors.joining(",")) + "]," +
                            "     \"log_dirs\": [" + String.join(",", logDirs) + "]" +
                            "    }" +
                            "   ]" +
                            "  }";

            return new LogDirReassignment(reassignmentJson, currentDir, newDir);
        }
    }
Describe the broker-level throttles in the cluster.

@return                A map whose keys are broker IDs and whose values are throttle
                       information.  The nested maps are keyed on throttle name.
tools/src/test/java/org/apache/kafka/tools/reassign/ReassignPartitionsCommandTest.java