public static Future<Void> unregisterNodes(
            Reconciliation reconciliation,
            Vertx vertx,
            AdminClientProvider adminClientProvider,
            PemTrustSet pemTrustSet,
            PemAuthIdentity pemAuthIdentity,
            List<Integer> nodeIdsToUnregister
    ) {
        try {
            String bootstrapHostname = KafkaResources.bootstrapServiceName(reconciliation.name()) + "." + reconciliation.namespace() + ".svc:" + KafkaCluster.REPLICATION_PORT;
            Admin adminClient = adminClientProvider.createAdminClient(bootstrapHostname, pemTrustSet, pemAuthIdentity);

            List<Future<Void>> futures = new ArrayList<>();
            for (Integer nodeId : nodeIdsToUnregister) {
                futures.add(unregisterNode(reconciliation, vertx, adminClient, nodeId));
            }

            return Future.all(futures)
                    .eventually(() -> {
                        adminClient.close();
                        return Future.succeededFuture();
                    })
                    .map((Void) null);
        } catch (KafkaException e) {
            LOGGER.warnCr(reconciliation, "Failed to unregister nodes", e);
            return Future.failedFuture(e);
        }
    }
@Test
    void testFailingNodeUnregistration(VertxTestContext context) {
        Admin mockAdmin = ResourceUtils.adminClient();

        ArgumentCaptor<Integer> unregisteredNodeIdCaptor = ArgumentCaptor.forClass(Integer.class);
        when(mockAdmin.unregisterBroker(unregisteredNodeIdCaptor.capture())).thenAnswer(i -> {
            if (i.getArgument(0, Integer.class) == 1919)   {
                KafkaFutureImpl<Void> unregistrationFuture = new KafkaFutureImpl<>();
                unregistrationFuture.completeExceptionally(new TimeoutException("Fake timeout"));
                UnregisterBrokerResult ubr = mock(UnregisterBrokerResult.class);
                when(ubr.all()).thenReturn(unregistrationFuture);
                return ubr;
            } else {
                UnregisterBrokerResult ubr = mock(UnregisterBrokerResult.class);
                when(ubr.all()).thenReturn(KafkaFuture.completedFuture(null));
                return ubr;
            }
        });

        AdminClientProvider mockProvider = ResourceUtils.adminClientProvider(mockAdmin);

        Checkpoint async = context.checkpoint();
        KafkaNodeUnregistration.unregisterNodes(Reconciliation.DUMMY_RECONCILIATION, vertx, mockProvider, null, null, List.of(1874, 1919))
                .onComplete(context.failing(v -> context.verify(() -> {
                    assertThat(unregisteredNodeIdCaptor.getAllValues().size(), is(2));
                    assertThat(unregisteredNodeIdCaptor.getAllValues(), hasItems(1874, 1919));

                    async.flag();
                })));
    }