public Flux<InternalBroker> getBrokers(KafkaCluster cluster) {
    var stats = statisticsCache.get(cluster);
    var partitionsDistribution = PartitionDistributionStats.create(stats);
    return adminClientService
        .get(cluster)
        .flatMap(ReactiveAdminClient::describeCluster)
        .map(description -> description.getNodes().stream()
            .map(node -> new InternalBroker(node, partitionsDistribution, stats))
            .collect(Collectors.toList()))
        .flatMapMany(Flux::fromIterable);
  }
@Test
  void getBrokersReturnsFilledBrokerDto() {
    var localCluster = clustersStorage.getClusterByName(LOCAL).get();
    StepVerifier.create(brokerService.getBrokers(localCluster))
        .expectNextMatches(b -> b.getId().equals(1)
            && b.getHost().equals(kafka.getHost())
            && b.getPort().equals(kafka.getFirstMappedPort()))
        .verifyComplete();
  }