@Test public void testMaintenanceEventTopicCreationUpdateAndRead() throws ExecutionException, InterruptedException, SamplingException { verify(TEST_TOPIC_PARTITION_COUNT, TEST_TOPIC_REPLICATION_FACTOR, TEST_TOPIC_RETENTION_TIME_MS, false); String newPartitionCount = String.valueOf(Integer.parseInt(TEST_TOPIC_PARTITION_COUNT) * 2); String newRF = String.valueOf(Short.parseShort(TEST_TOPIC_REPLICATION_FACTOR) + 1); String newRetentionMs = String.valueOf(Long.MAX_VALUE); KafkaCruiseControl mockKafkaCruiseControl = EasyMock.mock(KafkaCruiseControl.class); Map<String, Object> parameterConfigOverrides = new HashMap<>(4); parameterConfigOverrides.put(KAFKA_CRUISE_CONTROL_OBJECT_CONFIG, mockKafkaCruiseControl); parameterConfigOverrides.put(MAINTENANCE_EVENT_TOPIC_REPLICATION_FACTOR_CONFIG, newRF); parameterConfigOverrides.put(MAINTENANCE_EVENT_TOPIC_PARTITION_COUNT_CONFIG, newPartitionCount); parameterConfigOverrides.put(MAINTENANCE_EVENT_TOPIC_RETENTION_MS_CONFIG, newRetentionMs); long currentMockTime = TEST_REBALANCE_PLAN_TIME + DEFAULT_MAINTENANCE_PLAN_EXPIRATION_MS; EasyMock.expect(mockKafkaCruiseControl.timeMs()).andReturn(currentMockTime).anyTimes(); EasyMock.expect(mockKafkaCruiseControl.config()).andReturn(_config).anyTimes(); EasyMock.replay(mockKafkaCruiseControl); MaintenanceEventReader maintenanceEventReader = _config.getConfiguredInstance(AnomalyDetectorConfig.MAINTENANCE_EVENT_READER_CLASS_CONFIG, MaintenanceEventTopicReader.class, parameterConfigOverrides); assertNotNull(maintenanceEventReader); verify(newPartitionCount, newRF, newRetentionMs, true); Set<MaintenanceEvent> events = maintenanceEventReader.readEvents(TEST_TIMEOUT); EasyMock.verify(mockKafkaCruiseControl); assertEquals(1, events.size()); MaintenanceEvent maintenanceEvent = events.iterator().next(); assertEquals(MAINTENANCE_EVENT, maintenanceEvent.anomalyType()); assertEquals(currentMockTime, maintenanceEvent.detectionTimeMs()); assertEquals(REBALANCE, maintenanceEvent.maintenanceEventType()); }
@Override public Set<MaintenanceEvent> readEvents(Duration timeout) throws SamplingException { LOG.debug("Reading maintenance events."); long eventReadPeriodEndMs = _kafkaCruiseControl.timeMs(); if (refreshPartitionAssignment()) { _lastEventReadPeriodEndTimeMs = eventReadPeriodEndMs; return Collections.emptySet(); } long timeoutEndMs = eventReadPeriodEndMs + timeout.toMillis(); Set<MaintenanceEvent> maintenanceEvents = new HashSet<>(); try { Map<TopicPartition, Long> endOffsets = seekToRelevantOffsets(); LOG.debug("Started to consume from maintenance event topic partitions {}.", _currentPartitionAssignment); _consumer.resume(_consumer.paused()); Set<TopicPartition> partitionsToPause = new HashSet<>(); do { ConsumerRecords<String, MaintenancePlan> records = _consumer.poll(timeout); for (ConsumerRecord<String, MaintenancePlan> record : records) { if (record == null) { LOG.warn("Cannot parse record, please update your Cruise Control version."); continue; } long planGenerationTimeMs = record.value().timeMs(); if (planGenerationTimeMs + _maintenancePlanExpirationMs < eventReadPeriodEndMs) { LOG.warn("Discarding the expired plan {}. (Expired: {} Evaluated: {}).", record.value(), planGenerationTimeMs + _maintenancePlanExpirationMs, eventReadPeriodEndMs); } else if (planGenerationTimeMs >= eventReadPeriodEndMs) { TopicPartition tp = new TopicPartition(record.topic(), record.partition()); LOG.debug("Saw plan {} generated after the end time of event read period {}. Pausing {} at offset {}.", record.value(), eventReadPeriodEndMs, tp, record.offset()); partitionsToPause.add(tp); } else { addMaintenancePlan(record.value(), maintenanceEvents); } } if (!partitionsToPause.isEmpty()) { _consumer.pause(partitionsToPause); partitionsToPause.clear(); } } while (!consumptionDone(_consumer, endOffsets) && _kafkaCruiseControl.timeMs() < timeoutEndMs); if (maintenanceEvents.size() > 0) { LOG.info("Retrieved {} maintenance plans from partitions {} (range [{},{}]).", maintenanceEvents.size(), _currentPartitionAssignment, _lastEventReadPeriodEndTimeMs, eventReadPeriodEndMs); } } finally { _lastEventReadPeriodEndTimeMs = eventReadPeriodEndMs; } return maintenanceEvents; }
MaintenanceEventTopicReader implements MaintenanceEventReader { @Override public Set<MaintenanceEvent> readEvents(Duration timeout) throws SamplingException { LOG.debug("Reading maintenance events."); long eventReadPeriodEndMs = _kafkaCruiseControl.timeMs(); if (refreshPartitionAssignment()) { _lastEventReadPeriodEndTimeMs = eventReadPeriodEndMs; return Collections.emptySet(); } long timeoutEndMs = eventReadPeriodEndMs + timeout.toMillis(); Set<MaintenanceEvent> maintenanceEvents = new HashSet<>(); try { Map<TopicPartition, Long> endOffsets = seekToRelevantOffsets(); LOG.debug("Started to consume from maintenance event topic partitions {}.", _currentPartitionAssignment); _consumer.resume(_consumer.paused()); Set<TopicPartition> partitionsToPause = new HashSet<>(); do { ConsumerRecords<String, MaintenancePlan> records = _consumer.poll(timeout); for (ConsumerRecord<String, MaintenancePlan> record : records) { if (record == null) { LOG.warn("Cannot parse record, please update your Cruise Control version."); continue; } long planGenerationTimeMs = record.value().timeMs(); if (planGenerationTimeMs + _maintenancePlanExpirationMs < eventReadPeriodEndMs) { LOG.warn("Discarding the expired plan {}. (Expired: {} Evaluated: {}).", record.value(), planGenerationTimeMs + _maintenancePlanExpirationMs, eventReadPeriodEndMs); } else if (planGenerationTimeMs >= eventReadPeriodEndMs) { TopicPartition tp = new TopicPartition(record.topic(), record.partition()); LOG.debug("Saw plan {} generated after the end time of event read period {}. Pausing {} at offset {}.", record.value(), eventReadPeriodEndMs, tp, record.offset()); partitionsToPause.add(tp); } else { addMaintenancePlan(record.value(), maintenanceEvents); } } if (!partitionsToPause.isEmpty()) { _consumer.pause(partitionsToPause); partitionsToPause.clear(); } } while (!consumptionDone(_consumer, endOffsets) && _kafkaCruiseControl.timeMs() < timeoutEndMs); if (maintenanceEvents.size() > 0) { LOG.info("Retrieved {} maintenance plans from partitions {} (range [{},{}]).", maintenanceEvents.size(), _currentPartitionAssignment, _lastEventReadPeriodEndTimeMs, eventReadPeriodEndMs); } } finally { _lastEventReadPeriodEndTimeMs = eventReadPeriodEndMs; } return maintenanceEvents; } }
MaintenanceEventTopicReader implements MaintenanceEventReader { @Override public Set<MaintenanceEvent> readEvents(Duration timeout) throws SamplingException { LOG.debug("Reading maintenance events."); long eventReadPeriodEndMs = _kafkaCruiseControl.timeMs(); if (refreshPartitionAssignment()) { _lastEventReadPeriodEndTimeMs = eventReadPeriodEndMs; return Collections.emptySet(); } long timeoutEndMs = eventReadPeriodEndMs + timeout.toMillis(); Set<MaintenanceEvent> maintenanceEvents = new HashSet<>(); try { Map<TopicPartition, Long> endOffsets = seekToRelevantOffsets(); LOG.debug("Started to consume from maintenance event topic partitions {}.", _currentPartitionAssignment); _consumer.resume(_consumer.paused()); Set<TopicPartition> partitionsToPause = new HashSet<>(); do { ConsumerRecords<String, MaintenancePlan> records = _consumer.poll(timeout); for (ConsumerRecord<String, MaintenancePlan> record : records) { if (record == null) { LOG.warn("Cannot parse record, please update your Cruise Control version."); continue; } long planGenerationTimeMs = record.value().timeMs(); if (planGenerationTimeMs + _maintenancePlanExpirationMs < eventReadPeriodEndMs) { LOG.warn("Discarding the expired plan {}. (Expired: {} Evaluated: {}).", record.value(), planGenerationTimeMs + _maintenancePlanExpirationMs, eventReadPeriodEndMs); } else if (planGenerationTimeMs >= eventReadPeriodEndMs) { TopicPartition tp = new TopicPartition(record.topic(), record.partition()); LOG.debug("Saw plan {} generated after the end time of event read period {}. Pausing {} at offset {}.", record.value(), eventReadPeriodEndMs, tp, record.offset()); partitionsToPause.add(tp); } else { addMaintenancePlan(record.value(), maintenanceEvents); } } if (!partitionsToPause.isEmpty()) { _consumer.pause(partitionsToPause); partitionsToPause.clear(); } } while (!consumptionDone(_consumer, endOffsets) && _kafkaCruiseControl.timeMs() < timeoutEndMs); if (maintenanceEvents.size() > 0) { LOG.info("Retrieved {} maintenance plans from partitions {} (range [{},{}]).", maintenanceEvents.size(), _currentPartitionAssignment, _lastEventReadPeriodEndTimeMs, eventReadPeriodEndMs); } } finally { _lastEventReadPeriodEndTimeMs = eventReadPeriodEndMs; } return maintenanceEvents; }  }
MaintenanceEventTopicReader implements MaintenanceEventReader { @Override public Set<MaintenanceEvent> readEvents(Duration timeout) throws SamplingException { LOG.debug("Reading maintenance events."); long eventReadPeriodEndMs = _kafkaCruiseControl.timeMs(); if (refreshPartitionAssignment()) { _lastEventReadPeriodEndTimeMs = eventReadPeriodEndMs; return Collections.emptySet(); } long timeoutEndMs = eventReadPeriodEndMs + timeout.toMillis(); Set<MaintenanceEvent> maintenanceEvents = new HashSet<>(); try { Map<TopicPartition, Long> endOffsets = seekToRelevantOffsets(); LOG.debug("Started to consume from maintenance event topic partitions {}.", _currentPartitionAssignment); _consumer.resume(_consumer.paused()); Set<TopicPartition> partitionsToPause = new HashSet<>(); do { ConsumerRecords<String, MaintenancePlan> records = _consumer.poll(timeout); for (ConsumerRecord<String, MaintenancePlan> record : records) { if (record == null) { LOG.warn("Cannot parse record, please update your Cruise Control version."); continue; } long planGenerationTimeMs = record.value().timeMs(); if (planGenerationTimeMs + _maintenancePlanExpirationMs < eventReadPeriodEndMs) { LOG.warn("Discarding the expired plan {}. (Expired: {} Evaluated: {}).", record.value(), planGenerationTimeMs + _maintenancePlanExpirationMs, eventReadPeriodEndMs); } else if (planGenerationTimeMs >= eventReadPeriodEndMs) { TopicPartition tp = new TopicPartition(record.topic(), record.partition()); LOG.debug("Saw plan {} generated after the end time of event read period {}. Pausing {} at offset {}.", record.value(), eventReadPeriodEndMs, tp, record.offset()); partitionsToPause.add(tp); } else { addMaintenancePlan(record.value(), maintenanceEvents); } } if (!partitionsToPause.isEmpty()) { _consumer.pause(partitionsToPause); partitionsToPause.clear(); } } while (!consumptionDone(_consumer, endOffsets) && _kafkaCruiseControl.timeMs() < timeoutEndMs); if (maintenanceEvents.size() > 0) { LOG.info("Retrieved {} maintenance plans from partitions {} (range [{},{}]).", maintenanceEvents.size(), _currentPartitionAssignment, _lastEventReadPeriodEndTimeMs, eventReadPeriodEndMs); } } finally { _lastEventReadPeriodEndTimeMs = eventReadPeriodEndMs; } return maintenanceEvents; }  @Override Set<MaintenanceEvent> readEvents(Duration timeout); @Override void configure(Map<String, ?> configs); @Override void close(); }
MaintenanceEventTopicReader implements MaintenanceEventReader { @Override public Set<MaintenanceEvent> readEvents(Duration timeout) throws SamplingException { LOG.debug("Reading maintenance events."); long eventReadPeriodEndMs = _kafkaCruiseControl.timeMs(); if (refreshPartitionAssignment()) { _lastEventReadPeriodEndTimeMs = eventReadPeriodEndMs; return Collections.emptySet(); } long timeoutEndMs = eventReadPeriodEndMs + timeout.toMillis(); Set<MaintenanceEvent> maintenanceEvents = new HashSet<>(); try { Map<TopicPartition, Long> endOffsets = seekToRelevantOffsets(); LOG.debug("Started to consume from maintenance event topic partitions {}.", _currentPartitionAssignment); _consumer.resume(_consumer.paused()); Set<TopicPartition> partitionsToPause = new HashSet<>(); do { ConsumerRecords<String, MaintenancePlan> records = _consumer.poll(timeout); for (ConsumerRecord<String, MaintenancePlan> record : records) { if (record == null) { LOG.warn("Cannot parse record, please update your Cruise Control version."); continue; } long planGenerationTimeMs = record.value().timeMs(); if (planGenerationTimeMs + _maintenancePlanExpirationMs < eventReadPeriodEndMs) { LOG.warn("Discarding the expired plan {}. (Expired: {} Evaluated: {}).", record.value(), planGenerationTimeMs + _maintenancePlanExpirationMs, eventReadPeriodEndMs); } else if (planGenerationTimeMs >= eventReadPeriodEndMs) { TopicPartition tp = new TopicPartition(record.topic(), record.partition()); LOG.debug("Saw plan {} generated after the end time of event read period {}. Pausing {} at offset {}.", record.value(), eventReadPeriodEndMs, tp, record.offset()); partitionsToPause.add(tp); } else { addMaintenancePlan(record.value(), maintenanceEvents); } } if (!partitionsToPause.isEmpty()) { _consumer.pause(partitionsToPause); partitionsToPause.clear(); } } while (!consumptionDone(_consumer, endOffsets) && _kafkaCruiseControl.timeMs() < timeoutEndMs); if (maintenanceEvents.size() > 0) { LOG.info("Retrieved {} maintenance plans from partitions {} (range [{},{}]).", maintenanceEvents.size(), _currentPartitionAssignment, _lastEventReadPeriodEndTimeMs, eventReadPeriodEndMs); } } finally { _lastEventReadPeriodEndTimeMs = eventReadPeriodEndMs; } return maintenanceEvents; }  @Override Set<MaintenanceEvent> readEvents(Duration timeout); @Override void configure(Map<String, ?> configs); @Override void close(); static final String MAINTENANCE_PLAN_EXPIRATION_MS_CONFIG; static final long DEFAULT_MAINTENANCE_PLAN_EXPIRATION_MS; static final String MAINTENANCE_EVENT_TOPIC_CONFIG; static final String DEFAULT_MAINTENANCE_EVENT_TOPIC; static final String MAINTENANCE_EVENT_TOPIC_REPLICATION_FACTOR_CONFIG; static final short DEFAULT_MAINTENANCE_EVENT_TOPIC_REPLICATION_FACTOR; static final String MAINTENANCE_EVENT_TOPIC_PARTITION_COUNT_CONFIG; static final int DEFAULT_MAINTENANCE_EVENT_TOPIC_PARTITION_COUNT; static final String MAINTENANCE_EVENT_TOPIC_RETENTION_MS_CONFIG; static final long DEFAULT_MAINTENANCE_EVENT_TOPIC_RETENTION_TIME_MS; static final Duration CONSUMER_CLOSE_TIMEOUT; static final String CONSUMER_CLIENT_ID_PREFIX; static final long INIT_MAINTENANCE_HISTORY_MS; }