KafkaBrokerTopicObserver implements IZkChildListener { public KafkaBrokerTopicObserver(String brokerClusterName, String zkString, long refreshTimeIntervalInMillis) { LOGGER.info("Trying to init KafkaBrokerTopicObserver {} with ZK: {}", brokerClusterName, zkString); _kakfaClusterName = brokerClusterName; _refreshTimeIntervalInMillis = refreshTimeIntervalInMillis; _zkClient = new ZkClient(zkString, 30000, 30000, ZKStringSerializer$.MODULE$); _zkClient.subscribeChildChanges(KAFKA_TOPICS_PATH, this); _zkUtils = ZkUtils.apply(_zkClient, false); executorService.scheduleAtFixedRate(new Runnable() { @Override public void run() { tryToRefreshCache(); } }, 0, refreshTimeIntervalInMillis, TimeUnit.MILLISECONDS); registerMetric(); } KafkaBrokerTopicObserver(String brokerClusterName, String zkString, long refreshTimeIntervalInMillis); void start(); @Override void handleChildChange(String parentPath, List<String> currentChilds); void tryUpdateTopic(String topic); TopicPartition getTopicPartition(String topic); TopicPartition getTopicPartitionWithRefresh(String topic); Set<String> getAllTopics(); long getNumTopics(); void stop();  }
@Test public void testKafkaBrokerTopicObserver() { Assert.assertEquals(kafkaBrokerTopicObserver.getNumTopics(), 1); Assert.assertEquals(kafkaBrokerTopicObserver.getTopicPartition("testTopic0").getPartition(), 1); Random random = new Random(); int topicCount = 3 + random.nextInt(5); for (int i = 1; i < topicCount; ++i) { String topicName = "testTopic" + i; KafkaStarterUtils.createTopic(topicName, KafkaStarterUtils.DEFAULT_ZK_STR); try { Thread.sleep(1000); } catch (Exception e) { } Assert.assertEquals(kafkaBrokerTopicObserver.getNumTopics(), 1 + i); for (int j = 0; j <= i; ++j) { Assert.assertTrue(kafkaBrokerTopicObserver.getAllTopics().contains("testTopic" + j)); Assert.assertEquals(kafkaBrokerTopicObserver.getTopicPartition("testTopic" + j).getPartition(), 1); } } }