public Future<KafkaCluster> prepareKafkaCluster(
            Kafka kafkaCr,
            List<KafkaNodePool> nodePools,
            Map<String, Storage> oldStorage,
            Map<String, List<String>> currentPods,
            KafkaVersionChange versionChange,
            KafkaStatus kafkaStatus,
            boolean tryToFixProblems)   {
        return createKafkaCluster(kafkaCr, nodePools, oldStorage, currentPods, versionChange)
                .compose(kafka -> brokerRemovalCheck(kafkaCr, kafka))
                .compose(kafka -> {
                    if (checkFailed() && tryToFixProblems)   {
                        // We have a failure, and should try to fix issues
                        // Once we fix it, we call this method again, but this time with tryToFixProblems set to false
                        return revertScaleDown(kafka, kafkaCr, nodePools)
                                .compose(kafkaAndNodePools -> revertRoleChange(kafkaAndNodePools.kafkaCr(), kafkaAndNodePools.nodePoolCrs()))
                                .compose(kafkaAndNodePools -> prepareKafkaCluster(kafkaAndNodePools.kafkaCr(), kafkaAndNodePools.nodePoolCrs(), oldStorage, currentPods, versionChange, kafkaStatus, false));
                    } else if (checkFailed()) {
                        // We have a failure, but we should not try to fix it
                        List<String> errors = new ArrayList<>();

                        if (scaleDownCheckFailed)   {
                            errors.add("Cannot scale-down Kafka brokers " + kafka.removedNodes() + " because they have assigned partition-replicas.");
                        }

                        if (usedToBeBrokersCheckFailed) {
                            errors.add("Cannot remove the broker role from nodes " + kafka.usedToBeBrokerNodes() + " because they have assigned partition-replicas.");
                        }

                        return Future.failedFuture(new InvalidResourceException("Following errors were found when processing the Kafka custom resource: " + errors));
                    } else {
                        // If everything succeeded, we return the KafkaCluster object
                        // If any warning conditions exist from the reverted changes, we add them to the status
                        if (!warningConditions.isEmpty())   {
                            kafkaStatus.addConditions(warningConditions);
                        }

                        return Future.succeededFuture(kafka);
                    }
                });
    }
@Test
    public void testRevertScaleDownWithKRaft(VertxTestContext context) {
        ResourceOperatorSupplier supplier = ResourceUtils.supplierWithMocks(false);

        // Mock brokers-in-use check
        BrokersInUseCheck brokersInUseOps = supplier.brokersInUseCheck;
        when(brokersInUseOps.brokersInUse(any(), any(), any(), any())).thenReturn(Future.succeededFuture(Set.of(1000, 1001, 1002, 1003, 2004)));

        KafkaStatus kafkaStatus = new KafkaStatus();
        KafkaClusterCreator creator = new KafkaClusterCreator(vertx, RECONCILIATION, CO_CONFIG, KafkaMetadataConfigurationState.KRAFT, supplier);

        Checkpoint async = context.checkpoint();
        creator.prepareKafkaCluster(KAFKA, List.of(POOL_CONTROLLERS_WITH_STATUS_5_NODES, POOL_A_WITH_STATUS_5_NODES, POOL_B_WITH_STATUS_5_NODES), Map.of(), null, KafkaVersionTestUtils.DEFAULT_KRAFT_VERSION_CHANGE, kafkaStatus, true)
                .onComplete(context.succeeding(kc -> context.verify(() -> {
                    // Kafka cluster is created
                    assertThat(kc, is(notNullValue()));
                    assertThat(kc.nodes().size(), is(13));
                    assertThat(kc.nodes().stream().map(NodeRef::nodeId).collect(Collectors.toSet()), is(Set.of(1000, 1001, 1002, 1003, 1004, 2000, 2001, 2002, 2003, 2004, 3000, 3001, 3002)));
                    assertThat(kc.removedNodes(), is(Set.of(3003, 3004))); // Controllers are not affected

                    // Check the status conditions
                    assertThat(kafkaStatus.getConditions().size(), is(2));
                    assertThat(kafkaStatus.getConditions().get(0).getStatus(), is("True"));
                    assertThat(kafkaStatus.getConditions().get(0).getType(), is("Warning"));
                    assertThat(kafkaStatus.getConditions().get(0).getReason(), is("ScaleDownPreventionCheck"));
                    assertThat(kafkaStatus.getConditions().get(0).getMessage(), is("Reverting scale-down of KafkaNodePool pool-a by changing number of replicas to 5"));
                    assertThat(kafkaStatus.getConditions().get(1).getStatus(), is("True"));
                    assertThat(kafkaStatus.getConditions().get(1).getType(), is("Warning"));
                    assertThat(kafkaStatus.getConditions().get(1).getReason(), is("ScaleDownPreventionCheck"));
                    assertThat(kafkaStatus.getConditions().get(1).getMessage(), is("Reverting scale-down of KafkaNodePool pool-b by changing number of replicas to 5"));

                    // Scale-down reverted => should be called twice as we still scale down controllers after the revert is done
                    verify(supplier.brokersInUseCheck, times(2)).brokersInUse(any(), any(), any(), any());

                    async.flag();
                })));
    }