ZookeeperBrokerStorage implements BrokerStorage { @Override @SuppressWarnings("unchecked") public BrokerDetails readBrokerDetails(Integer brokerId) { try { Broker broker = kafkaZkClient.getBroker(brokerId).get(); Option<Node> node = broker.getNode(brokerListenerName); String host = node.get().host(); int port = node.get().port(); return new BrokerDetails(host, port); } catch (Exception exception) { throw new BrokerInfoNotAvailableException(brokerId, exception); } } @Inject  ZookeeperBrokerStorage(@Named(CuratorType.KAFKA) CuratorFramework curatorFramework,
                                  KafkaZkClient kafkaZkClient, String brokerListenerName); @Override @SuppressWarnings("unchecked") int readLeaderForPartition(TopicAndPartition topicAndPartition); @Override Multimap<Integer, TopicAndPartition> readLeadersForPartitions(Set<TopicAndPartition> topicAndPartitionSet); @Override @SuppressWarnings("unchecked") BrokerDetails readBrokerDetails(Integer brokerId); @Override List<Integer> readPartitionsIds(String topicName);  }
@Test public void shouldReadBrokerDetails() throws Exception { createBrokerDetails(1, "localhost", 9092); BrokerDetails details = brokerStorage.readBrokerDetails(1); assertThat(details.getHost()).isEqualTo("localhost"); assertThat(details.getPort()).isEqualTo(9092); }
@Test public void shouldGetErrorWhileReadingBrokerDetails() { catchException(brokerStorage).readBrokerDetails(5); assertThat((Exception) caughtException()).isInstanceOf(BrokerInfoNotAvailableException.class); }