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 checkWithoutKRaftMetadataConfigInKRaftModeProducesNoWarning() {
        // Kafka with Ephemeral storage
        KafkaNodePool ephemeralPool = new KafkaNodePoolBuilder(POOL_A)
                .editSpec()
                    .withNewEphemeralStorage()
                    .endEphemeralStorage()
                .endSpec()
                .build();

        KafkaSpecChecker checker = generateChecker(KAFKA, List.of(CONTROLLERS, ephemeralPool), KafkaVersionTestUtils.DEFAULT_KRAFT_VERSION_CHANGE);

        List<Condition> warnings = checker.run(true);
        assertThat(warnings, hasSize(0));

        // Kafka with Persistent storage
        KafkaNodePool persistentPool = new KafkaNodePoolBuilder(POOL_A)
                .editSpec()
                    .withNewPersistentClaimStorage()
                        .withSize("100Gi")
                    .endPersistentClaimStorage()
                .endSpec()
                .build();

        checker = generateChecker(KAFKA, List.of(CONTROLLERS, persistentPool), KafkaVersionTestUtils.DEFAULT_KRAFT_VERSION_CHANGE);

        warnings = checker.run(true);
        assertThat(warnings, hasSize(0));

        // Kafka with JBOD storage
        KafkaNodePool jbodPool = new KafkaNodePoolBuilder(POOL_A)
                .editSpec()
                    .withNewJbodStorage()
                        .addNewPersistentClaimStorageVolume()
                            .withId(0)
                            .withSize("100Gi")
                        .endPersistentClaimStorageVolume()
                        .addNewPersistentClaimStorageVolume()
                            .withId(0)
                            .withSize("100Gi")
                        .endPersistentClaimStorageVolume()
                    .endJbodStorage()
                .endSpec()
                .build();

        checker = generateChecker(KAFKA, List.of(CONTROLLERS, jbodPool), KafkaVersionTestUtils.DEFAULT_KRAFT_VERSION_CHANGE);

        warnings = checker.run(true);
        assertThat(warnings, hasSize(0));
    }