SourceKafkaClusterValidationManager { public String validateSourceKafkaCluster() { Set<String> notExistedTopics = new HashSet<String>(); Map<String, Integer> misMatchedPartitionNumberTopics = new HashMap<String, Integer>(); int numMismatchedTopicPartitions = 0; for (String topic : _helixMirrorMakerManager.getTopicLists()) { TopicPartition tp = _sourceKafkaTopicObserver.getTopicPartition(topic); if (tp == null) { LOGGER.warn("Topic {} is not in source kafka broker!", topic); notExistedTopics.add(topic); } else { int numPartitionsInMirrorMaker = _helixMirrorMakerManager.getIdealStateForTopic(topic).getNumPartitions(); if (numPartitionsInMirrorMaker != tp.getPartition()) { int mismatchedPartitions = Math.abs(numPartitionsInMirrorMaker - tp.getPartition()); if (_enableAutoTopicExpansion && (tp.getPartition() > numPartitionsInMirrorMaker)) { LOGGER.warn( "Trying to expand topic {} from {} partitions in mirror maker to {} from source kafka broker!", topic, numPartitionsInMirrorMaker, tp.getPartition()); _numAutoExpandedTopics.inc(); _numAutoExpandedTopicPartitions.inc(mismatchedPartitions); _helixMirrorMakerManager.expandTopicInMirrorMaker(tp); } else { numMismatchedTopicPartitions += mismatchedPartitions; misMatchedPartitionNumberTopics.put(topic, mismatchedPartitions); LOGGER.warn( "Number of partitions not matched for topic {} between mirrormaker:{} and source kafka broker: {}!", topic, numPartitionsInMirrorMaker, tp.getPartition()); } } } } JSONObject mismatchedTopicPartitionsJson = constructMismatchedTopicPartitionsJson(misMatchedPartitionNumberTopics); JSONObject validationResultJson = constructValidationResultJson(notExistedTopics.size(), misMatchedPartitionNumberTopics.size(), numMismatchedTopicPartitions, mismatchedTopicPartitionsJson); if (_helixMirrorMakerManager.isLeader()) { updateMetrics(notExistedTopics.size(), misMatchedPartitionNumberTopics.size(), numMismatchedTopicPartitions, misMatchedPartitionNumberTopics); } return validationResultJson.toJSONString(); } SourceKafkaClusterValidationManager(KafkaBrokerTopicObserver sourceKafkaTopicObserver,
      HelixMirrorMakerManager helixMirrorMakerManager);  SourceKafkaClusterValidationManager(KafkaBrokerTopicObserver sourceKafkaTopicObserver,
      HelixMirrorMakerManager helixMirrorMakerManager, boolean enableAutoTopicExpansion); void start(); void stop(); String validateSourceKafkaCluster();  }
@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; 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; 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; 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}"); }