@Test public void testOnBrokerFailure() { final long failureTime1 = 200L; final long failureTime2 = 400L; final long startTime = 500L; KafkaCruiseControl mockKafkaCruiseControl = EasyMock.mock(KafkaCruiseControl.class); Properties props = KafkaCruiseControlUnitTestUtils.getKafkaCruiseControlProperties(); KafkaCruiseControlConfig kafkaCruiseControlConfig = new KafkaCruiseControlConfig(props); EasyMock.expect(mockKafkaCruiseControl.config()).andReturn(kafkaCruiseControlConfig).atLeastOnce(); EasyMock.replay(mockKafkaCruiseControl); Time mockTime = new MockTime(0, startTime, TimeUnit.NANOSECONDS.convert(startTime, TimeUnit.MILLISECONDS)); TestingBrokerFailureAutoFixNotifier anomalyNotifier = new TestingBrokerFailureAutoFixNotifier(mockTime); anomalyNotifier.configure(Collections.singletonMap(SelfHealingNotifier.SELF_HEALING_BROKER_FAILURE_ENABLED_CONFIG, "true")); Map<Integer, Long> failedBrokers = new HashMap<>(); failedBrokers.put(1, failureTime1); failedBrokers.put(2, failureTime2); Map<String, Object> parameterConfigOverrides = new HashMap<>(4); parameterConfigOverrides.put(KAFKA_CRUISE_CONTROL_OBJECT_CONFIG, mockKafkaCruiseControl); parameterConfigOverrides.put(FAILED_BROKERS_OBJECT_CONFIG, failedBrokers); parameterConfigOverrides.put(ANOMALY_DETECTION_TIME_MS_OBJECT_CONFIG, failureTime1); parameterConfigOverrides.put(BROKER_FAILURES_FIXABLE_CONFIG, true); AnomalyNotificationResult result = anomalyNotifier.onBrokerFailure( kafkaCruiseControlConfig.getConfiguredInstance(AnomalyDetectorConfig.BROKER_FAILURES_CLASS_CONFIG, BrokerFailures.class, parameterConfigOverrides)); assertEquals(AnomalyNotificationResult.Action.CHECK, result.action()); assertEquals(SelfHealingNotifier.DEFAULT_ALERT_THRESHOLD_MS + failureTime1 - mockTime.milliseconds(), result.delay()); assertFalse(anomalyNotifier._alertCalled.get(KafkaAnomalyType.BROKER_FAILURE)); mockTime.sleep(result.delay() - 1); result = anomalyNotifier.onBrokerFailure( kafkaCruiseControlConfig.getConfiguredInstance(AnomalyDetectorConfig.BROKER_FAILURES_CLASS_CONFIG, BrokerFailures.class, parameterConfigOverrides)); assertEquals(AnomalyNotificationResult.Action.CHECK, result.action()); assertEquals(1, result.delay()); assertFalse(anomalyNotifier._alertCalled.get(KafkaAnomalyType.BROKER_FAILURE)); mockTime.sleep(1); anomalyNotifier.resetAlert(KafkaAnomalyType.BROKER_FAILURE); result = anomalyNotifier.onBrokerFailure( kafkaCruiseControlConfig.getConfiguredInstance(AnomalyDetectorConfig.BROKER_FAILURES_CLASS_CONFIG, BrokerFailures.class, parameterConfigOverrides)); assertEquals(AnomalyNotificationResult.Action.CHECK, result.action()); assertEquals(SelfHealingNotifier.DEFAULT_AUTO_FIX_THRESHOLD_MS + failureTime1 - mockTime.milliseconds(), result.delay()); assertTrue(anomalyNotifier._alertCalled.get(KafkaAnomalyType.BROKER_FAILURE)); mockTime.sleep(result.delay() - 1); anomalyNotifier.resetAlert(KafkaAnomalyType.BROKER_FAILURE); result = anomalyNotifier.onBrokerFailure( kafkaCruiseControlConfig.getConfiguredInstance(AnomalyDetectorConfig.BROKER_FAILURES_CLASS_CONFIG, BrokerFailures.class, parameterConfigOverrides)); assertEquals(AnomalyNotificationResult.Action.CHECK, result.action()); assertEquals(1, result.delay()); assertFalse(anomalyNotifier._alertCalled.get(KafkaAnomalyType.BROKER_FAILURE)); assertFalse(anomalyNotifier._autoFixTriggered.get(KafkaAnomalyType.BROKER_FAILURE)); mockTime.sleep(1); anomalyNotifier.resetAlert(KafkaAnomalyType.BROKER_FAILURE); result = anomalyNotifier.onBrokerFailure( kafkaCruiseControlConfig.getConfiguredInstance(AnomalyDetectorConfig.BROKER_FAILURES_CLASS_CONFIG, BrokerFailures.class, parameterConfigOverrides)); assertEquals(AnomalyNotificationResult.Action.FIX, result.action()); assertEquals(-1L, result.delay()); assertTrue(anomalyNotifier._alertCalled.get(KafkaAnomalyType.BROKER_FAILURE)); assertTrue(anomalyNotifier._autoFixTriggered.get(KafkaAnomalyType.BROKER_FAILURE)); assertFalse(anomalyNotifier._alertCalled.get(KafkaAnomalyType.GOAL_VIOLATION)); assertFalse(anomalyNotifier._alertCalled.get(KafkaAnomalyType.METRIC_ANOMALY)); assertFalse(anomalyNotifier._alertCalled.get(KafkaAnomalyType.DISK_FAILURE)); assertFalse(anomalyNotifier._alertCalled.get(KafkaAnomalyType.TOPIC_ANOMALY)); }
@Override public AnomalyNotificationResult onBrokerFailure(BrokerFailures brokerFailures) { long earliestFailureTimeMs = Long.MAX_VALUE; for (long t : brokerFailures.failedBrokers().values()) { earliestFailureTimeMs = Math.min(earliestFailureTimeMs, t); } long nowMs = _time.milliseconds(); long alertTimeMs = earliestFailureTimeMs + _brokerFailureAlertThresholdMs; long selfHealingTimeMs = earliestFailureTimeMs + _selfHealingThresholdMs; AnomalyNotificationResult result = null; if (nowMs < alertTimeMs) { long delayMs = alertTimeMs - nowMs; result = AnomalyNotificationResult.check(delayMs); } else if (nowMs < selfHealingTimeMs) { if (hasNewFailureToAlert(brokerFailures, false)) { alert(brokerFailures, false, selfHealingTimeMs, KafkaAnomalyType.BROKER_FAILURE); } long delay = selfHealingTimeMs - nowMs; result = AnomalyNotificationResult.check(delay); } else { boolean autoFixTriggered = _selfHealingEnabled.get(KafkaAnomalyType.BROKER_FAILURE) && brokerFailures.fixable(); if (hasNewFailureToAlert(brokerFailures, autoFixTriggered)) { alert(brokerFailures, autoFixTriggered, selfHealingTimeMs, KafkaAnomalyType.BROKER_FAILURE); } result = autoFixTriggered ? AnomalyNotificationResult.fix() : AnomalyNotificationResult.ignore(); } return result; }
SelfHealingNotifier implements AnomalyNotifier { @Override public AnomalyNotificationResult onBrokerFailure(BrokerFailures brokerFailures) { long earliestFailureTimeMs = Long.MAX_VALUE; for (long t : brokerFailures.failedBrokers().values()) { earliestFailureTimeMs = Math.min(earliestFailureTimeMs, t); } long nowMs = _time.milliseconds(); long alertTimeMs = earliestFailureTimeMs + _brokerFailureAlertThresholdMs; long selfHealingTimeMs = earliestFailureTimeMs + _selfHealingThresholdMs; AnomalyNotificationResult result = null; if (nowMs < alertTimeMs) { long delayMs = alertTimeMs - nowMs; result = AnomalyNotificationResult.check(delayMs); } else if (nowMs < selfHealingTimeMs) { if (hasNewFailureToAlert(brokerFailures, false)) { alert(brokerFailures, false, selfHealingTimeMs, KafkaAnomalyType.BROKER_FAILURE); } long delay = selfHealingTimeMs - nowMs; result = AnomalyNotificationResult.check(delay); } else { boolean autoFixTriggered = _selfHealingEnabled.get(KafkaAnomalyType.BROKER_FAILURE) && brokerFailures.fixable(); if (hasNewFailureToAlert(brokerFailures, autoFixTriggered)) { alert(brokerFailures, autoFixTriggered, selfHealingTimeMs, KafkaAnomalyType.BROKER_FAILURE); } result = autoFixTriggered ? AnomalyNotificationResult.fix() : AnomalyNotificationResult.ignore(); } return result; } }
SelfHealingNotifier implements AnomalyNotifier { @Override public AnomalyNotificationResult onBrokerFailure(BrokerFailures brokerFailures) { long earliestFailureTimeMs = Long.MAX_VALUE; for (long t : brokerFailures.failedBrokers().values()) { earliestFailureTimeMs = Math.min(earliestFailureTimeMs, t); } long nowMs = _time.milliseconds(); long alertTimeMs = earliestFailureTimeMs + _brokerFailureAlertThresholdMs; long selfHealingTimeMs = earliestFailureTimeMs + _selfHealingThresholdMs; AnomalyNotificationResult result = null; if (nowMs < alertTimeMs) { long delayMs = alertTimeMs - nowMs; result = AnomalyNotificationResult.check(delayMs); } else if (nowMs < selfHealingTimeMs) { if (hasNewFailureToAlert(brokerFailures, false)) { alert(brokerFailures, false, selfHealingTimeMs, KafkaAnomalyType.BROKER_FAILURE); } long delay = selfHealingTimeMs - nowMs; result = AnomalyNotificationResult.check(delay); } else { boolean autoFixTriggered = _selfHealingEnabled.get(KafkaAnomalyType.BROKER_FAILURE) && brokerFailures.fixable(); if (hasNewFailureToAlert(brokerFailures, autoFixTriggered)) { alert(brokerFailures, autoFixTriggered, selfHealingTimeMs, KafkaAnomalyType.BROKER_FAILURE); } result = autoFixTriggered ? AnomalyNotificationResult.fix() : AnomalyNotificationResult.ignore(); } return result; } SelfHealingNotifier();   SelfHealingNotifier(Time time); }
SelfHealingNotifier implements AnomalyNotifier { @Override public AnomalyNotificationResult onBrokerFailure(BrokerFailures brokerFailures) { long earliestFailureTimeMs = Long.MAX_VALUE; for (long t : brokerFailures.failedBrokers().values()) { earliestFailureTimeMs = Math.min(earliestFailureTimeMs, t); } long nowMs = _time.milliseconds(); long alertTimeMs = earliestFailureTimeMs + _brokerFailureAlertThresholdMs; long selfHealingTimeMs = earliestFailureTimeMs + _selfHealingThresholdMs; AnomalyNotificationResult result = null; if (nowMs < alertTimeMs) { long delayMs = alertTimeMs - nowMs; result = AnomalyNotificationResult.check(delayMs); } else if (nowMs < selfHealingTimeMs) { if (hasNewFailureToAlert(brokerFailures, false)) { alert(brokerFailures, false, selfHealingTimeMs, KafkaAnomalyType.BROKER_FAILURE); } long delay = selfHealingTimeMs - nowMs; result = AnomalyNotificationResult.check(delay); } else { boolean autoFixTriggered = _selfHealingEnabled.get(KafkaAnomalyType.BROKER_FAILURE) && brokerFailures.fixable(); if (hasNewFailureToAlert(brokerFailures, autoFixTriggered)) { alert(brokerFailures, autoFixTriggered, selfHealingTimeMs, KafkaAnomalyType.BROKER_FAILURE); } result = autoFixTriggered ? AnomalyNotificationResult.fix() : AnomalyNotificationResult.ignore(); } return result; } SelfHealingNotifier();   SelfHealingNotifier(Time time); @Override AnomalyNotificationResult onGoalViolation(GoalViolations goalViolations); @Override AnomalyNotificationResult onMetricAnomaly(KafkaMetricAnomaly metricAnomaly); @Override AnomalyNotificationResult onTopicAnomaly(TopicAnomaly topicAnomaly); @Override AnomalyNotificationResult onMaintenanceEvent(MaintenanceEvent maintenanceEvent); @Override AnomalyNotificationResult onDiskFailure(DiskFailures diskFailures); @Override Map<AnomalyType, Boolean> selfHealingEnabled(); @Override synchronized boolean setSelfHealingFor(AnomalyType anomalyType, boolean isSelfHealingEnabled); @Override synchronized Map<AnomalyType, Float> selfHealingEnabledRatio(); @Override AnomalyNotificationResult onBrokerFailure(BrokerFailures brokerFailures); void alert(Object anomaly, boolean autoFixTriggered, long selfHealingStartTime, AnomalyType anomalyType); @Override void configure(Map<String, ?> config); @Override long uptimeMs(long nowMs); }
SelfHealingNotifier implements AnomalyNotifier { @Override public AnomalyNotificationResult onBrokerFailure(BrokerFailures brokerFailures) { long earliestFailureTimeMs = Long.MAX_VALUE; for (long t : brokerFailures.failedBrokers().values()) { earliestFailureTimeMs = Math.min(earliestFailureTimeMs, t); } long nowMs = _time.milliseconds(); long alertTimeMs = earliestFailureTimeMs + _brokerFailureAlertThresholdMs; long selfHealingTimeMs = earliestFailureTimeMs + _selfHealingThresholdMs; AnomalyNotificationResult result = null; if (nowMs < alertTimeMs) { long delayMs = alertTimeMs - nowMs; result = AnomalyNotificationResult.check(delayMs); } else if (nowMs < selfHealingTimeMs) { if (hasNewFailureToAlert(brokerFailures, false)) { alert(brokerFailures, false, selfHealingTimeMs, KafkaAnomalyType.BROKER_FAILURE); } long delay = selfHealingTimeMs - nowMs; result = AnomalyNotificationResult.check(delay); } else { boolean autoFixTriggered = _selfHealingEnabled.get(KafkaAnomalyType.BROKER_FAILURE) && brokerFailures.fixable(); if (hasNewFailureToAlert(brokerFailures, autoFixTriggered)) { alert(brokerFailures, autoFixTriggered, selfHealingTimeMs, KafkaAnomalyType.BROKER_FAILURE); } result = autoFixTriggered ? AnomalyNotificationResult.fix() : AnomalyNotificationResult.ignore(); } return result; } SelfHealingNotifier();   SelfHealingNotifier(Time time); @Override AnomalyNotificationResult onGoalViolation(GoalViolations goalViolations); @Override AnomalyNotificationResult onMetricAnomaly(KafkaMetricAnomaly metricAnomaly); @Override AnomalyNotificationResult onTopicAnomaly(TopicAnomaly topicAnomaly); @Override AnomalyNotificationResult onMaintenanceEvent(MaintenanceEvent maintenanceEvent); @Override AnomalyNotificationResult onDiskFailure(DiskFailures diskFailures); @Override Map<AnomalyType, Boolean> selfHealingEnabled(); @Override synchronized boolean setSelfHealingFor(AnomalyType anomalyType, boolean isSelfHealingEnabled); @Override synchronized Map<AnomalyType, Float> selfHealingEnabledRatio(); @Override AnomalyNotificationResult onBrokerFailure(BrokerFailures brokerFailures); void alert(Object anomaly, boolean autoFixTriggered, long selfHealingStartTime, AnomalyType anomalyType); @Override void configure(Map<String, ?> config); @Override long uptimeMs(long nowMs); static final String BROKER_FAILURE_ALERT_THRESHOLD_MS_CONFIG; static final String SELF_HEALING_ENABLED_CONFIG; static final String SELF_HEALING_BROKER_FAILURE_ENABLED_CONFIG; static final String SELF_HEALING_GOAL_VIOLATION_ENABLED_CONFIG; static final String SELF_HEALING_METRIC_ANOMALY_ENABLED_CONFIG; static final String SELF_HEALING_DISK_FAILURE_ENABLED_CONFIG; static final String SELF_HEALING_TOPIC_ANOMALY_ENABLED_CONFIG; static final String SELF_HEALING_MAINTENANCE_EVENT_ENABLED_CONFIG; static final String BROKER_FAILURE_SELF_HEALING_THRESHOLD_MS_CONFIG; }