List<Condition> run(boolean useKRaft) {
        List<Condition> warnings = new ArrayList<>();

        checkKafkaReplicationConfig(warnings);
        checkKafkaBrokersStorage(warnings);

        if (useKRaft)   {
            // Additional checks done for KRaft clusters
            checkKRaftControllerStorage(warnings);
            checkKRaftControllerCount(warnings);
            checkKafkaMetadataVersion(warnings);
            checkInterBrokerProtocolVersionInKRaft(warnings);
            checkLogMessageFormatVersionInKRaft(warnings);
        } else {
            // Additional checks done for ZooKeeper-based clusters
            checkKafkaLogMessageFormatVersion(warnings);
            checkKafkaInterBrokerProtocolVersion(warnings);
            checkKRaftMetadataStorageConfiguredForZooBasedCLuster(warnings);
        }

        return warnings;
    }
@Test
    public void testKRaftWithTwoControllers() {
        KafkaNodePool controllers = new KafkaNodePoolBuilder(CONTROLLERS)
                .editSpec()
                    .withReplicas(2)
                .endSpec()
                .build();

        KafkaSpecChecker checker = generateChecker(KAFKA, List.of(controllers, POOL_A), KafkaVersionTestUtils.DEFAULT_KRAFT_VERSION_CHANGE);

        List<Condition> warnings = checker.run(true);

        assertThat(warnings, hasSize(1));
        assertThat(warnings.get(0).getReason(), is("KafkaKRaftControllerNodeCount"));
        assertThat(warnings.get(0).getMessage(), is("Running KRaft controller quorum with two nodes is not advisable as both nodes will be needed to avoid downtime. It is recommended that a minimum of three nodes are used."));
    }