private Mono<Collection<TopicPartition>> filterPartitionsWithLeaderCheck(Collection<TopicPartition> partitions,
                                                                           boolean failOnUnknownLeader) {
    var targetTopics = partitions.stream().map(TopicPartition::topic).collect(Collectors.toSet());
    return describeTopicsImpl(targetTopics)
        .map(descriptions ->
            filterPartitionsWithLeaderCheck(
                descriptions.values(), partitions::contains, failOnUnknownLeader));
  }
@Test
  void filterPartitionsWithLeaderCheckSkipsPartitionsFromTopicWhereSomePartitionsHaveNoLeader() {
    var filteredPartitions = ReactiveAdminClient.filterPartitionsWithLeaderCheck(
        List.of(
            // contains partitions with no leader
            new TopicDescription("noLeaderTopic", false,
                List.of(
                    new TopicPartitionInfo(0, new Node(1, "n1", 9092), List.of(), List.of()),
                    new TopicPartitionInfo(1, null, List.of(), List.of()))),
            // should be skipped by predicate
            new TopicDescription("skippingByPredicate", false,
                List.of(
                    new TopicPartitionInfo(0, new Node(1, "n1", 9092), List.of(), List.of()))),
            // good topic
            new TopicDescription("good", false,
                List.of(
                    new TopicPartitionInfo(0, new Node(1, "n1", 9092), List.of(), List.of()),
                    new TopicPartitionInfo(1, new Node(2, "n2", 9092), List.of(), List.of()))
            )),
        p -> !p.topic().equals("skippingByPredicate"),
        false
    );

    assertThat(filteredPartitions)
        .containsExactlyInAnyOrder(
            new TopicPartition("good", 0),
            new TopicPartition("good", 1)
        );
  }