public Future<Void> maybeRollingUpdate(Reconciliation reconciliation, int replicas, Labels selectorLabels, Function<Pod, List<String>> podRestart, TlsPemIdentity coTlsPemIdentity) {
        String namespace = reconciliation.namespace();

        // We prepare the list of expected Pods. This is needed as we need to account for pods which might be missing.
        // We need to wait for them before rolling any running pods to avoid problems.
        List<String> expectedPodNames = new ArrayList<>();
        for (int i = 0; i < replicas; i++)  {
            expectedPodNames.add(KafkaResources.zookeeperPodName(reconciliation.name(), i));
        }

        return podOperator.listAsync(namespace, selectorLabels)
                .compose(pods -> {
                    ZookeeperClusterRollContext clusterRollContext = new ZookeeperClusterRollContext();

                    for (String podName : expectedPodNames) {
                        Pod pod = pods.stream().filter(p -> podName.equals(p.getMetadata().getName())).findFirst().orElse(null);

                        if (pod != null)    {
                            List<String> restartReasons = podRestart.apply(pod);
                            final boolean ready = podOperator.isReady(namespace, pod.getMetadata().getName());
                            ZookeeperPodContext podContext = new ZookeeperPodContext(podName, restartReasons, true, ready);
                            if (restartReasons != null && !restartReasons.isEmpty())    {
                                LOGGER.debugCr(reconciliation, "Pod {} should be rolled due to {}", podContext.getPodName(), restartReasons);
                            } else {
                                LOGGER.debugCr(reconciliation, "Pod {} does not need to be rolled", podContext.getPodName());
                            }
                            clusterRollContext.add(podContext);
                        } else {
                            // Pod does not exist, but we still add it to the roll context because we should not roll
                            // any other pods before it is ready
                            LOGGER.debugCr(reconciliation, "Pod {} does not exist and cannot be rolled", podName);
                            ZookeeperPodContext podContext = new ZookeeperPodContext(podName, null, false, false);
                            clusterRollContext.add(podContext);
                        }
                    }

                    if (clusterRollContext.requiresRestart())   {
                        return Future.succeededFuture(clusterRollContext);
                    } else {
                        return Future.succeededFuture(null);
                    }
                }).compose(clusterRollContext -> {
                    if (clusterRollContext != null)  {
                        Promise<Void> promise = Promise.promise();
                        Future<String> leaderFuture = leaderFinder.findZookeeperLeader(reconciliation, clusterRollContext.podNames(), coTlsPemIdentity);

                        leaderFuture.compose(leader -> {
                            LOGGER.debugCr(reconciliation, "Zookeeper leader is " + (ZookeeperLeaderFinder.UNKNOWN_LEADER.equals(leader) ? "unknown" : "pod " + leader));
                            Future<Void> fut = Future.succeededFuture();

                            // Then roll each non-leader pod => the leader is rolled last
                            for (ZookeeperPodContext podContext : clusterRollContext.getPodContextsWithNonExistingAndNonReadyFirst())  {
                                if (podContext.requiresRestart() && !podContext.getPodName().equals(leader)) {
                                    LOGGER.debugCr(reconciliation, "Pod {} needs to be restarted", podContext.getPodName());
                                    // roll the pod and wait until it is ready
                                    // this prevents rolling into faulty state (note: this applies just for ZK pods)
                                    fut = fut.compose(ignore -> restartPod(reconciliation, podContext.getPodName(), podContext.reasonsToRestart));
                                } else {
                                    if (podContext.requiresRestart()) {
                                        LOGGER.debugCr(reconciliation, "Deferring restart of leader {}", podContext.getPodName());
                                    } else {
                                        LOGGER.debugCr(reconciliation, "Pod {} does not need to be restarted", podContext.getPodName());
                                    }
                                    fut = fut.compose(ignore -> podOperator.readiness(reconciliation, reconciliation.namespace(), podContext.getPodName(), READINESS_POLLING_INTERVAL_MS, operationTimeoutMs));
                                }
                            }

                            // Check if we have a leader and if it needs rolling
                            if (ZookeeperLeaderFinder.UNKNOWN_LEADER.equals(leader) || clusterRollContext.get(leader) == null || !clusterRollContext.get(leader).requiresRestart()) {
                                return fut;
                            } else {
                                // Roll the leader pod
                                return fut.compose(ar -> {
                                    // the leader is rolled as the last
                                    LOGGER.debugCr(reconciliation, "Restarting leader pod (previously deferred) {}", leader);
                                    return restartPod(reconciliation, leader, clusterRollContext.get(leader).reasonsToRestart);
                                });
                            }
                        }).onComplete(promise);

                        return promise.future();
                    } else {
                        return Future.succeededFuture();
                    }
                });
    }
@Test
    public void testAllPodsAreRolled(VertxTestContext context)  {
        PodOperator podOperator = mock(PodOperator.class);
        when(podOperator.listAsync(any(), any(Labels.class))).thenReturn(Future.succeededFuture(PODS));

        ZookeeperLeaderFinder leaderFinder = mock(ZookeeperLeaderFinder.class);
        when(leaderFinder.findZookeeperLeader(any(), any(), any())).thenReturn(Future.succeededFuture(ZookeeperLeaderFinder.UNKNOWN_LEADER));

        MockZooKeeperRoller roller = new MockZooKeeperRoller(podOperator, leaderFinder, 300_00L);

        roller.maybeRollingUpdate(Reconciliation.DUMMY_RECONCILIATION, 3, DUMMY_SELECTOR, pod -> List.of("Should restart"), DUMMY_IDENTITY)
                .onComplete(context.succeeding(v -> context.verify(() -> {
                    assertThat(roller.podRestarts.size(), is(3));
                    assertThat(roller.podRestarts.contains("name-zookeeper-0"), is(true));
                    assertThat(roller.podRestarts.contains("name-zookeeper-1"), is(true));
                    assertThat(roller.podRestarts.contains("name-zookeeper-2"), is(true));

                    context.completeNow();
                })));
    }