public Future<Void> reconcile()    {
        LOGGER.infoCr(reconciliation, "Deleting all the ZooKeeper related resources");
        return jmxSecret()
                .compose(i -> deleteNetworkPolicy())
                .compose(i -> deleteServiceAccount())
                .compose(i -> deleteService())
                .compose(i -> deleteHeadlessService())
                .compose(i -> deleteCertificateSecret())
                .compose(i -> deleteLoggingAndMetricsConfigMap())
                .compose(i -> deletePodDisruptionBudget())
                .compose(i -> deletePodSet())
                .compose(i -> deletePersistentClaims());
    }
@Test
    public void testZookeeperEraserReconcileFailedDueToServiceAccountDeletionTimeout(VertxTestContext context) {

        ResourceOperatorSupplier supplier = ResourceUtils.supplierWithMocks(false);
        ServiceAccountOperator mockSaOps = supplier.serviceAccountOperations;
        ServiceOperator mockServiceOps = supplier.serviceOperations;
        NetworkPolicyOperator mockNetPolicyOps = supplier.networkPolicyOperator;
        ConfigMapOperator mockCmOps = supplier.configMapOperations;
        StrimziPodSetOperator mockPodSetOps = supplier.strimziPodSetOperator;
        PodDisruptionBudgetOperator mockPdbOps = supplier.podDisruptionBudgetOperator;
        SecretOperator mockSecretOps = supplier.secretOperations;
        PvcOperator mockPvcOps = supplier.pvcOperations;
        SharedEnvironmentProvider sharedEnvironmentProvider = supplier.sharedEnvironmentProvider;

        ZookeeperCluster zkCluster = ZookeeperCluster.fromCrd(RECONCILIATION, KAFKA, VERSIONS, sharedEnvironmentProvider);

        ArgumentCaptor<String> podSetDeletionCaptor = ArgumentCaptor.forClass(String.class);
        when(mockPodSetOps.deleteAsync(any(), anyString(), podSetDeletionCaptor.capture(), anyBoolean())).thenAnswer(i -> Future.succeededFuture());

        ArgumentCaptor<String> secretDeletionCaptor = ArgumentCaptor.forClass(String.class);
        when(mockSecretOps.deleteAsync(any(), anyString(), secretDeletionCaptor.capture(), anyBoolean())).thenReturn(Future.succeededFuture());

        ArgumentCaptor<String> saDeletionCaptor = ArgumentCaptor.forClass(String.class);
        when(mockSaOps.deleteAsync(any(), anyString(), saDeletionCaptor.capture(), anyBoolean())).thenReturn(Future.failedFuture(new TimeoutException("Timed out while deleting the resource")));

        ArgumentCaptor<String> serviceDeletionCaptor = ArgumentCaptor.forClass(String.class);
        when(mockServiceOps.deleteAsync(any(), anyString(), serviceDeletionCaptor.capture(), anyBoolean())).thenReturn(Future.succeededFuture());

        ArgumentCaptor<String> netPolicyDeletionCaptor = ArgumentCaptor.forClass(String.class);
        when(mockNetPolicyOps.deleteAsync(any(), anyString(), netPolicyDeletionCaptor.capture(), anyBoolean())).thenReturn(Future.succeededFuture());

        ArgumentCaptor<String> cmDeletionCaptor = ArgumentCaptor.forClass(String.class);
        when(mockCmOps.deleteAsync(any(), anyString(), cmDeletionCaptor.capture(), anyBoolean())).thenReturn(Future.succeededFuture());

        ArgumentCaptor<String> pdbDeletionCaptor = ArgumentCaptor.forClass(String.class);
        when(mockPdbOps.deleteAsync(any(), anyString(), pdbDeletionCaptor.capture(), anyBoolean())).thenReturn(Future.succeededFuture());

        // Mock the PVC Operator
        Map<String, PersistentVolumeClaim> zkPvcs = createZooPvcs(NAMESPACE, zkCluster.getStorage(), zkCluster.nodes(),
                (replica, storageId) -> VolumeUtils.DATA_VOLUME_NAME + "-" + KafkaResources.zookeeperPodName(KAFKA.getMetadata().getName(), replica), deleteClaim(KAFKA.getSpec().getZookeeper().getStorage()));

        ArgumentCaptor<PersistentVolumeClaim> pvcCaptor = ArgumentCaptor.forClass(PersistentVolumeClaim.class);
        when(mockPvcOps.reconcile(any(), anyString(), anyString(), pvcCaptor.capture())).thenReturn(Future.succeededFuture());

        when(mockPvcOps.getAsync(anyString(), ArgumentMatchers.startsWith("data-")))
                .thenAnswer(invocation -> {
                    String pvcName = invocation.getArgument(1);
                    if (pvcName.contains(zkCluster.getComponentName())) {
                        return Future.succeededFuture(zkPvcs.get(pvcName));
                    }
                    return Future.succeededFuture(null);
                });
        when(mockPvcOps.listAsync(anyString(), ArgumentMatchers.any(Labels.class)))
                .thenAnswer(invocation -> Future.succeededFuture(zkPvcs.values().stream().toList()));

        // test reconcile
        ZooKeeperEraser zkEraser = new ZooKeeperEraser(
                RECONCILIATION,
                supplier
        );

        Checkpoint async = context.checkpoint();
        zkEraser.reconcile()
                .onComplete(context.failing(v -> context.verify(() -> {
                    verify(mockCmOps, never()).deleteAsync(any(), any(), any(), anyBoolean());
                    verify(mockSaOps, times(1)).deleteAsync(any(), any(), any(), anyBoolean());
                    verify(mockServiceOps, never()).deleteAsync(any(), any(), any(), anyBoolean());
                    verify(mockSecretOps, times(1)).deleteAsync(any(), any(), any(), anyBoolean());
                    verify(mockNetPolicyOps, times(1)).deleteAsync(any(), any(), any(), anyBoolean());
                    verify(mockPodSetOps, never()).deleteAsync(any(), any(), any(), anyBoolean());
                    verify(mockPdbOps, never()).deleteAsync(any(), any(), any(), anyBoolean());
                    verify(mockPvcOps, never()).getAsync(any(), any());
                    verify(mockPvcOps, never()).listAsync(any(), ArgumentMatchers.any(Labels.class));
                    // no reconcile since there was no PVC deletion
                    verify(mockPvcOps, never()).reconcile(any(), any(), any(), any());
                    assertThat(pvcCaptor.getAllValues().size(), is(0));

                    assertThat(netPolicyDeletionCaptor.getAllValues(), is(List.of("my-cluster-network-policy-zookeeper")));
                    assertThat(saDeletionCaptor.getAllValues(), is(List.of("my-cluster-zookeeper")));
                    assertThat(secretDeletionCaptor.getAllValues(), is(List.of("my-cluster-zookeeper-jmx")));

                    // asserting error message
                    assertThat(v.getMessage(), is("Timed out while deleting the resource"));
                    async.flag();
                })));
    }