KafkaSpecChecker { public List<Condition> run() { List<Condition> warnings = new ArrayList<>(); checkKafkaLogMessageFormatVersion(warnings); checkKafkaStorage(warnings); checkZooKeeperStorage(warnings); checkZooKeeperReplicas(warnings); return warnings; } KafkaSpecChecker(KafkaSpec spec, KafkaCluster kafkaCluster, ZookeeperCluster zkCluster); List<Condition> run();  }
@Test public void checkEmptyWarnings() { Kafka kafka = ResourceUtils.createKafka(NAMESPACE, NAME, 3, IMAGE, HEALTH_DELAY, HEALTH_TIMEOUT); KafkaSpecChecker checker = generateChecker(kafka); assertThat(checker.run(), empty()); }
@Test public void checkKafkaJbodStorage() { Kafka kafka = new KafkaBuilder(ResourceUtils.createKafka(NAMESPACE, NAME, 1, IMAGE, HEALTH_DELAY, HEALTH_TIMEOUT, Collections.emptyMap(), Collections.emptyMap(), Collections.emptyMap(), new JbodStorageBuilder().withVolumes(new EphemeralStorageBuilder().withId(1).build(), new EphemeralStorageBuilder().withId(2).build()).build(), new EphemeralStorage(), null, null, null, null)) .editSpec() .editZookeeper() .withReplicas(3) .endZookeeper() .endSpec() .build(); KafkaSpecChecker checker = generateChecker(kafka); List<Condition> warnings = checker.run(); assertThat(warnings, hasSize(1)); Condition warning = warnings.get(0); assertThat(warning.getReason(), is("KafkaStorage")); assertThat(warning.getStatus(), is("True")); assertThat(warning.getMessage(), is("A Kafka cluster with a single replica and ephemeral storage will lose topic messages after any restart or rolling update.")); }
@Test public void checkZookeeperEvenReplicas() { Kafka kafka = ResourceUtils.createKafka(NAMESPACE, NAME, 4, IMAGE, HEALTH_DELAY, HEALTH_TIMEOUT); KafkaSpecChecker checker = generateChecker(kafka); List<Condition> warnings = checker.run(); assertThat(warnings, hasSize(1)); Condition warning = warnings.get(0); assertThat(warning.getReason(), is("ZooKeeperReplicas")); assertThat(warning.getStatus(), is("True")); assertThat(warning.getMessage(), is("Running ZooKeeper with an odd number of replicas is recommended.")); }
@Test public void checkKafkaVersion() { Map<String, Object> kafkaOptions = new HashMap<>(); kafkaOptions.put(KafkaConfiguration.LOG_MESSAGE_FORMAT_VERSION, KafkaVersionTestUtils.PREVIOUS_FORMAT_VERSION); Kafka kafka = new KafkaBuilder(ResourceUtils.createKafka(NAMESPACE, NAME, 3, IMAGE, HEALTH_DELAY, HEALTH_TIMEOUT, Collections.emptyMap(), kafkaOptions, Collections.emptyMap(), new EphemeralStorage(), new EphemeralStorage(), null, null, null, null)) .editSpec() .editKafka() .withVersion(KafkaVersionTestUtils.LATEST_KAFKA_VERSION) .endKafka() .endSpec() .build(); KafkaSpecChecker checker = generateChecker(kafka); List<Condition> warnings = checker.run(); assertThat(warnings, hasSize(1)); Condition warning = warnings.get(0); assertThat(warning.getReason(), is("KafkaLogMessageFormatVersion")); assertThat(warning.getStatus(), is("True")); assertThat(warning.getMessage(), is("log.message.format.version does not match the Kafka cluster version, which suggests that an upgrade is incomplete.")); }
@Test public void checkMultipleWarnings() { Kafka kafka = ResourceUtils.createKafka(NAMESPACE, NAME, 1, IMAGE, HEALTH_DELAY, HEALTH_TIMEOUT, Collections.emptyMap(), Collections.emptyMap(), Collections.emptyMap(), new EphemeralStorage(), new EphemeralStorage(), null, null, null, null); KafkaSpecChecker checker = generateChecker(kafka); List<Condition> warnings = checker.run(); assertThat(warnings, hasSize(2)); }