@Test
public void testValidation() {

  String validationResult = sourceKafkaClusterValidationManager.validateSourceKafkaCluster();
  System.out.println(validationResult);
  Assert.assertEquals(validationResult,
      "{\"numMissingTopics\":0,\"numMismatchedTopics\":0,\"mismatchedTopicPartitions\":{},\"numMismatchedTopicPartitions\":0}");
  for (int i = 0; i < 2; ++i) {
    Assert.assertEquals(helixMirrorMakerManager.getTopicLists().size(), 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);
    }
    validationResult = sourceKafkaClusterValidationManager.validateSourceKafkaCluster();
    System.out.println(validationResult);
    Assert.assertEquals(validationResult,
      "{\"numMissingTopics\":0,\"numMismatchedTopics\":0,\"mismatchedTopicPartitions\":{},\"numMismatchedTopicPartitions\":0}");
    helixMirrorMakerManager.addTopicToMirrorMaker(topicName, 1);
  }
  for (int i = 2; i < 4; ++i) {
    helixMirrorMakerManager.addTopicToMirrorMaker("testTopic" + i, 1);
  }
  JSONObject validationResultJson =
      JSONObject.parseObject(sourceKafkaClusterValidationManager.validateSourceKafkaCluster());
  System.out.println(validationResultJson);
  Assert.assertEquals(validationResultJson.get("numMissingTopics"), 2);
  for (int i = 2; i < 4; ++i) {
    Assert.assertEquals(helixMirrorMakerManager.getTopicLists().size(), 4);
    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);
    }
    validationResultJson =
        JSONObject.parseObject(sourceKafkaClusterValidationManager.validateSourceKafkaCluster());
    System.out.println(validationResultJson);
    Assert.assertEquals(validationResultJson.get("numMissingTopics"), 3 - i);
  }
  for (int i = 4; i < 6; ++i) {
    helixMirrorMakerManager.addTopicToMirrorMaker("testTopic" + i, 2);
  }
  validationResultJson =
      JSONObject.parseObject(sourceKafkaClusterValidationManager.validateSourceKafkaCluster());
  System.out.println(validationResultJson);
  Assert.assertEquals(validationResultJson.get("numMissingTopics"), 2);
  for (int i = 4; i < 6; ++i) {
    Assert.assertEquals(helixMirrorMakerManager.getTopicLists().size(), 6);
    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);
    }
    validationResultJson =
        JSONObject.parseObject(sourceKafkaClusterValidationManager.validateSourceKafkaCluster());
    System.out.println(validationResultJson);
    Assert.assertEquals(validationResultJson.get("numMissingTopics"), 5 - i);
    Assert.assertEquals(validationResultJson.get("numMismatchedTopics"), i - 3);
    Assert.assertEquals(validationResultJson.get("numMismatchedTopicPartitions"), i - 3);
    for (int h = 4; h <= i; ++h) {
      Assert.assertEquals(
          validationResultJson.getJSONObject("mismatchedTopicPartitions").get("testTopic" + h),
          1);
    }
  }
  for (int i = 4; i < 6; ++i) {
    helixMirrorMakerManager.deleteTopicInMirrorMaker("testTopic" + i);
    helixMirrorMakerManager.addTopicToMirrorMaker("testTopic" + i, 1);
  }
  validationResult = sourceKafkaClusterValidationManager.validateSourceKafkaCluster();
  System.out.println(validationResult);
  Assert.assertEquals(validationResult,
      "{\"numMissingTopics\":0,\"numMismatchedTopics\":0,\"mismatchedTopicPartitions\":{},\"numMismatchedTopicPartitions\":0}");
 
}