BrokerNodeFunction { public BrokerNode apply(Predicate<String> hostNamePredicate) { Collection<Node> nodes = KafkaFutures.join(client.describeCluster().nodes()); try { Node node = find(nodes, n -> hostNamePredicate.test(n.host())); log.debug("Using broker {}", node); return new BrokerNode(node.id(), ofNullable(node.rack()).orElse("none"), node.host()); } catch (NoSuchElementException e) { throw new RuntimeException("No broker found on localhost!"); } }  BrokerNode apply(Predicate<String> hostNamePredicate);  }
@Test public void success() throws Exception { BrokerNode brokerNode = new BrokerNode(1, "rack", "host"); Node node = new Node(1, "host", 9092, "rack"); KafkaFuture<Collection<Node>> kafkaFuture = completedFuture(singleton(node)); doReturn(describeClusterResult).when(adminClient).describeCluster(); doReturn(kafkaFuture).when(describeClusterResult).nodes(); BrokerNode result = underTest.apply(x -> true); assertThat(result, is(brokerNode)); }
@Test public void noRack() throws Exception { BrokerNode brokerNode = new BrokerNode(1, "none", "host"); Node node = new Node(1, "host", 9092, null); KafkaFuture<Collection<Node>> kafkaFuture = completedFuture(singleton(node)); doReturn(describeClusterResult).when(adminClient).describeCluster(); doReturn(kafkaFuture).when(describeClusterResult).nodes(); BrokerNode result = underTest.apply(x -> true); assertThat(result, is(brokerNode)); }
@Test(expected = RuntimeException.class) public void notFound() throws Exception { Node node = new Node(1, "host", 9092, null); KafkaFuture<Collection<Node>> kafkaFuture = completedFuture(singleton(node)); doReturn(describeClusterResult).when(adminClient).describeCluster(); doReturn(kafkaFuture).when(describeClusterResult).nodes(); underTest.apply(x -> false); }