DiskByLogDirFunction { public Map<String, Disk> apply(Map<TopicPartition, Replica> logDirsByPartition) { return Flux .fromIterable(logDirsByPartition.values()) .map(Replica::getLogDir) .collect(toSet()) .flatMapIterable(identity()) .map(File::new) .collectMap(d -> d.getAbsolutePath(), d -> new Disk(d.getFreeSpace(), d.getTotalSpace())) .block(); }  Map<String, Disk> apply(Map<TopicPartition, Replica> logDirsByPartition);  }
@Test public void typical() throws Exception { String logDir = temp.getRoot().getAbsolutePath(); TopicPartition topicPartition = new TopicPartition("topic", 0); Map<TopicPartition, Replica> replicaByPartition = Collections.singletonMap(topicPartition, new Replica(logDir, 0L)); Map<String, Disk> result = underTest.apply(replicaByPartition); assertThat(result.size(), is(1)); Disk disk = result.get(logDir); assertThat(disk.getFree() > 0L, is(true)); assertThat(disk.getTotal() > 0L, is(true)); }