ZookeeperBrokerStorage implements BrokerStorage { @Override public List<Integer> readPartitionsIds(String topicName) { try { List<String> partitionsAsStrings = curatorFramework.getChildren().forPath(String.format(PARTITIONS, topicName)); List<Integer> partitions = transform(partitionsAsStrings, Ints.stringConverter()); return Ordering.natural().sortedCopy(partitions); } catch (Exception exception) { throw new PartitionsNotFoundForGivenTopicException(topicName, 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 shouldReadPartitionsIds() throws Exception { String topicName = "readPartitionsIdsTestTopic"; List<Integer> partitions = Lists.newArrayList(0, 1, 2); createPartitionsForTopic(topicName, partitions); List<Integer> partitionsFromBroker = brokerStorage.readPartitionsIds(topicName); assertThat(partitionsFromBroker).isEqualTo(partitions); }
@Test public void shouldThrowExceptionWhenReadPartitionsForNotExistingTopic() throws Exception { String nonExistingTopic = "readPartitionIdsNonExistingTopic"; catchException(brokerStorage).readPartitionsIds(nonExistingTopic); assertThat((Exception) caughtException()) .isInstanceOf(PartitionsNotFoundForGivenTopicException.class) .hasMessageContaining(nonExistingTopic); }