test.java

    @Test
    public void testScaleDownWithPartitionReplicasAndEnabledBrokerCheck(VertxTestContext context)  {

        Kafka patchKafka = new KafkaBuilder(KAFKA)
                .editMetadata()
                .addToAnnotations(Map.of(ANNO_STRIMZI_IO_BYPASS_BROKER_SCALEDOWN_CHECK, "false"))
                .endMetadata()
                .build();
        

        ResourceOperatorSupplier supplier = ResourceUtils.supplierWithMocks(false);

        // Mock the secrets needed for Kafka client
        SecretOperator mockSecretOps = supplier.secretOperations;
        Secret secret = new Secret();
        when(mockSecretOps.getAsync(eq(NAMESPACE), eq(KafkaResources.clusterCaCertificateSecretName(CLUSTER_NAME)))).thenReturn(Future.succeededFuture(secret));
        when(mockSecretOps.getAsync(eq(NAMESPACE), eq(KafkaResources.secretName(CLUSTER_NAME)))).thenReturn(Future.succeededFuture(secret));

        PreventBrokerScaleDownOperations operations = supplier.brokerScaleDownOperations;
        when(operations.canScaleDownBrokers(any(),any(), any(), any(), any())).thenReturn(Future.succeededFuture(Set.of(3)));

        // Run the test
        MockKafkaReconciler kr = new MockKafkaReconciler(
                new Reconciliation("test-trigger", Kafka.RESOURCE_KIND, NAMESPACE, CLUSTER_NAME),
                vertx,
                CONFIG,
                supplier,
                new PlatformFeaturesAvailability(false, KUBERNETES_VERSION),
                patchKafka,
                List.of(POOL_A, POOL_B),
                VERSION_CHANGE,
                Map.of(),
                Map.of(CLUSTER_NAME + "-kafka", IntStream.rangeClosed(0, 4).mapToObj(i -> CLUSTER_NAME + "-kafka-" + i).toList()),
                CLUSTER_CA,
                CLIENTS_CA);


        Checkpoint async = context.checkpoint();

            kr.reconcile(patchKafka.getStatus(), Clock.systemUTC())
                    .onComplete(context.failing(v -> context.verify(() -> {

                        async.flag();
                    })));

    }