public Future<Set<Integer>> brokersInUse(Reconciliation reconciliation, Vertx vertx, TlsPemIdentity coTlsPemIdentity, AdminClientProvider adminClientProvider) {
        try {
            String bootstrapHostname = KafkaResources.bootstrapServiceName(reconciliation.name()) + "." + reconciliation.namespace() + ".svc:" + KafkaCluster.REPLICATION_PORT;
            LOGGER.debugCr(reconciliation, "Creating AdminClient for Kafka cluster in namespace {}", reconciliation.namespace());
            Admin kafkaAdmin = adminClientProvider.createAdminClient(bootstrapHostname, coTlsPemIdentity.pemTrustSet(), coTlsPemIdentity.pemAuthIdentity());

            return topicNames(reconciliation, vertx, kafkaAdmin)
                    .compose(names -> describeTopics(reconciliation, vertx, kafkaAdmin, names))
                    .compose(topicDescriptions -> {
                        Set<Integer> brokersWithPartitionReplicas = new HashSet<>();

                        for (TopicDescription td : topicDescriptions.values()) {
                            for (TopicPartitionInfo pd : td.partitions()) {
                                for (org.apache.kafka.common.Node broker : pd.replicas()) {
                                    brokersWithPartitionReplicas.add(broker.id());
                                }
                            }
                        }

                        kafkaAdmin.close();
                        return Future.succeededFuture(brokersWithPartitionReplicas);
                    }).recover(error -> {
                        LOGGER.warnCr(reconciliation, "Failed to get list of brokers in use", error);
                        kafkaAdmin.close();
                        return Future.failedFuture(error);
                    });
        } catch (KafkaException e) {
            LOGGER.warnCr(reconciliation, "Failed to check if broker contains any partition replicas", e);
            return Future.failedFuture(e);
        }
    }
@Test
    public void testBrokersInUse(VertxTestContext context) {
        Admin admin = mock(Admin.class);
        AdminClientProvider mock = mock(AdminClientProvider.class);
        when(mock.createAdminClient(anyString(), any(), any())).thenReturn(admin);

        // Mock topic description
        TopicDescription t1 = new TopicDescription("my-topic", false, List.of(new TopicPartitionInfo(0, NODE.apply(0), List.of(NODE.apply(0)), List.of(NODE.apply(0)))));
        TopicDescription t2 = new TopicDescription("my-topic2", false, List.of(new TopicPartitionInfo(0, NODE.apply(1), List.of(NODE.apply(1)), List.of(NODE.apply(1)))));
        TopicDescription t3 = new TopicDescription("my-topic3", false, List.of(new TopicPartitionInfo(0, NODE.apply(2), List.of(NODE.apply(2)), List.of(NODE.apply(2)))));
        DescribeTopicsResult dtr = mock(DescribeTopicsResult.class);
        when(dtr.allTopicNames()).thenReturn(KafkaFuture.completedFuture(Map.of(t1.name(), t1, t2.name(), t2, t3.name(), t3)));

        @SuppressWarnings(value = "unchecked")
        ArgumentCaptor<Collection<String>> topicListCaptor = ArgumentCaptor.forClass(Collection.class);
        when(admin.describeTopics(topicListCaptor.capture())).thenReturn(dtr);

        // Mock list topics
        ListTopicsResult ltr = mock(ListTopicsResult.class);
        when(ltr.names()).thenReturn(KafkaFuture.completedFuture(Set.of("my-topic", "my-topic2", "my-topic3")));
        when(admin.listTopics(any())).thenReturn(ltr);

        // Get brokers in use
        Checkpoint checkpoint = context.checkpoint();
        BrokersInUseCheck operations = new BrokersInUseCheck();
        operations.brokersInUse(RECONCILIATION, vertx, DUMMY_IDENTITY, mock)
                .onComplete(context.succeeding(brokersInUse -> {
                    Collection<String> topicList = topicListCaptor.getValue();
                    assertThat(topicList.size(), is(3));
                    assertThat(topicList, hasItems("my-topic", "my-topic2", "my-topic3"));

                    assertThat(brokersInUse.size(), is(3));
                    assertThat(brokersInUse, is(Set.of(0, 1, 2)));

                    checkpoint.flag();
                }));
    }