public Future<Collection<Integer>> resizeAndReconcilePvcs(KafkaStatus kafkaStatus, List<PersistentVolumeClaim> pvcs) {
        Set<Integer> podIdsToRestart = new HashSet<>();
        List<Future<Void>> futures = new ArrayList<>(pvcs.size());

        for (PersistentVolumeClaim desiredPvc : pvcs)  {
            Future<Void> perPvcFuture = pvcOperator.getAsync(reconciliation.namespace(), desiredPvc.getMetadata().getName())
                    .compose(currentPvc -> {
                        if (currentPvc == null || currentPvc.getStatus() == null || !"Bound".equals(currentPvc.getStatus().getPhase())) {
                            // This branch handles the following conditions:
                            // * The PVC doesn't exist yet, we should create it
                            // * The PVC is not Bound, we should reconcile it
                            return pvcOperator.reconcile(reconciliation, reconciliation.namespace(), desiredPvc.getMetadata().getName(), desiredPvc)
                                    .map((Void) null);
                        } else if (currentPvc.getStatus().getConditions().stream().anyMatch(cond -> "Resizing".equals(cond.getType()) && "true".equals(cond.getStatus().toLowerCase(Locale.ENGLISH))))  {
                            // The PVC is Bound, but it is already resizing => Nothing to do, we should let it resize
                            LOGGER.debugCr(reconciliation, "The PVC {} is resizing, nothing to do", desiredPvc.getMetadata().getName());
                            return Future.succeededFuture();
                        } else if (currentPvc.getStatus().getConditions().stream().anyMatch(cond -> "FileSystemResizePending".equals(cond.getType()) && "true".equals(cond.getStatus().toLowerCase(Locale.ENGLISH))))  {
                            // The PVC is Bound and resized but waiting for FS resizing => We need to restart the pod which is using it
                            podIdsToRestart.add(getPodIndexFromPvcName(desiredPvc.getMetadata().getName()));
                            LOGGER.infoCr(reconciliation, "The PVC {} is waiting for file system resizing and the pod using it might need to be restarted.", desiredPvc.getMetadata().getName());
                            return Future.succeededFuture();
                        } else {
                            // The PVC is Bound and resizing is not in progress => We should check if the SC supports resizing and check if size changed
                            Long currentSize = StorageUtils.convertToMillibytes(currentPvc.getSpec().getResources().getRequests().get("storage"));
                            Long desiredSize = StorageUtils.convertToMillibytes(desiredPvc.getSpec().getResources().getRequests().get("storage"));

                            if (!currentSize.equals(desiredSize))   {
                                // The sizes are different => we should resize (shrinking will be handled in StorageDiff, so we do not need to check that)
                                return resizePvc(kafkaStatus, currentPvc, desiredPvc);
                            } else  {
                                // size didn't change, just reconcile
                                return pvcOperator.reconcile(reconciliation, reconciliation.namespace(), desiredPvc.getMetadata().getName(), desiredPvc)
                                        .map((Void) null);
                            }
                        }
                    });

            futures.add(perPvcFuture);
        }

        return Future.all(futures)
                .map(podIdsToRestart);
    }
@Test
    public void testVolumesResized(VertxTestContext context)  {
        List<PersistentVolumeClaim> pvcs = List.of(
                createPvc("data-pod-0"),
                createPvc("data-pod-1"),
                createPvc("data-pod-2")
        );

        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);
                    PersistentVolumeClaim currentPvc = pvcs.stream().filter(pvc -> pvcName.equals(pvc.getMetadata().getName())).findFirst().orElse(null);

                    if (currentPvc != null) {
                        PersistentVolumeClaim pvcWithStatus = new PersistentVolumeClaimBuilder(currentPvc)
                                .withNewStatus()
                                    .withPhase("Bound")
                                    .withCapacity(Map.of("storage", new Quantity("100Gi", null)))
                                .endStatus()
                                .build();

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

        // Mock the StorageClass Operator
        StorageClassOperator mockSco = supplier.storageClassOperations;
        when(mockSco.getAsync(eq(STORAGE_CLASS_NAME))).thenReturn(Future.succeededFuture(RESIZABLE_STORAGE_CLASS));

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

        Checkpoint async = context.checkpoint();
        reconciler.resizeAndReconcilePvcs(new KafkaStatus(), pvcs)
                .onComplete(res -> {
                    assertThat(res.succeeded(), is(true));

                    assertThat(res.result().size(), is(0));

                    assertThat(pvcCaptor.getAllValues().size(), is(3));
                    assertThat(pvcCaptor.getAllValues(), is(pvcs));

                    async.flag();
                });
    }