@Test public void testSkipTopicWithLargeMinISR() throws InterruptedException, ExecutionException, TimeoutException { KafkaCruiseControl mockKafkaCruiseControl = mockKafkaCruiseControl(); AdminClient mockAdminClient = mockAdminClient((short) 2); TopicReplicationFactorAnomalyFinder anomalyFinder = new TopicReplicationFactorAnomalyFinder(mockKafkaCruiseControl, TARGET_TOPIC_REPLICATION_FACTOR, mockAdminClient); Set<TopicAnomaly> topicAnomalies = anomalyFinder.topicAnomalies(); assertTrue(topicAnomalies.isEmpty()); EasyMock.verify(mockKafkaCruiseControl, mockAdminClient); }
@Override public Set<TopicAnomaly> topicAnomalies() { LOG.info("Start to detect topic replication factor anomaly."); Cluster cluster = _kafkaCruiseControl.kafkaCluster(); Set<String> topicsToCheck = new HashSet<>(); for (String topic : cluster.topics()) { if (_topicExcludedFromCheck.matcher(topic).matches()) { continue; } for (PartitionInfo partition : cluster.partitionsForTopic(topic)) { if (partition.replicas().length != _targetReplicationFactor) { topicsToCheck.add(topic); break; } } } refreshTopicMinISRCache(); if (!topicsToCheck.isEmpty()) { maybeRetrieveAndCacheTopicMinISR(topicsToCheck); Map<Short, Set<TopicReplicationFactorAnomalyEntry>> badTopicsByReplicationFactor = populateBadTopicsByReplicationFactor(topicsToCheck, cluster); if (!badTopicsByReplicationFactor.isEmpty()) { return Collections.singleton(createTopicReplicationFactorAnomaly(badTopicsByReplicationFactor, _targetReplicationFactor)); } } return Collections.emptySet(); }
TopicReplicationFactorAnomalyFinder implements TopicAnomalyFinder { @Override public Set<TopicAnomaly> topicAnomalies() { LOG.info("Start to detect topic replication factor anomaly."); Cluster cluster = _kafkaCruiseControl.kafkaCluster(); Set<String> topicsToCheck = new HashSet<>(); for (String topic : cluster.topics()) { if (_topicExcludedFromCheck.matcher(topic).matches()) { continue; } for (PartitionInfo partition : cluster.partitionsForTopic(topic)) { if (partition.replicas().length != _targetReplicationFactor) { topicsToCheck.add(topic); break; } } } refreshTopicMinISRCache(); if (!topicsToCheck.isEmpty()) { maybeRetrieveAndCacheTopicMinISR(topicsToCheck); Map<Short, Set<TopicReplicationFactorAnomalyEntry>> badTopicsByReplicationFactor = populateBadTopicsByReplicationFactor(topicsToCheck, cluster); if (!badTopicsByReplicationFactor.isEmpty()) { return Collections.singleton(createTopicReplicationFactorAnomaly(badTopicsByReplicationFactor, _targetReplicationFactor)); } } return Collections.emptySet(); } }
TopicReplicationFactorAnomalyFinder implements TopicAnomalyFinder { @Override public Set<TopicAnomaly> topicAnomalies() { LOG.info("Start to detect topic replication factor anomaly."); Cluster cluster = _kafkaCruiseControl.kafkaCluster(); Set<String> topicsToCheck = new HashSet<>(); for (String topic : cluster.topics()) { if (_topicExcludedFromCheck.matcher(topic).matches()) { continue; } for (PartitionInfo partition : cluster.partitionsForTopic(topic)) { if (partition.replicas().length != _targetReplicationFactor) { topicsToCheck.add(topic); break; } } } refreshTopicMinISRCache(); if (!topicsToCheck.isEmpty()) { maybeRetrieveAndCacheTopicMinISR(topicsToCheck); Map<Short, Set<TopicReplicationFactorAnomalyEntry>> badTopicsByReplicationFactor = populateBadTopicsByReplicationFactor(topicsToCheck, cluster); if (!badTopicsByReplicationFactor.isEmpty()) { return Collections.singleton(createTopicReplicationFactorAnomaly(badTopicsByReplicationFactor, _targetReplicationFactor)); } } return Collections.emptySet(); } TopicReplicationFactorAnomalyFinder();   TopicReplicationFactorAnomalyFinder(KafkaCruiseControl kafkaCruiseControl,
                                      short targetReplicationFactor,
                                      AdminClient adminClient); }
