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 testListTopicsFailure(VertxTestContext context) {
        Admin admin = mock(Admin.class);
        AdminClientProvider mock = mock(AdminClientProvider.class);
        when(mock.createAdminClient(anyString(), any(), any())).thenReturn(admin);

        // Mock list topics
        @SuppressWarnings(value = "unchecked")
        KafkaFuture<Set<String>> kf = mock(KafkaFuture.class);
        when(kf.whenComplete(any())).thenAnswer(i -> {
            KafkaFuture.BiConsumer<Void, Throwable> action = i.getArgument(0);
            action.accept(null, new Throwable("Test error ..."));
            return null;
        });
        ListTopicsResult ltr = mock(ListTopicsResult.class);
        when(ltr.names()).thenReturn(kf);
        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.failing(e -> {
                    assertThat(e.getMessage(), is("Test error ..."));

                    checkpoint.flag();
                }));
    }