@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 testAdjustTopicWithLargeMinISR() throws InterruptedException, ExecutionException, TimeoutException {
    KafkaCruiseControl mockKafkaCruiseControl = mockKafkaCruiseControl();
    short expectedMinISR = 2;
    AdminClient mockAdminClient = mockAdminClient(expectedMinISR);
    TopicReplicationFactorAnomalyFinder anomalyFinder = new TopicReplicationFactorAnomalyFinder(mockKafkaCruiseControl,
                                                                                                TARGET_TOPIC_REPLICATION_FACTOR,
                                                                                                TOPIC_REPLICATION_FACTOR_MARGIN,
                                                                                                mockAdminClient);
    Set<TopicAnomaly> topicAnomalies = anomalyFinder.topicAnomalies();
    assertEquals(1, topicAnomalies.size());
    TopicReplicationFactorAnomaly topicReplicationFactorAnomaly = (TopicReplicationFactorAnomaly) topicAnomalies.iterator().next();
    assertEquals(1, topicReplicationFactorAnomaly.badTopicsByDesiredRF().size());
    // We expect the desired replication factor to be 4 (i.e. TOPIC_REPLICATION_FACTOR_MARGIN + expectedMinISR)
    assertEquals(TOPIC, topicReplicationFactorAnomaly.badTopicsByDesiredRF()
                                                     .get((short) (TOPIC_REPLICATION_FACTOR_MARGIN + expectedMinISR))
                                                     .iterator().next().topicName());
    // We expect 1 out of 2 partitions of the topic to violate the target RF.
    assertEquals(0.5, topicReplicationFactorAnomaly.badTopicsByDesiredRF()
                                                   .get((short) (TOPIC_REPLICATION_FACTOR_MARGIN + expectedMinISR))
                                                   .iterator().next().violationRatio(), DELTA);
    EasyMock.verify(mockKafkaCruiseControl, mockAdminClient);
  }