Future<Boolean> canRollController(int nodeId) {
        LOGGER.debugCr(reconciliation, "Determining whether controller pod {} can be rolled", nodeId);
        return describeMetadataQuorum().map(info -> {
            boolean canRoll = isQuorumHealthyWithoutNode(nodeId, info);
            if (!canRoll) {
                LOGGER.debugCr(reconciliation, "Not restarting controller pod {}. Restart would affect the quorum health", nodeId);
            }
            return canRoll;
        }).recover(error -> {
            LOGGER.warnCr(reconciliation, "Error determining whether it is safe to restart controller pod {}", nodeId, error);
            return Future.failedFuture(error);
        });
    }
@Test
    public void canRollControllerWhenItselfIsBehind(VertxTestContext context) {
        Map<Integer, OptionalLong> controllers = new HashMap<>();
        controllers.put(1, OptionalLong.of(10000L));
        controllers.put(2, OptionalLong.of(7000L));
        controllers.put(3, OptionalLong.of(10000L));
        Admin admin = setUpMocks(1, controllers);
        KafkaQuorumCheck quorumCheck = new KafkaQuorumCheck(Reconciliation.DUMMY_RECONCILIATION, admin, vertx, CONTROLLER_QUORUM_FETCH_TIMEOUT_MS);
        quorumCheck.canRollController(2).onComplete(context.succeeding(result -> {
            context.verify(() -> assertTrue(result));
            context.completeNow();
        }));
    }