@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) {
      // Not reaching alerting threshold yet.
      long delayMs = alertTimeMs - nowMs;
      result = AnomalyNotificationResult.check(delayMs);
    } else if (nowMs < selfHealingTimeMs) {
      // Reached alert threshold. Alert but do not fix.
      if (hasNewFailureToAlert(brokerFailures, false)) {
        alert(brokerFailures, false, selfHealingTimeMs, KafkaAnomalyType.BROKER_FAILURE);
      }
      long delayMs = selfHealingTimeMs - nowMs;
      result = AnomalyNotificationResult.check(delayMs);
    } else {
      // Reached auto fix threshold. Alert and fix if self healing is enabled and anomaly is fixable.
      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;
  }
@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 = Map.of(1, failureTime1, 2, failureTime2);
    Map<String, Object> parameterConfigOverrides = Map.of(KAFKA_CRUISE_CONTROL_OBJECT_CONFIG, mockKafkaCruiseControl,
                                                          FAILED_BROKERS_OBJECT_CONFIG, failedBrokers,
                                                          ANOMALY_DETECTION_TIME_MS_OBJECT_CONFIG, failureTime1,
                                                          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.isAlertCalledFor(KafkaAnomalyType.BROKER_FAILURE));

    // Sleep to 1 ms before alert.
    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.isAlertCalledFor(KafkaAnomalyType.BROKER_FAILURE));

    // Sleep 1 ms
    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.isAlertCalledFor(KafkaAnomalyType.BROKER_FAILURE));

    // Sleep to 1 ms before alert.
    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.isAlertCalledFor(KafkaAnomalyType.BROKER_FAILURE));
    assertFalse(anomalyNotifier.isAutoFixTriggeredFor(KafkaAnomalyType.BROKER_FAILURE));

    // Sleep 1 ms
    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.isAlertCalledFor(KafkaAnomalyType.BROKER_FAILURE));
    assertTrue(anomalyNotifier.isAutoFixTriggeredFor(KafkaAnomalyType.BROKER_FAILURE));
    assertFalse(anomalyNotifier.isAlertCalledFor(KafkaAnomalyType.GOAL_VIOLATION));
    assertFalse(anomalyNotifier.isAlertCalledFor(KafkaAnomalyType.METRIC_ANOMALY));
    assertFalse(anomalyNotifier.isAlertCalledFor(KafkaAnomalyType.DISK_FAILURE));
    assertFalse(anomalyNotifier.isAlertCalledFor(KafkaAnomalyType.TOPIC_ANOMALY));

    EasyMock.verify(mockKafkaCruiseControl);
  }