private KafkaRebalanceStatus updateStatus(KafkaRebalance kafkaRebalance,
                                              KafkaRebalanceStatus desiredStatus,
                                              Throwable e) {
        // Leave the current status when the desired state is null
        if (desiredStatus != null) {

            Condition cond = KafkaRebalanceUtils.rebalanceStateCondition(desiredStatus);

            List<Condition> previous = Collections.emptyList();
            if (desiredStatus.getConditions() != null) {
                previous = desiredStatus.getConditions().stream().filter(condition -> condition != cond).collect(Collectors.toList());
            }

            // If a throwable is supplied, it is set in the status with priority
            if (e != null) {
                StatusUtils.setStatusConditionAndObservedGeneration(kafkaRebalance, desiredStatus, KafkaRebalanceState.NotReady.toString(), e);
                desiredStatus.setConditions(Stream.concat(desiredStatus.getConditions().stream(), previous.stream()).collect(Collectors.toList()));
            } else if (cond != null) {
                StatusUtils.setStatusConditionAndObservedGeneration(kafkaRebalance, desiredStatus, cond);
                desiredStatus.setConditions(Stream.concat(desiredStatus.getConditions().stream(), previous.stream()).collect(Collectors.toList()));
            } else {
                throw new IllegalArgumentException("Status related exception and the Status condition's type cannot both be null");
            }
            return desiredStatus;
        }
        return kafkaRebalance.getStatus();
    }
@Test
    public void shouldGenerateNewProposalWhenUserTaskNotFound(VertxTestContext context) throws IOException, URISyntaxException {
        cruiseControlServer.setupCCRebalanceResponse(2, CruiseControlEndpoints.REBALANCE);

        KafkaRebalance kr = new KafkaRebalanceBuilder(createKafkaRebalance(namespace, CLUSTER_NAME, RESOURCE_NAME, EMPTY_KAFKA_REBALANCE_SPEC, true))
                .withNewStatus()
                .withObservedGeneration(1L)
                .withConditions(new ConditionBuilder()
                        .withType("ProposalReady")
                        .withStatus("True")
                        .build())
                .endStatus()
                .build();

        Crds.kafkaRebalanceOperation(client).inNamespace(namespace).resource(kr).create();
        Crds.kafkaRebalanceOperation(client).inNamespace(namespace).resource(kr).updateStatus();

        crdCreateKafka();
        crdCreateCruiseControlSecrets();

        Checkpoint checkpoint = context.checkpoint();
        krao.reconcile(new Reconciliation("test-trigger", KafkaRebalance.RESOURCE_KIND, namespace, RESOURCE_NAME))
                .onComplete(context.succeeding(v -> {
                    // the resource moved to `Rebalancing` since auto-approval annotation is set on the resource
                    assertState(context, client, namespace, RESOURCE_NAME, KafkaRebalanceState.Rebalancing);
                }))
                .compose(v -> {
                    // Sets a user_tasks response with an empty task list simulating CC restart
                    cruiseControlServer.setupUserTasktoEmpty();
                    return krao.reconcile(new Reconciliation("test-trigger", KafkaRebalance.RESOURCE_KIND, namespace, kr.getMetadata().getName()));
                })
                .onComplete(context.succeeding(v -> {
                    // the resource transitioned from 'Rebalancing' to 'PendingProposal'
                    assertState(context, client, namespace, RESOURCE_NAME, KafkaRebalanceState.PendingProposal);
                    checkpoint.flag();
                }));
    }