@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(); }))); } }
Future<Boolean> canRoll(int podId) { log.debug("Determining whether broker {} can be rolled", podId); return canRollBroker(descriptions, podId); }
KafkaAvailability { Future<Boolean> canRoll(int podId) { log.debug("Determining whether broker {} can be rolled", podId); return canRollBroker(descriptions, podId); } }
KafkaAvailability { Future<Boolean> canRoll(int podId) { log.debug("Determining whether broker {} can be rolled", podId); return canRollBroker(descriptions, podId); } KafkaAvailability(Admin ac); }
KafkaAvailability { Future<Boolean> canRoll(int podId) { log.debug("Determining whether broker {} can be rolled", podId); return canRollBroker(descriptions, podId); } KafkaAvailability(Admin ac);  }
KafkaAvailability { Future<Boolean> canRoll(int podId) { log.debug("Determining whether broker {} can be rolled", podId); return canRollBroker(descriptions, podId); } KafkaAvailability(Admin ac);   }