BrokerSupplier implements Supplier<Broker> { @Override public Broker get() { Map<TopicPartition, Replica> replicaByPartition = replicaByPartitionFunction.apply(brokerNode.getId()); Set<TopicPartition> partitions = replicaByPartition.keySet(); Set<String> topics = Flux.fromIterable(partitions).map(TopicPartition::topic).collect(toSet()).block(); Map<TopicPartition, LeaderInSync> leaderInSyncByPartition = leaderInSyncByPartitionFunction .apply(brokerNode.getId(), topics); Map<String, Duration> retentionByTopic = retentionByTopicFunction.apply(topics); Map<TopicPartition, Offsets> offsetsByPartition = offsetsByPartitionFunction.apply(partitions); Map<TopicPartition, Long> sizeByPartition = sizeByPartitionFunction.apply(replicaByPartition); Map<String, Disk> diskByLogDir = diskByLogDirFunction.apply(replicaByPartition); return Flux .fromIterable(partitions) .map(p -> replica(replicaByPartition, offsetsByPartition, leaderInSyncByPartition, sizeByPartition, p)) .groupBy(Tuple3::getT1) .flatMap( byLogDir -> byLogDir .sort(comparing(Tuple3::getT2)) .windowUntil(keyChanged(Tuple3::getT2), true) .flatMap( byTopic -> byTopic .collectMultimap(Tuple3::getT2, Tuple3::getT3) .flatMapIterable(Map::entrySet) .map( e -> new Topic( e.getKey(), retentionByTopic.getOrDefault(e.getKey(), ZERO), newArrayList(e.getValue())))) .collectList() .map(ts -> logDir(diskByLogDir, byLogDir, ts))) .collectList() .map(lds -> new Broker(brokerNode.getId(), brokerNode.getRack(), lds)) .block(); } BrokerSupplier(
      Predicate<String> hostPredicate,
      BrokerNodeFunction brokerNodeFunction,
      ReplicaByPartitionFunction replicaByPartitionFunction,
      LeaderInSyncByPartitionFunction leaderInSyncByPartitionFunction,
      RetentionByTopicFunction retentionByTopicFunction,
      OffsetsByPartitionFunction offsetsByPartitionFunction,
      SizeByPartitionFunction sizeByPartitionFunction,
      DiskByLogDirFunction diskByLogDirFunction); @Override Broker get();  }
@Test public void typical() throws Exception { String t = "topic"; Collection<String> ts = singleton(t); TopicPartition tp = new TopicPartition(t, 0); Collection<TopicPartition> tps = singleton(tp); Map<TopicPartition, Replica> replicaByPartition = singletonMap(tp, new Replica("logDir", 42L)); doReturn(new BrokerNode(0, "rack", "host")).when(brokerNodeFunction).apply(hostPredicate); doReturn(replicaByPartition).when(replicaByPartitionFunction).apply(0); doReturn(singletonMap(tp, new LeaderInSync(false, true))).when(leaderInSyncByPartitionFunction).apply(0, ts); doReturn(singletonMap(t, Duration.ofMillis(1))).when(retentionByTopicFunction).apply(ts); doReturn(singletonMap(tp, new Offsets(1L, 3L))).when(offsetsByPartitionFunction).apply(tps); doReturn(singletonMap(tp, 12L)).when(sizeByPartitionFunction).apply(replicaByPartition); doReturn(singletonMap("logDir", new DiskByLogDirFunction.Disk(123L, 234L))).when(diskByLogDirFunction).apply( replicaByPartition); underTest = new BrokerSupplier(hostPredicate, brokerNodeFunction, replicaByPartitionFunction, leaderInSyncByPartitionFunction, retentionByTopicFunction, offsetsByPartitionFunction, sizeByPartitionFunction, diskByLogDirFunction); Broker broker = underTest.get(); assertThat(broker.getId(), is(0)); assertThat(broker.getRack(), is("rack")); List<LogDir> logDirs = broker.getLogDirs(); assertThat(logDirs.size(), is(1)); LogDir logDir = logDirs.get(0); assertThat(logDir.getPath(), is("logDir")); assertThat(logDir.getDiskFree(), is(123L)); assertThat(logDir.getDiskTotal(), is(234L)); List<Topic> topics = logDir.getTopics(); assertThat(topics.size(), is(1)); Topic topic = topics.get(0); assertThat(topic.getName(), is("topic")); assertThat(topic.getRetention(), is(Duration.ofMillis(1))); List<PartitionReplica> replicas = topic.getPartitionReplicas(); assertThat(replicas.size(), is(1)); PartitionReplica replica = replicas.get(0); assertThat(replica.getPartition(), is(0)); assertThat(replica.isLeader(), is(false)); assertThat(replica.isInSync(), is(true)); assertThat(replica.getSizeOnDisk(), is(12L)); assertThat(replica.getLogSize(), is(42L)); assertThat(replica.getBeginningOffset(), is(1L)); assertThat(replica.getEndOffset(), is(3L)); assertThat(replica.getRecordCount(), is(2L)); }