@Test public void testKafkaBrokerTopicObserver() { Assert.assertEquals(kafkaBrokerTopicObserver.getNumTopics(), 1); Assert.assertEquals(kafkaBrokerTopicObserver.getTopicPartition("testTopic0").getPartition(), 1); for (int i = 1; i < 10; ++i) { String topicName = "testTopic" + i; KafkaStarterUtils.createTopic(topicName, KafkaStarterUtils.DEFAULT_ZK_STR); try { Thread.sleep(5000); } 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); } } }
public KafkaBrokerTopicObserver(String brokerClusterName, String zkString) { LOGGER.info("Trying to init KafkaBrokerTopicObserver {} with ZK: {}", brokerClusterName, zkString); _kakfaClusterName = brokerClusterName; _zkUtils = ZkUtils.apply(zkString, 30000, 30000, false); _zkClient = ZkUtils.createZkClient(zkString, 30000, 30000); _zkClient.subscribeChildChanges(KAFKA_TOPICS_PATH, this); registerMetric(); executorService.scheduleAtFixedRate(new Runnable() { @Override public void run() { tryToRefreshCache(); } }, 0, 600, TimeUnit.SECONDS); }
KafkaBrokerTopicObserver implements IZkChildListener { public KafkaBrokerTopicObserver(String brokerClusterName, String zkString) { LOGGER.info("Trying to init KafkaBrokerTopicObserver {} with ZK: {}", brokerClusterName, zkString); _kakfaClusterName = brokerClusterName; _zkUtils = ZkUtils.apply(zkString, 30000, 30000, false); _zkClient = ZkUtils.createZkClient(zkString, 30000, 30000); _zkClient.subscribeChildChanges(KAFKA_TOPICS_PATH, this); registerMetric(); executorService.scheduleAtFixedRate(new Runnable() { @Override public void run() { tryToRefreshCache(); } }, 0, 600, TimeUnit.SECONDS); } }
KafkaBrokerTopicObserver implements IZkChildListener { public KafkaBrokerTopicObserver(String brokerClusterName, String zkString) { LOGGER.info("Trying to init KafkaBrokerTopicObserver {} with ZK: {}", brokerClusterName, zkString); _kakfaClusterName = brokerClusterName; _zkUtils = ZkUtils.apply(zkString, 30000, 30000, false); _zkClient = ZkUtils.createZkClient(zkString, 30000, 30000); _zkClient.subscribeChildChanges(KAFKA_TOPICS_PATH, this); registerMetric(); executorService.scheduleAtFixedRate(new Runnable() { @Override public void run() { tryToRefreshCache(); } }, 0, 600, TimeUnit.SECONDS); } KafkaBrokerTopicObserver(String brokerClusterName, String zkString); }
KafkaBrokerTopicObserver implements IZkChildListener { public KafkaBrokerTopicObserver(String brokerClusterName, String zkString) { LOGGER.info("Trying to init KafkaBrokerTopicObserver {} with ZK: {}", brokerClusterName, zkString); _kakfaClusterName = brokerClusterName; _zkUtils = ZkUtils.apply(zkString, 30000, 30000, false); _zkClient = ZkUtils.createZkClient(zkString, 30000, 30000); _zkClient.subscribeChildChanges(KAFKA_TOPICS_PATH, this); registerMetric(); executorService.scheduleAtFixedRate(new Runnable() { @Override public void run() { tryToRefreshCache(); } }, 0, 600, TimeUnit.SECONDS); } KafkaBrokerTopicObserver(String brokerClusterName, String zkString); @Override void handleChildChange(String parentPath, List<String> currentChilds); TopicPartition getTopicPartition(String topic); Set<String> getAllTopics(); long getNumTopics(); void stop(); }
KafkaBrokerTopicObserver implements IZkChildListener { public KafkaBrokerTopicObserver(String brokerClusterName, String zkString) { LOGGER.info("Trying to init KafkaBrokerTopicObserver {} with ZK: {}", brokerClusterName, zkString); _kakfaClusterName = brokerClusterName; _zkUtils = ZkUtils.apply(zkString, 30000, 30000, false); _zkClient = ZkUtils.createZkClient(zkString, 30000, 30000); _zkClient.subscribeChildChanges(KAFKA_TOPICS_PATH, this); registerMetric(); executorService.scheduleAtFixedRate(new Runnable() { @Override public void run() { tryToRefreshCache(); } }, 0, 600, TimeUnit.SECONDS); } KafkaBrokerTopicObserver(String brokerClusterName, String zkString); @Override void handleChildChange(String parentPath, List<String> currentChilds); TopicPartition getTopicPartition(String topic); Set<String> getAllTopics(); long getNumTopics(); void stop();  }