@Override
  public Set<TopicAnomaly> topicAnomalies() {
    LOG.info("Start to detect topic replication factor anomaly.");
    Cluster cluster = _kafkaCruiseControl.kafkaCluster();
    Set<String> topicsToCheck;
    if (_topicExcludedFromCheck.pattern().isEmpty()) {
      topicsToCheck = new HashSet<>(cluster.topics());
    } else {
      topicsToCheck = new HashSet<>();
      cluster.topics().stream().filter(topic -> !_topicExcludedFromCheck.matcher(topic).matches()).forEach(topicsToCheck::add);
    }
    refreshTopicMinISRCache();
    if (!topicsToCheck.isEmpty()) {
      maybeRetrieveAndCacheTopicMinISR(topicsToCheck);
      Map<Short, Set<TopicReplicationFactorAnomalyEntry>> badTopicsByDesiredRF = populateBadTopicsByDesiredRF(topicsToCheck, cluster);
      if (!badTopicsByDesiredRF.isEmpty()) {
        return Collections.singleton(createTopicReplicationFactorAnomaly(badTopicsByDesiredRF, _targetReplicationFactor));
      }
    }
    return Collections.emptySet();
  }
@Test
  public void testAnomalyDetection() throws InterruptedException, ExecutionException, TimeoutException {
    KafkaCruiseControl mockKafkaCruiseControl = mockKafkaCruiseControl();
    AdminClient mockAdminClient = mockAdminClient((short) 1);
    TopicReplicationFactorAnomalyFinder anomalyFinder = new TopicReplicationFactorAnomalyFinder(mockKafkaCruiseControl,
                                                                                                TARGET_TOPIC_REPLICATION_FACTOR,
                                                                                                TOPIC_REPLICATION_FACTOR_MARGIN,
                                                                                                mockAdminClient);
    Set<TopicAnomaly> topicAnomalies = anomalyFinder.topicAnomalies();
    assertEquals(1, topicAnomalies.size());
    EasyMock.verify(mockKafkaCruiseControl, mockAdminClient);
  }