@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;
    // Create Kafka topic
    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);
    }
  }
}