ReplicaByPartitionFunction { public Map<TopicPartition, Replica> apply(int brokerId) { return Mono .fromSupplier(() -> client.describeLogDirs(singleton(brokerId)).values().get(brokerId)) .map(KafkaFutures::join) .flatMapIterable(Map::entrySet) .flatMap(e -> Flux.fromIterable(e.getValue().replicaInfos.entrySet()).map( entry -> Tuples.of(entry.getKey(), new Replica(e.getKey(), entry.getValue().size)))) .collectMap(Tuple2::getT1, Tuple2::getT2) .block(); }  Map<TopicPartition, Replica> apply(int brokerId);  }
@Test public void typical() throws Exception { ReplicaInfo ri = new ReplicaInfo(42L, 0L, false); TopicPartition topicPartition = new TopicPartition("topic", 0); Map<TopicPartition, ReplicaInfo> replicaInfos = Collections.singletonMap(topicPartition, ri); LogDirInfo ldi = new LogDirInfo(null, replicaInfos); Map<String, LogDirInfo> ldis = Collections.singletonMap("logDir", ldi); KafkaFuture<Map<String, LogDirInfo>> kafkaFuture = KafkaFuture.completedFuture(ldis); Map<Integer, KafkaFuture<Map<String, LogDirInfo>>> values = Collections.singletonMap(brokerId, kafkaFuture); doReturn(describeLogDirsResult).when(adminClient).describeLogDirs(brokerIds); doReturn(values).when(describeLogDirsResult).values(); Map<TopicPartition, Replica> result = underTest.apply(brokerId); assertThat(result.size(), is(1)); Replica replica = new Replica("logDir", 42L); assertThat(result.get(topicPartition), is(replica)); }