public Future<Void> deletePersistentClaims(List<String> maybeDeletePvcs, List<String> desiredPvcs) {
        List<Future<Void>> futures = new ArrayList<>();

        maybeDeletePvcs.removeAll(desiredPvcs);

        for (String pvcName : maybeDeletePvcs)  {
            LOGGER.debugCr(reconciliation, "Considering PVC {} for deletion", pvcName);
            futures.add(considerPersistentClaimDeletion(pvcName));
        }

        return Future.all(futures)
                .map((Void) null);
    }
@Test
    public void testVolumesDeletion(VertxTestContext context)  {
        PersistentVolumeClaim pvcWithDeleteClaim = new PersistentVolumeClaimBuilder(createPvc("data-pod-3"))
                .editMetadata()
                    .withAnnotations(Map.of(Annotations.ANNO_STRIMZI_IO_DELETE_CLAIM, "true"))
                .endMetadata()
                .build();

        List<String> desiredPvcs = List.of(
                "data-pod-0",
                "data-pod-1",
                "data-pod-2"
        );

        List<String> maybeDeletePvcs = List.of(
                "data-pod-0",
                "data-pod-1",
                "data-pod-2",
                "data-pod-3",
                "data-pod-4"
        );

        List<PersistentVolumeClaim> pvcs = List.of(
                createPvc("data-pod-0"),
                createPvc("data-pod-1"),
                createPvc("data-pod-2"),
                pvcWithDeleteClaim,
                createPvc("data-pod-4")
        );

        ResourceOperatorSupplier supplier = ResourceUtils.supplierWithMocks(false);

        // Mock the PVC Operator
        PvcOperator mockPvcOps = supplier.pvcOperations;
        when(mockPvcOps.getAsync(eq(NAMESPACE), ArgumentMatchers.startsWith("data-")))
                .thenAnswer(invocation -> {
                    String pvcName = invocation.getArgument(1);
                    return Future.succeededFuture(pvcs.stream().filter(pvc -> pvcName.equals(pvc.getMetadata().getName())).findFirst().orElse(null));
                });
        ArgumentCaptor<String> pvcNameCaptor = ArgumentCaptor.forClass(String.class);
        ArgumentCaptor<PersistentVolumeClaim> pvcCaptor = ArgumentCaptor.forClass(PersistentVolumeClaim.class);
        when(mockPvcOps.reconcile(any(), anyString(), pvcNameCaptor.capture(), pvcCaptor.capture())).thenReturn(Future.succeededFuture());

        // Reconcile the PVCs
        PvcReconciler reconciler = new PvcReconciler(
                new Reconciliation("test-trigger", Kafka.RESOURCE_KIND, NAMESPACE, CLUSTER_NAME),
                mockPvcOps,
                supplier.storageClassOperations
        );

        Checkpoint async = context.checkpoint();
        reconciler.deletePersistentClaims(new ArrayList<>(maybeDeletePvcs), new ArrayList<>(desiredPvcs))
                .onComplete(res -> {
                    assertThat(res.succeeded(), is(true));

                    assertThat(pvcNameCaptor.getAllValues().size(), is(1));
                    assertThat(pvcNameCaptor.getValue(), is("data-pod-3"));

                    assertThat(pvcCaptor.getAllValues().size(), is(1));
                    assertThat(pvcCaptor.getValue(), is(nullValue()));

                    async.flag();
                });
    }