@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)); }
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!"); } }
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!"); } } }
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!"); } }  }
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); }
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);  }