KafkaAvailability { Future<Boolean> canRoll(int podId) { log.debug("Determining whether broker {} can be rolled", podId); return canRollBroker(descriptions, podId); } KafkaAvailability(Admin ac);   }
@Test public void testBelowMinIsr(VertxTestContext context) { KSB ksb = new KSB() .addNewTopic("A", false) .addToConfig(TopicConfig.MIN_IN_SYNC_REPLICAS_CONFIG, "2") .addNewPartition(0) .replicaOn(0, 1, 3) .leader(0) .isr(0, 1) .endPartition() .endTopic() .addNewTopic("B", false) .addToConfig(TopicConfig.MIN_IN_SYNC_REPLICAS_CONFIG, "2") .addNewPartition(0) .replicaOn(0, 1, 3) .leader(1) .isr(1) .endPartition() .endTopic() .addBroker(4); KafkaAvailability kafkaAvailability = new KafkaAvailability(ksb.ac()); Checkpoint a = context.checkpoint(ksb.brokers.size()); for (Integer brokerId : ksb.brokers.keySet()) { kafkaAvailability.canRoll(brokerId).onComplete(context.succeeding(canRoll -> context.verify(() -> { if (brokerId == 4) { assertTrue(canRoll, "broker " + brokerId + " should be rollable, having no partitions"); } else { assertFalse(canRoll, "broker " + brokerId + " should not be rollable, being minisr = 2 and it's only replicated on two brokers"); } a.flag(); }))); } }
@Test public void testAtMinIsr(VertxTestContext context) { KSB ksb = new KSB() .addNewTopic("A", false) .addToConfig(TopicConfig.MIN_IN_SYNC_REPLICAS_CONFIG, "2") .addNewPartition(0) .replicaOn(0, 1) .leader(0) .isr(0, 1) .endPartition() .endTopic() .addNewTopic("B", false) .addToConfig(TopicConfig.MIN_IN_SYNC_REPLICAS_CONFIG, "2") .addNewPartition(0) .replicaOn(0, 1) .leader(1) .isr(0, 1) .endPartition() .endTopic() .addBroker(2); KafkaAvailability kafkaAvailability = new KafkaAvailability(ksb.ac()); Checkpoint a = context.checkpoint(ksb.brokers.size()); for (Integer brokerId : ksb.brokers.keySet()) { kafkaAvailability.canRoll(brokerId).onComplete(context.succeeding(canRoll -> context.verify(() -> { if (brokerId == 2) { assertTrue(canRoll, "broker " + brokerId + " should be rollable, having no partitions"); } else { assertTrue(canRoll, "broker " + brokerId + " should be rollable, because although rolling it will impact availability minisr=|replicas|"); } a.flag(); }))); } }
@Test public void testAboveMinIsr(VertxTestContext context) { KSB ksb = new KSB() .addNewTopic("A", false) .addToConfig(TopicConfig.MIN_IN_SYNC_REPLICAS_CONFIG, "1") .addNewPartition(0) .replicaOn(0, 1, 2) .leader(0) .isr(0, 1, 2) .endPartition() .endTopic() .addNewTopic("B", false) .addToConfig(TopicConfig.MIN_IN_SYNC_REPLICAS_CONFIG, "1") .addNewPartition(0) .replicaOn(0, 1, 2) .leader(1) .isr(0, 1, 2) .endPartition() .endTopic() .addBroker(3); KafkaAvailability kafkaAvailability = new KafkaAvailability(ksb.ac()); Checkpoint a = context.checkpoint(ksb.brokers.size()); for (Integer brokerId : ksb.brokers.keySet()) { kafkaAvailability.canRoll(brokerId).onComplete(context.succeeding(canRoll -> context.verify(() -> { assertTrue(canRoll, "broker " + brokerId + " should be rollable, being minisr = 1 and having two brokers in its isr"); a.flag(); }))); } }
@Test public void testMinIsrEqualsReplicas(VertxTestContext context) { KSB ksb = new KSB() .addNewTopic("A", false) .addToConfig(TopicConfig.MIN_IN_SYNC_REPLICAS_CONFIG, "3") .addNewPartition(0) .replicaOn(0, 1, 2) .leader(0) .isr(0, 1, 2) .endPartition() .endTopic() .addBroker(3); KafkaAvailability kafkaAvailability = new KafkaAvailability(ksb.ac()); Checkpoint a = context.checkpoint(ksb.brokers.size()); for (Integer brokerId : ksb.brokers.keySet()) { kafkaAvailability.canRoll(brokerId).onComplete(context.succeeding(canRoll -> context.verify(() -> { assertTrue(canRoll, "broker " + brokerId + " should be rollable, being minisr = 3, but only 3 replicas"); a.flag(); }))); } }
@Test public void testMinIsrEqualsReplicasWithOfflineReplicas(VertxTestContext context) { KSB ksb = new KSB() .addNewTopic("A", false) .addToConfig(TopicConfig.MIN_IN_SYNC_REPLICAS_CONFIG, "3") .addNewPartition(0) .replicaOn(0, 1, 2) .leader(0) .isr(0, 1) .endPartition() .endTopic() .addBroker(3); KafkaAvailability kafkaAvailability = new KafkaAvailability(ksb.ac()); Checkpoint a = context.checkpoint(ksb.brokers.size()); for (Integer brokerId : ksb.brokers.keySet()) { kafkaAvailability.canRoll(brokerId).onComplete(context.succeeding(canRoll -> context.verify(() -> { assertTrue(canRoll, "broker " + brokerId + " should be rollable, being minisr = 3, but only 3 replicas"); a.flag(); }))); } }
@Test public void testMinIsrMoreThanReplicas(VertxTestContext context) { KSB ksb = new KSB() .addNewTopic("A", false) .addToConfig(TopicConfig.MIN_IN_SYNC_REPLICAS_CONFIG, "2") .addNewPartition(0) .replicaOn(0) .leader(0) .isr(0) .endPartition() .endTopic() .addBroker(3); KafkaAvailability kafkaAvailability = new KafkaAvailability(ksb.ac()); Checkpoint a = context.checkpoint(ksb.brokers.size()); for (Integer brokerId : ksb.brokers.keySet()) { kafkaAvailability.canRoll(brokerId).onComplete(context.succeeding(canRoll -> context.verify(() -> { assertTrue(canRoll, "broker " + brokerId + " should be rollable, being minisr = 2, but only 1 replicas"); a.flag(); }))); } }
@Test public void testNoLeader(VertxTestContext context) { KSB ksb = new KSB() .addNewTopic("A", false) .addToConfig(TopicConfig.MIN_IN_SYNC_REPLICAS_CONFIG, "1") .addNewPartition(0) .replicaOn(0, 1, 2) .isr(1, 2) .endPartition() .endTopic() .addNewTopic("B", false) .addToConfig(TopicConfig.MIN_IN_SYNC_REPLICAS_CONFIG, "1") .addNewPartition(0) .replicaOn(0, 1, 2) .isr(0) .endPartition() .endTopic() .addBroker(3); KafkaAvailability kafkaSorted = new KafkaAvailability(ksb.ac()); Checkpoint a = context.checkpoint(ksb.brokers.size()); for (Integer brokerId : ksb.brokers.keySet()) { kafkaSorted.canRoll(brokerId).onComplete(context.succeeding(canRoll -> context.verify(() -> { if (brokerId == 0) { assertFalse(canRoll, "broker " + brokerId + " should not be rollable, because B/0 would be below min isr"); } else { assertTrue(canRoll, "broker " + brokerId + " should be rollable, being minisr = 1 and having two brokers in its isr"); } a.flag(); }))); } }
@Test public void testNoMinIsr(VertxTestContext context) { KSB ksb = new KSB() .addNewTopic("A", false) .addNewPartition(0) .replicaOn(0, 1, 2) .leader(0) .isr(0, 1, 2) .endPartition() .endTopic() .addNewTopic("B", false) .addNewPartition(0) .replicaOn(0, 1, 2) .leader(1) .isr(1, 0, 2) .endPartition() .endTopic() .addBroker(3); KafkaAvailability kafkaAvailability = new KafkaAvailability(ksb.ac()); Checkpoint a = context.checkpoint(ksb.brokers.size()); for (Integer brokerId : ksb.brokers.keySet()) { kafkaAvailability.canRoll(brokerId).onComplete(context.succeeding(canRoll -> context.verify(() -> { assertTrue(canRoll, "broker " + brokerId + " should be rollable, being minisr = 1 and having two brokers in its isr"); a.flag(); }))); } }
@Test public void testCanRollThrowsTimeoutExceptionWhenTopicsListThrowsException(VertxTestContext context) { KSB ksb = new KSB() .addNewTopic("A", false) .addToConfig(TopicConfig.MIN_IN_SYNC_REPLICAS_CONFIG, "1") .addNewPartition(0) .replicaOn(0, 1, 2) .leader(0) .isr(0, 1, 2) .endPartition() .endTopic() .addNewTopic("B", false) .addToConfig(TopicConfig.MIN_IN_SYNC_REPLICAS_CONFIG, "1") .addNewPartition(0) .replicaOn(0, 1, 2) .leader(1) .isr(0, 1, 2) .endPartition() .endTopic() .addBroker(3) .listTopicsResult(new TimeoutException()); KafkaAvailability kafkaAvailability = new KafkaAvailability(ksb.ac()); Checkpoint a = context.checkpoint(ksb.brokers.size()); for (Integer brokerId : ksb.brokers.keySet()) { kafkaAvailability.canRoll(brokerId).onComplete(context.failing(e -> context.verify(() -> { assertThat(e, instanceOf(TimeoutException.class)); a.flag(); }))); } }
@Test public void testCanRollThrowsExceptionWhenTopicDescribeThrows(VertxTestContext context) { KSB ksb = new KSB() .addNewTopic("A", false) .addToConfig(TopicConfig.MIN_IN_SYNC_REPLICAS_CONFIG, "1") .addNewPartition(0) .replicaOn(0, 1, 2) .leader(0) .isr(0, 1, 2) .endPartition() .endTopic() .addNewTopic("B", false) .addToConfig(TopicConfig.MIN_IN_SYNC_REPLICAS_CONFIG, "1") .addNewPartition(0) .replicaOn(0, 1, 2) .leader(1) .isr(0, 1, 2) .endPartition() .endTopic() .addBroker(3) .describeTopicsResult("A", new UnknownTopicOrPartitionException()); KafkaAvailability kafkaAvailability = new KafkaAvailability(ksb.ac()); Checkpoint a = context.checkpoint(ksb.brokers.size()); for (Integer brokerId : ksb.brokers.keySet()) { kafkaAvailability.canRoll(brokerId).onComplete(context.failing(e -> context.verify(() -> { assertThat(e, instanceOf(UnknownTopicOrPartitionException.class)); a.flag(); }))); } }
@Test public void testCanRollThrowsExceptionWhenDescribeConfigsThrows(VertxTestContext context) { KSB ksb = new KSB() .addNewTopic("A", false) .addToConfig(TopicConfig.MIN_IN_SYNC_REPLICAS_CONFIG, "1") .addNewPartition(0) .replicaOn(0, 1, 2) .leader(0) .isr(0, 1, 2) .endPartition() .endTopic() .addNewTopic("B", false) .addToConfig(TopicConfig.MIN_IN_SYNC_REPLICAS_CONFIG, "1") .addNewPartition(0) .replicaOn(0, 1, 2) .leader(1) .isr(0, 1, 2) .endPartition() .endTopic() .addBroker(3) .describeConfigsResult(new ConfigResource(ConfigResource.Type.TOPIC, "A"), new UnknownTopicOrPartitionException()); KafkaAvailability kafkaAvailability = new KafkaAvailability(ksb.ac()); Checkpoint a = context.checkpoint(ksb.brokers.size()); for (Integer brokerId : ksb.brokers.keySet()) { if (brokerId <= 2) { kafkaAvailability.canRoll(brokerId).onComplete(context.failing(e -> context.verify(() -> { assertThat(e, instanceOf(UnknownTopicOrPartitionException.class)); a.flag(); }))); } else { kafkaAvailability.canRoll(brokerId).onComplete(context.succeeding(canRoll -> a.flag())); } } }