TopicReplicationFactorAnomalyFinder implements TopicAnomalyFinder { @Override public Set<TopicAnomaly> topicAnomalies() { LOG.info("Start to detect topic replication factor anomaly."); Cluster cluster = _kafkaCruiseControl.kafkaCluster(); Set<String> topicsToCheck = new HashSet<>(); for (String topic : cluster.topics()) { if (_topicExcludedFromCheck.matcher(topic).matches()) { continue; } for (PartitionInfo partition : cluster.partitionsForTopic(topic)) { if (partition.replicas().length != _targetReplicationFactor) { topicsToCheck.add(topic); break; } } } refreshTopicMinISRCache(); if (!topicsToCheck.isEmpty()) { maybeRetrieveAndCacheTopicMinISR(topicsToCheck); Map<Short, Set<TopicReplicationFactorAnomalyEntry>> badTopicsByReplicationFactor = populateBadTopicsByReplicationFactor(topicsToCheck, cluster); if (!badTopicsByReplicationFactor.isEmpty()) { return Collections.singleton(createTopicReplicationFactorAnomaly(badTopicsByReplicationFactor, _targetReplicationFactor)); } } return Collections.emptySet(); } TopicReplicationFactorAnomalyFinder();   TopicReplicationFactorAnomalyFinder(KafkaCruiseControl kafkaCruiseControl,
                                      short targetReplicationFactor,
                                      AdminClient adminClient); @Override Set<TopicAnomaly> topicAnomalies(); @Override void configure(Map<String, ?> configs); }
TopicReplicationFactorAnomalyFinder implements TopicAnomalyFinder { @Override public Set<TopicAnomaly> topicAnomalies() { LOG.info("Start to detect topic replication factor anomaly."); Cluster cluster = _kafkaCruiseControl.kafkaCluster(); Set<String> topicsToCheck = new HashSet<>(); for (String topic : cluster.topics()) { if (_topicExcludedFromCheck.matcher(topic).matches()) { continue; } for (PartitionInfo partition : cluster.partitionsForTopic(topic)) { if (partition.replicas().length != _targetReplicationFactor) { topicsToCheck.add(topic); break; } } } refreshTopicMinISRCache(); if (!topicsToCheck.isEmpty()) { maybeRetrieveAndCacheTopicMinISR(topicsToCheck); Map<Short, Set<TopicReplicationFactorAnomalyEntry>> badTopicsByReplicationFactor = populateBadTopicsByReplicationFactor(topicsToCheck, cluster); if (!badTopicsByReplicationFactor.isEmpty()) { return Collections.singleton(createTopicReplicationFactorAnomaly(badTopicsByReplicationFactor, _targetReplicationFactor)); } } return Collections.emptySet(); } TopicReplicationFactorAnomalyFinder();   TopicReplicationFactorAnomalyFinder(KafkaCruiseControl kafkaCruiseControl,
                                      short targetReplicationFactor,
                                      AdminClient adminClient); @Override Set<TopicAnomaly> topicAnomalies(); @Override void configure(Map<String, ?> configs); static final String SELF_HEALING_TARGET_TOPIC_REPLICATION_FACTOR_CONFIG; static final String TOPIC_EXCLUDED_FROM_REPLICATION_FACTOR_CHECK; static final String DEFAULT_TOPIC_EXCLUDED_FROM_REPLICATION_FACTOR_CHECK; static final String TOPIC_REPLICATION_FACTOR_ANOMALY_CLASS_CONFIG; static final Class<?> DEFAULT_TOPIC_REPLICATION_FACTOR_ANOMALY_CLASS; static final String BAD_TOPICS_BY_REPLICATION_FACTOR_CONFIG; static final String TOPIC_REPLICATION_FACTOR_MARGIN_CONFIG; static final short DEFAULT_TOPIC_REPLICATION_FACTOR_MARGIN; static final String TOPIC_MIN_ISR_RECORD_RETENTION_TIME_MS_CONFIG; static final long DEFAULT_TOPIC_MIN_ISR_RECORD_RETENTION_TIME_MS; static final long DESCRIBE_TOPIC_CONFIG_TIMEOUT_MS; }