Repository: linkedin/cruise-control (java)

Issue:
Handling of broker failure anomalies ignored due to temporary self heal disable
## Issue

When self healing is disabled and a broker failure anomaly reaches the auto fix threshold, today's behavior is the anomaly gets ignored.
In cases where self healing is permanently disabled this is accepted. However in cases where self healing is disabled temporarily, this behavior is not ideal as the under replicated partitions from this broker failure gets fixed only with a human intervention or subsequent self heal for another anomaly.

## Proposed Approach

When the broker failure anomaly reaches the auto fix threshold and self healing is disabled, instead of ignoring the anomaly recheck the anomaly after certain time period in the hope that self healing gets re enabled back and a fix can be triggered.

- Introduce a configurable max retry count so that the anomaly recheck is not done forever in cases where self healing is permanently disabled.
- Introduce a a recheck interval config.


Pull request:
Fix ignored broker failure anomalies when self healing is disabled
## Summary

### Why:  
When self healing is disabled and a broker failure anomaly reaches the auto fix threshold, today's behavior is the anomaly gets ignored.
In cases where self healing is permanently disabled this is accepted. However in cases where self healing is disabled temporarily, this behavior is not ideal as the under replicated partitions from this broker failure gets fixed only with a human intervention or subsequent self heal for another anomaly.
    
### What: 
When the broker failure anomaly reaches the auto fix threshold and self healing is disabled, instead of ignoring the anomaly, recheck the anomaly after certain delay in the hope that self healing gets re-enabled back and a fix can be triggered. Introduce a configurable max retry count so that the anomaly recheck is not done forever in cases where self healing is permanently disabled. After the tries exceed the max allowed, the anomaly can be ignored. In addition introduce a config which gives the delay after which subsequent check is executed.

## Expected Behavior
Anomaly should be checked after certain time when self healing is disabled.

## Actual Behavior
Anomaly is getting ignored

## Steps to Reproduce
1. Disable self healing.
2. Trigger a broker failure.
3. Wait until the broker failure anomaly reaches the auto fix threshold.
4. Reenable self healing.
5. The broker failure anomaly is not fixed.


## Additional evidence

Testing Logs:

Scenario 1:
1. Self healing is disabled.
2. Broker 1468 fails.
3. Broker failure anomaly reaches the auto fix threshold and self heal is still disabled. The anomaly should be ignored
4. Note: Tested with max retry count 3 and recheck interval 2 minutes. The first check with delay is for the case where auto fix threshold is not yet reached.

```
# Self heal disabled

USER TASK ID                          CLIENT ADDRESS                   START TIME            STATUS      REQUEST URL
9cf37b50-4dc9-4407-affb-dfd91aa95a0a  <client_ip> 2025-04-21T19:14:04Z  Completed   POST /kafkacruisecontrol/admin?disable_self_healing_for=broker_failure

# Broker anomaly and status timeline

anomalyId       brokerId        status  detectionTime   statusUpdateTime
017e6f7e-13d5-407c-aaba-dfcae46bbe85    1468    CHECK_WITH_DELAY        2025-04-21 19:17:20     2025-04-21 19:17:20
94d79b77-993b-4ccb-b67c-eedf6073a47b    1468    CHECK_WITH_DELAY        2025-04-21 19:19:20     2025-04-21 19:19:20
2c7fe743-1792-4c0e-8b75-04e2acecad5b    1468    CHECK_WITH_DELAY        2025-04-21 19:21:20     2025-04-21 19:21:20
4f4b3cc7-2ee4-48d8-bee5-f32a3a19cc05    1468    CHECK_WITH_DELAY        2025-04-21 19:23:20     2025-04-21 19:23:20
5aab73a2-ae6d-4865-8f34-73a8ec1ba75c    1468    IGNORED 2025-04-21 19:25:20     2025-04-21 19:25:20

```

Scenario 2:
1. Self healing is disabled.
2. Broker 4600 failed.
3. Broker failure anomaly reached auto fix threshold. 
4. After 2 retries of check anomaly, self healing is reenabled. Fix should be started.
5. Note: Tested with max retry count 3 and recheck interval 2 minutes.

```
# Self heal timeline

1b58c1f3-5dfb-45b0-9958-529d3418ea60 <client_ip>                  2025-04-21T20:00:44Z  Completed   POST /kafkacruisecontrol/admin?disable_self_healing_for=broker_failure
9a9be753-7f13-431e-9093-e04d719e7979  <client_ip>  2025-04-21T20:13:07Z  Completed   POST /kafkacruisecontrol/admin?enable_self_healing_for=broker_failure

# Broker anomaly and status timeline

10331511-3ae1-4028-a487-20fce5175458    4600    CHECK_WITH_DELAY        2025-04-21 20:08:43     2025-04-21 20:08:43
e6ead586-cab6-4362-883b-fc33664470a7    4600    CHECK_WITH_DELAY        2025-04-21 20:10:43     2025-04-21 20:10:43
7283326e-b7b2-486d-85b4-b55decc4e228    4600    CHECK_WITH_DELAY        2025-04-21 20:12:43     2025-04-21 20:12:43
ab4e40b2-6df1-4cf0-a6a8-444d340d75d7    4600    FIX_STARTED     2025-04-21 20:14:43     2025-04-21 20:14:45
```

## Categorization
- [ ] documentation
- [x] bugfix
- [ ] new feature
- [ ] refactor
- [ ] security/CVE
- [ ] other

This PR resolves #2269.


Patch:
diff --git a/cruise-control/src/main/java/com/linkedin/kafka/cruisecontrol/detector/AbstractBrokerFailureDetector.java b/cruise-control/src/main/java/com/linkedin/kafka/cruisecontrol/detector/AbstractBrokerFailureDetector.java
index c241cadd..08a0b6b5 100644
--- a/cruise-control/src/main/java/com/linkedin/kafka/cruisecontrol/detector/AbstractBrokerFailureDetector.java
+++ b/cruise-control/src/main/java/com/linkedin/kafka/cruisecontrol/detector/AbstractBrokerFailureDetector.java
@@ -39,6 +39,7 @@ public abstract class AbstractBrokerFailureDetector extends AbstractAnomalyDetec
   public static final String FAILED_BROKERS_OBJECT_CONFIG = "failed.brokers.object";
   // Config to indicate whether detected broker failures are fixable or not.
   public static final String BROKER_FAILURES_FIXABLE_CONFIG = "broker.failures.fixable.object";
+  public static final String BROKER_FAILURE_CHECK_WITH_DELAY_RETRY_COUNT = "broker.failure.check.with.delay.retry.count";
 
   protected final Map<Integer, Long> _failedBrokers;
   protected final File _failedBrokersFile;
@@ -62,6 +63,16 @@ public abstract class AbstractBrokerFailureDetector extends AbstractAnomalyDetec
    * @param skipReportingIfNotUpdated {@code true} if broker failure reporting will be skipped if failed brokers have not changed.
    */
   synchronized void detectBrokerFailures(boolean skipReportingIfNotUpdated) {
+    detectBrokerFailures(skipReportingIfNotUpdated, 0);
+  }
+
+  /**
+   * Detect broker failures. Skip reporting if the failed brokers have not changed and skipReportingIfNotUpdated is true.
+   *
+   * @param skipReportingIfNotUpdated {@code true} if broker failure reporting will be skipped if failed brokers have not changed.
+   * @param brokerFailureCheckWithDelayRetryCount {@code true} to maintain the brokerFailureCheckWithDelayRetryCount
+   */
+  synchronized void detectBrokerFailures(boolean skipReportingIfNotUpdated, int brokerFailureCheckWithDelayRetryCount) {
     try {
       _aliveBrokers = aliveBrokers();
 
@@ -73,7 +84,8 @@ public abstract class AbstractBrokerFailureDetector extends AbstractAnomalyDetec
       }
       if (!skipReportingIfNotUpdated || updated) {
         // Report the failures to anomaly detector to handle.
-        reportBrokerFailures();
+        // Pass the current retry count to the broker failure.
+        reportBrokerFailures(brokerFailureCheckWithDelayRetryCount);
       }
     } catch (Throwable e) {
       LOG.warn("Broker failure detector received exception: ", e);
@@ -185,7 +197,7 @@ public abstract class AbstractBrokerFailureDetector extends AbstractAnomalyDetec
            || (double) failedBrokerCount / (failedBrokerCount + aliveBrokerCount) > _fixableFailedBrokerPercentageThreshold;
   }
 
-  private void reportBrokerFailures() {
+  private void reportBrokerFailures(int brokerFailureCheckWithDelayRetryCount) {
     if (!_failedBrokers.isEmpty()) {
       Map<String, Object> parameterConfigOverrides = new HashMap<>();
       parameterConfigOverrides.put(KAFKA_CRUISE_CONTROL_OBJECT_CONFIG, _kafkaCruiseControl);
@@ -194,6 +206,7 @@ public abstract class AbstractBrokerFailureDetector extends AbstractAnomalyDetec
       parameterConfigOverrides.put(ANOMALY_DETECTION_TIME_MS_OBJECT_CONFIG, _kafkaCruiseControl.timeMs());
       parameterConfigOverrides.put(BROKER_FAILURES_FIXABLE_CONFIG,
                                    !tooManyFailedBrokers(failedBrokers.size(), _aliveBrokers.size()));
+      parameterConfigOverrides.put(BROKER_FAILURE_CHECK_WITH_DELAY_RETRY_COUNT, brokerFailureCheckWithDelayRetryCount);
 
       BrokerFailures brokerFailures = _kafkaCruiseControl.config().getConfiguredInstance(AnomalyDetectorConfig.BROKER_FAILURES_CLASS_CONFIG,
                                                                                          BrokerFailures.class,
diff --git a/cruise-control/src/main/java/com/linkedin/kafka/cruisecontrol/detector/AnomalyDetectorManager.java b/cruise-control/src/main/java/com/linkedin/kafka/cruisecontrol/detector/AnomalyDetectorManager.java
index 8d86931f..afa26aa6 100644
--- a/cruise-control/src/main/java/com/linkedin/kafka/cruisecontrol/detector/AnomalyDetectorManager.java
+++ b/cruise-control/src/main/java/com/linkedin/kafka/cruisecontrol/detector/AnomalyDetectorManager.java
@@ -350,7 +350,7 @@ public class AnomalyDetectorManager {
         _anomalyInProgress = null;
         try {
           _anomalyInProgress = _anomalies.take();
-          LOG.trace("Processing anomaly {}.", _anomalyInProgress);
+          LOG.trace("Processing anomaly id: {} {}.", _anomalyInProgress.anomalyId(), _anomalyInProgress);
           if (_anomalyInProgress == SHUTDOWN_ANOMALY) {
             // Service has shutdown.
             _anomalyInProgress = null;
@@ -376,8 +376,8 @@ public class AnomalyDetectorManager {
           postProcessAnomalyInProgress = true;
         }
         if (postProcessAnomalyInProgress) {
-          LOG.info("Post processing anomaly {}.", _anomalyInProgress);
-          postProcessAnomalyInProgress(_brokerFailureDetectionBackoffMs);
+          LOG.info("Post processing anomaly {} id: {}.", _anomalyInProgress, _anomalyInProgress.anomalyId());
+          postProcessAnomalyInProgress(_brokerFailureDetectionBackoffMs, false);
         }
       }
       LOG.info("Anomaly handler exited.");
@@ -392,7 +392,7 @@ public class AnomalyDetectorManager {
       ExecutorState.State executionState = _kafkaCruiseControl.executionState();
       if (executionState != ExecutorState.State.NO_TASK_IN_PROGRESS && !_anomalyInProgress.stopOngoingExecution()) {
         LOG.info("Post processing anomaly {} because executor is in {} state.", _anomalyInProgress, executionState);
-        postProcessAnomalyInProgress(_brokerFailureDetectionBackoffMs);
+        postProcessAnomalyInProgress(_brokerFailureDetectionBackoffMs, false);
       } else {
         processAnomalyInProgress(anomalyType);
       }
@@ -415,7 +415,7 @@ public class AnomalyDetectorManager {
             break;
           case CHECK:
             LOG.info("Post processing anomaly {} for {}.", _anomalyInProgress, AnomalyState.Status.CHECK_WITH_DELAY);
-            postProcessAnomalyInProgress(notificationResult.delay());
+            postProcessAnomalyInProgress(notificationResult.delay(), true);
             break;
           case IGNORE:
             _anomalyDetectorState.onAnomalyHandle(_anomalyInProgress, AnomalyState.Status.IGNORED);
@@ -475,8 +475,9 @@ public class AnomalyDetectorManager {
      * schedules a broker failure detection after the given delay.
      *
      * @param delayMs The delay for broker failure detection.
+     * @param carryForwardRetryCount Retry count is carried forward only in case of check with delays done for the broker failure anomalies
      */
-    private void postProcessAnomalyInProgress(long delayMs) {
+    private void postProcessAnomalyInProgress(long delayMs, boolean carryForwardRetryCount) {
       // Anomaly detector does delayed check for broker failures, otherwise it ignores the anomaly.
       if (_anomalyInProgress.anomalyType() == KafkaAnomalyType.BROKER_FAILURE) {
         synchronized (_shutdownLock) {
@@ -485,7 +486,17 @@ public class AnomalyDetectorManager {
           } else {
             LOG.debug("Scheduling broker failure detection with delay of {} ms", delayMs);
             _numCheckedWithDelay.incrementAndGet();
-            _detectorScheduler.schedule(() -> _brokerFailureDetector.detectBrokerFailures(false), delayMs, TimeUnit.MILLISECONDS);
+            // By default, the retry count is 0.
+            int retryCount = 0;
+            if (carryForwardRetryCount) {
+              // Only in cases of check with delay, we carry forward the retry count.
+              BrokerFailures brokerFailures = (BrokerFailures) _anomalyInProgress;
+              // Carry forward the count of anomaly fix checks done until now
+              retryCount = brokerFailures.brokerFailureCheckWithDelayRetryCount() + 1;
+            }
+            int finalRetryCount = retryCount;
+            _detectorScheduler.schedule(() -> _brokerFailureDetector.detectBrokerFailures(false,
+                finalRetryCount), delayMs, TimeUnit.MILLISECONDS);
             _anomalyDetectorState.onAnomalyHandle(_anomalyInProgress, AnomalyState.Status.CHECK_WITH_DELAY);
           }
         }
diff --git a/cruise-control/src/main/java/com/linkedin/kafka/cruisecontrol/detector/BrokerFailures.java b/cruise-control/src/main/java/com/linkedin/kafka/cruisecontrol/detector/BrokerFailures.java
index 262a0d2e..2df84c9c 100644
--- a/cruise-control/src/main/java/com/linkedin/kafka/cruisecontrol/detector/BrokerFailures.java
+++ b/cruise-control/src/main/java/com/linkedin/kafka/cruisecontrol/detector/BrokerFailures.java
@@ -29,6 +29,7 @@ public class BrokerFailures extends KafkaAnomaly {
   protected Map<Integer, Long> _failedBrokers;
   protected RemoveBrokersRunnable _removeBrokersRunnable;
   protected boolean _fixable;
+  protected int _brokerFailureCheckWithDelayRetryCount;
 
   /**
    * An anomaly to indicate broker failure(s).
@@ -54,6 +55,10 @@ public class BrokerFailures extends KafkaAnomaly {
     return _fixable;
   }
 
+  public int brokerFailureCheckWithDelayRetryCount() {
+    return _brokerFailureCheckWithDelayRetryCount;
+  }
+
   @Override
   public boolean fix() throws KafkaCruiseControlException {
     boolean hasProposalsToFix = false;
@@ -90,6 +95,19 @@ public class BrokerFailures extends KafkaAnomaly {
     return sb.toString();
   }
 
+  /**
+   * Configure the current retry count number for the broker failure check with delay.
+   * @param configs The configuration map.
+   */
+  protected void configureBrokerFailureCheckWithDelayRetryCount(Map<String, ?> configs) {
+    if (configs.containsKey(AbstractBrokerFailureDetector.BROKER_FAILURE_CHECK_WITH_DELAY_RETRY_COUNT)) {
+      _brokerFailureCheckWithDelayRetryCount = (int) configs.get(AbstractBrokerFailureDetector.BROKER_FAILURE_CHECK_WITH_DELAY_RETRY_COUNT);
+    } else {
+      // If unset we use the default value as 0.
+      _brokerFailureCheckWithDelayRetryCount = 0;
+    }
+  }
+
   @SuppressWarnings("unchecked")
   @Override
   public void configure(Map<String, ?> configs) {
@@ -100,6 +118,7 @@ public class BrokerFailures extends KafkaAnomaly {
       throw new IllegalArgumentException("Missing broker ids for failed brokers anomaly.");
     }
     _fixable = (Boolean) configs.get(AbstractBrokerFailureDetector.BROKER_FAILURES_FIXABLE_CONFIG);
+    configureBrokerFailureCheckWithDelayRetryCount(configs);
     _optimizationResult = null;
     KafkaCruiseControlConfig config = kafkaCruiseControl.config();
     boolean allowCapacityEstimation = config.getBoolean(ANOMALY_DETECTION_ALLOW_CAPACITY_ESTIMATION_CONFIG);
diff --git a/cruise-control/src/main/java/com/linkedin/kafka/cruisecontrol/detector/notifier/SelfHealingNotifier.java b/cruise-control/src/main/java/com/linkedin/kafka/cruisecontrol/detector/notifier/SelfHealingNotifier.java
index eff36923..00375bd2 100644
--- a/cruise-control/src/main/java/com/linkedin/kafka/cruisecontrol/detector/notifier/SelfHealingNotifier.java
+++ b/cruise-control/src/main/java/com/linkedin/kafka/cruisecontrol/detector/notifier/SelfHealingNotifier.java
@@ -66,8 +66,12 @@ public class SelfHealingNotifier implements AnomalyNotifier {
   public static final String SELF_HEALING_TOPIC_ANOMALY_ENABLED_CONFIG = "self.healing.topic.anomaly.enabled";
   public static final String SELF_HEALING_MAINTENANCE_EVENT_ENABLED_CONFIG = "self.healing.maintenance.event.enabled";
   public static final String BROKER_FAILURE_SELF_HEALING_THRESHOLD_MS_CONFIG = "broker.failure.self.healing.threshold.ms";
+  public static final String BROKER_FAILURE_CHECK_WITH_DELAY_MAX_RETRY_COUNT = "broker.failure.check.with.delay.max.retry.count";
+  public static final String BROKER_FAILURE_CHECK_WITH_DELAY_INTERVAL_MS = "broker.failure.check.with.delay.interval.ms";
   static final long DEFAULT_ALERT_THRESHOLD_MS = TimeUnit.MINUTES.toMillis(15);
   static final long DEFAULT_AUTO_FIX_THRESHOLD_MS = TimeUnit.MINUTES.toMillis(30);
+  static final int DEFAULT_BROKER_FAILURE_CHECK_WITH_DELAY_MAX_RETRY_COUNT = 10;
+  static final long DEFAULT_BROKER_FAILURE_CHECK_WITH_DELAY_INTERVAL_MS = TimeUnit.MINUTES.toMillis(10);
 
   private static final Logger LOG = LoggerFactory.getLogger(SelfHealingNotifier.class);
   protected final Time _time;
@@ -77,6 +81,8 @@ public class SelfHealingNotifier implements AnomalyNotifier {
   protected final Map<AnomalyType, Long> _selfHealingEnabledHistoricalDurationMs;
   protected long _brokerFailureAlertThresholdMs;
   protected long _selfHealingThresholdMs;
+  protected int _brokerFailureCheckWithDelayMaxRetryCount;
+  protected long _brokerFailureCheckWithDelayIntervalMs;
   // A cache that keeps the most recent broker failure for each broker.
   protected final Map<Boolean, Map<Integer, Long>> _latestFailedBrokersByAutoFixTriggered;
 
@@ -248,11 +254,36 @@ public class SelfHealingNotifier implements AnomalyNotifier {
       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);
+      boolean selfHealingEnabled = _selfHealingEnabled.get(KafkaAnomalyType.BROKER_FAILURE);
+      boolean brokerFailureFixable = brokerFailures.fixable();
+      boolean autoFixTriggered = selfHealingEnabled && brokerFailures.fixable();
+
+      if (!brokerFailureFixable) {
+        // If broker failure is not fixable then the anomaly can be ignored
+        result = AnomalyNotificationResult.ignore();
+      } else if (selfHealingEnabled) {
+        // If self healing is enabled and broker failure is fixable the fix should be made
+        if (hasNewFailureToAlert(brokerFailures, autoFixTriggered)) {
+          alert(brokerFailures, autoFixTriggered, selfHealingTimeMs, KafkaAnomalyType.BROKER_FAILURE);
+        }
+        result = AnomalyNotificationResult.fix();
+      } else {
+        // In the case self healing is disabled, we keep checking the anomaly until
+        // we try for _brokerFailureCheckWithDelayMaxRetryCount times
+        // After we exceed this, depending on the self healing state, we can ignore or fix the anomaly
+        // This check is to ensure that the broker failure is not ignored.
+        // The max is so that we do not keep checking forever in case self healing is disabled forever.
+        if (brokerFailures.brokerFailureCheckWithDelayRetryCount() <= _brokerFailureCheckWithDelayMaxRetryCount) {
+          // This means that we can retry for checking with delay
+          if (hasNewFailureToAlert(brokerFailures, autoFixTriggered)) {
+            alert(brokerFailures, autoFixTriggered, selfHealingTimeMs, KafkaAnomalyType.BROKER_FAILURE);
+          }
+          result = AnomalyNotificationResult.check(_brokerFailureCheckWithDelayIntervalMs);
+        } else {
+          // This means that we have reached the max retry count and we can ignore the anomaly
+          result = AnomalyNotificationResult.ignore();
+        }
       }
-      result = autoFixTriggered ? AnomalyNotificationResult.fix() : AnomalyNotificationResult.ignore();
     }
     return result;
   }
@@ -280,6 +311,15 @@ public class SelfHealingNotifier implements AnomalyNotifier {
     _brokerFailureAlertThresholdMs = alertThreshold == null ? DEFAULT_ALERT_THRESHOLD_MS : Long.parseLong(alertThreshold);
     String fixThreshold = (String) config.get(BROKER_FAILURE_SELF_HEALING_THRESHOLD_MS_CONFIG);
     _selfHealingThresholdMs = fixThreshold == null ? DEFAULT_AUTO_FIX_THRESHOLD_MS : Long.parseLong(fixThreshold);
+
+    String brokerFailureCheckWithDelayMaxRetryCount = (String) config.get(BROKER_FAILURE_CHECK_WITH_DELAY_MAX_RETRY_COUNT);
+    _brokerFailureCheckWithDelayMaxRetryCount = brokerFailureCheckWithDelayMaxRetryCount == null
+      ? DEFAULT_BROKER_FAILURE_CHECK_WITH_DELAY_MAX_RETRY_COUNT : Integer.parseInt(brokerFailureCheckWithDelayMaxRetryCount);
+
+    String brokerFailureCheckWithDelayIntervalMs = (String) config.get(BROKER_FAILURE_CHECK_WITH_DELAY_INTERVAL_MS);
+    _brokerFailureCheckWithDelayIntervalMs = brokerFailureCheckWithDelayIntervalMs == null
+      ? DEFAULT_BROKER_FAILURE_CHECK_WITH_DELAY_INTERVAL_MS : Long.parseLong(brokerFailureCheckWithDelayIntervalMs);
+
     if (_brokerFailureAlertThresholdMs > _selfHealingThresholdMs) {
       throw new IllegalArgumentException(String.format("The failure detection threshold %d cannot be larger than "
                                                        + "the auto fix threshold. %d",
diff --git a/docs/wiki/User Guide/Configurations.md b/docs/wiki/User Guide/Configurations.md
index 51192f20..957cd07a 100644
--- a/docs/wiki/User Guide/Configurations.md
+++ b/docs/wiki/User Guide/Configurations.md
@@ -362,17 +362,19 @@ with a non-default capacity. See:
 3. The [relevant configuration](https://github.com/linkedin/cruise-control/blob/migrate_to_kafka_2_4/cruise-control/src/main/java/com/linkedin/kafka/cruisecontrol/config/constants/MonitorConfig.java#L294) to set the config resolver to be used.
 
 ### SelfHealingNotifier configurations
-| Name                                     | Type    | Required? | Default Value            | Description                                                                                                                                                                                                                                                          |
-|------------------------------------------|---------|-----------|--------------------------|----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|
-| broker.failure.alert.threshold.ms        | Long    | N         | 900,000                  | Defines the threshold to mark a broker as dead. If a non-empty broker leaves the cluster at time T and did not join the cluster before T + broker.failure.alert.threshold.ms, the broker is defined as dead broker since T. An alert will be triggered in this case. |
-| broker.failure.self.healing.threshold.ms | Long    | N         | 1,800,000                | If self-healing is enabled and a broker is dead at T,,self-healing will be triggered at T + broker.failure.self.healing.threshold.ms.                                                                                                                                |
-| self.healing.enabled                     | Boolean | N         | false                    | Whether to enable Enable self healing for all anomaly detectors, unless the particular anomaly detector is explicitly disabled.                                                                                                                                      |
-| self.healing.broker.failure.enabled      | Boolean | N         | <self.healing.enabled>   | Whether enable self-healing for detected broker failure or not. If disabled, the SelfHealingNotifier will only log the anomaly, but not take a fix action.                                                                                                           |
-| self.healing.goal.violation.enabled      | Boolean | N         | <self.healing.enabled>   | Whether enable self-healing for detected goal violation or not. If disabled, the SelfHealingNotifier will only log the anomaly, but not take a fix action.                                                                                                           |
-| self.healing.metric.anomaly.enabled      | Boolean | N         | <self.healing.enabled>   | Whether enable self-healing for detected metric anomaly or not. If disabled, the SelfHealingNotifier will only log the anomaly, but not take a fix action.                                                                                                           |
-| self.healing.disk.failure.enabled        | Boolean | N         | <self.healing.enabled>   | Whether enable self-healing for detected disk failure or not. If disabled, the SelfHealingNotifier will only log the anomaly, but not take a fix action.                                                                                                             |
-| self.healing.topic.anomaly.enabled       | Boolean | N         | <self.healing.enabled>   | Whether enable self-healing for detected topic anomaly or not. If disabled, the SelfHealingNotifier will only log the anomaly, but not take a fix action.                                                                                                            |
-| self.healing.maintenance.event.enabled   | Boolean | N         | <self.healing.enabled>   | Whether enable self-healing for detected maintenance event or not. If disabled, the SelfHealingNotifier will only log the anomaly, but not take a fix action.                                                                                                        |
+| Name                                            | Type    | Required? | Default Value          | Description                                                                                                                                                                                                                                                          |
+|-------------------------------------------------|---------|-----------|------------------------|----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|
+| broker.failure.alert.threshold.ms               | Long    | N         | 900,000                | Defines the threshold to mark a broker as dead. If a non-empty broker leaves the cluster at time T and did not join the cluster before T + broker.failure.alert.threshold.ms, the broker is defined as dead broker since T. An alert will be triggered in this case. |
+| broker.failure.self.healing.threshold.ms        | Long    | N         | 1,800,000              | If self-healing is enabled and a broker is dead at T,,self-healing will be triggered at T + broker.failure.self.healing.threshold.ms.                                                                                                                                |
+| broker.failure.check.with.delay.max.retry.count | Int     | N         | 10                     | Specifies the maximum number of times to recheck a broker failure anomaly detected when self-healing is disabled, before giving up on further checks.                                                                                                                |
+| broker.failure.check.with.delay.interval.ms     | Long    | N         | 600,000                | Specifies the delay interval (in milliseconds) between consecutive broker failure anomaly rechecks when self-healing is disabled.                                                                                                                                    |
+| self.healing.enabled                            | Boolean | N         | false                  | Whether to enable Enable self healing for all anomaly detectors, unless the particular anomaly detector is explicitly disabled.                                                                                                                                      |
+| self.healing.broker.failure.enabled             | Boolean | N         | <self.healing.enabled> | Whether enable self-healing for detected broker failure or not. If disabled, the SelfHealingNotifier will only log the anomaly, but not take a fix action.                                                                                                           |
+| self.healing.goal.violation.enabled             | Boolean | N         | <self.healing.enabled> | Whether enable self-healing for detected goal violation or not. If disabled, the SelfHealingNotifier will only log the anomaly, but not take a fix action.                                                                                                           |
+| self.healing.metric.anomaly.enabled             | Boolean | N         | <self.healing.enabled> | Whether enable self-healing for detected metric anomaly or not. If disabled, the SelfHealingNotifier will only log the anomaly, but not take a fix action.                                                                                                           |
+| self.healing.disk.failure.enabled               | Boolean | N         | <self.healing.enabled> | Whether enable self-healing for detected disk failure or not. If disabled, the SelfHealingNotifier will only log the anomaly, but not take a fix action.                                                                                                             |
+| self.healing.topic.anomaly.enabled              | Boolean | N         | <self.healing.enabled> | Whether enable self-healing for detected topic anomaly or not. If disabled, the SelfHealingNotifier will only log the anomaly, but not take a fix action.                                                                                                            |
+| self.healing.maintenance.event.enabled          | Boolean | N         | <self.healing.enabled> | Whether enable self-healing for detected maintenance event or not. If disabled, the SelfHealingNotifier will only log the anomaly, but not take a fix action.                                                                                                        |
 
 ### CruiseControlMetricsReporter configurations
 | Name                                                          | Type      | Required? | Default Value            | Description                                                                                                                                                                                                                                            |


Test patch:
diff --git a/cruise-control/src/test/java/com/linkedin/kafka/cruisecontrol/detector/notifier/SelfHealingNotifierTest.java b/cruise-control/src/test/java/com/linkedin/kafka/cruisecontrol/detector/notifier/SelfHealingNotifierTest.java
index 00eea130..78509898 100644
--- a/cruise-control/src/test/java/com/linkedin/kafka/cruisecontrol/detector/notifier/SelfHealingNotifierTest.java
+++ b/cruise-control/src/test/java/com/linkedin/kafka/cruisecontrol/detector/notifier/SelfHealingNotifierTest.java
@@ -15,6 +15,7 @@ import com.linkedin.kafka.cruisecontrol.detector.DiskFailures;
 import com.linkedin.kafka.cruisecontrol.detector.GoalViolations;
 import com.linkedin.kafka.cruisecontrol.detector.KafkaMetricAnomaly;
 import com.linkedin.kafka.cruisecontrol.detector.TopicReplicationFactorAnomaly;
+import com.linkedin.kafka.cruisecontrol.detector.AbstractBrokerFailureDetector;
 import com.linkedin.kafka.cruisecontrol.monitor.sampling.holder.BrokerEntity;
 import java.util.Collections;
 import java.util.HashMap;
@@ -177,11 +178,43 @@ public class SelfHealingNotifierTest {
         kafkaCruiseControlConfig.getConfiguredInstance(AnomalyDetectorConfig.BROKER_FAILURES_CLASS_CONFIG,
                                                        BrokerFailures.class,
                                                        parameterConfigOverrides));
-    assertEquals(AnomalyNotificationResult.Action.IGNORE, result.action());
+    assertEquals(AnomalyNotificationResult.Action.CHECK, result.action());
     assertTrue(anomalyNotifier.isAlertCalledFor(KafkaAnomalyType.BROKER_FAILURE));
     assertFalse(anomalyNotifier.isAutoFixTriggeredFor(KafkaAnomalyType.BROKER_FAILURE));
 
+    // (2) Test that if retry count exceeds the max retry count, the anomaly is ignored when self healing is disabled.
     // (2) Test goal violation anomaly can be detected by notifier.
+    anomalyNotifier.resetAlert(KafkaAnomalyType.BROKER_FAILURE);
+    Map<String, Object> parameterConfigOverrides2 = new HashMap<>(parameterConfigOverrides);
+    // Set the current retry count to be greater than DEFAULT_BROKER_FAILURE_CHECK_WITH_DELAY_MAX_RETRY_COUNT
+    parameterConfigOverrides2.put(AbstractBrokerFailureDetector.BROKER_FAILURE_CHECK_WITH_DELAY_RETRY_COUNT,
+        SelfHealingNotifier.DEFAULT_BROKER_FAILURE_CHECK_WITH_DELAY_MAX_RETRY_COUNT + 1);
+    result = anomalyNotifier.onBrokerFailure(kafkaCruiseControlConfig.getConfiguredInstance(AnomalyDetectorConfig.BROKER_FAILURES_CLASS_CONFIG,
+        BrokerFailures.class, parameterConfigOverrides2));
+    // The result should be ignore
+    assertEquals(AnomalyNotificationResult.Action.IGNORE, result.action());
+    assertFalse(anomalyNotifier.isAutoFixTriggeredFor(KafkaAnomalyType.BROKER_FAILURE));
+
+    // (3) Self healing is enabled back when the check with delay is executed but the retry count is less than max
+    // In this case it should be able to trigger a fix
+    TestingBrokerFailureAutoFixNotifier selfHealingEnabledAnomalyNotifier = new TestingBrokerFailureAutoFixNotifier(mockTime);
+    // Set the self healing enabled
+    Map<String, String> selfHealingExplicitlyEnabled = Map.of(SelfHealingNotifier.SELF_HEALING_BROKER_FAILURE_ENABLED_CONFIG, "true",
+        // Set to verify the overriding of specific config over general config
+        SelfHealingNotifier.SELF_HEALING_ENABLED_CONFIG, "true");
+    selfHealingEnabledAnomalyNotifier.configure(selfHealingExplicitlyEnabled);
+    selfHealingEnabledAnomalyNotifier.resetAlert(KafkaAnomalyType.BROKER_FAILURE);
+    Map<String, Object> parameterConfigOverrides3 = new HashMap<>(parameterConfigOverrides);
+    parameterConfigOverrides3.put(AbstractBrokerFailureDetector.BROKER_FAILURE_CHECK_WITH_DELAY_RETRY_COUNT,
+        SelfHealingNotifier.DEFAULT_BROKER_FAILURE_CHECK_WITH_DELAY_MAX_RETRY_COUNT - 1);
+    result = selfHealingEnabledAnomalyNotifier.onBrokerFailure(
+        kafkaCruiseControlConfig.getConfiguredInstance(AnomalyDetectorConfig.BROKER_FAILURES_CLASS_CONFIG,
+            BrokerFailures.class,
+            parameterConfigOverrides3));
+    assertEquals(AnomalyNotificationResult.Action.FIX, result.action());
+    assertTrue(selfHealingEnabledAnomalyNotifier.isAutoFixTriggeredFor(KafkaAnomalyType.BROKER_FAILURE));
+
+    // (4) Test goal violation anomaly can be detected by notifier.
     anomalyNotifier.resetAlert(KafkaAnomalyType.GOAL_VIOLATION);
     result = anomalyNotifier.onGoalViolation(
         kafkaCruiseControlConfig.getConfiguredInstance(AnomalyDetectorConfig.GOAL_VIOLATIONS_CLASS_CONFIG,
@@ -191,7 +224,7 @@ public class SelfHealingNotifierTest {
     assertTrue(anomalyNotifier.isAlertCalledFor(KafkaAnomalyType.GOAL_VIOLATION));
     assertFalse(anomalyNotifier.isAutoFixTriggeredFor(KafkaAnomalyType.GOAL_VIOLATION));
 
-    // (3) Test metric anomaly can be detected by notifier.
+    // (5) Test metric anomaly can be detected by notifier.
     anomalyNotifier.resetAlert(KafkaAnomalyType.METRIC_ANOMALY);
     result = anomalyNotifier.onMetricAnomaly(kafkaCruiseControlConfig.getConfiguredInstance(AnomalyDetectorConfig.METRIC_ANOMALY_CLASS_CONFIG,
                                                                                             KafkaMetricAnomaly.class,
@@ -200,7 +233,7 @@ public class SelfHealingNotifierTest {
     assertTrue(anomalyNotifier.isAlertCalledFor(KafkaAnomalyType.METRIC_ANOMALY));
     assertFalse(anomalyNotifier.isAutoFixTriggeredFor(KafkaAnomalyType.METRIC_ANOMALY));
 
-    // (4) Test disk failure anomaly can be detected by notifier.
+    // (6) Test disk failure anomaly can be detected by notifier.
     anomalyNotifier.resetAlert(KafkaAnomalyType.DISK_FAILURE);
     result = anomalyNotifier.onDiskFailure(kafkaCruiseControlConfig.getConfiguredInstance(AnomalyDetectorConfig.DISK_FAILURES_CLASS_CONFIG,
                                                                                           DiskFailures.class,
@@ -209,7 +242,7 @@ public class SelfHealingNotifierTest {
     assertTrue(anomalyNotifier.isAlertCalledFor(KafkaAnomalyType.DISK_FAILURE));
     assertFalse(anomalyNotifier.isAutoFixTriggeredFor(KafkaAnomalyType.DISK_FAILURE));
 
-    // (5) Test topic anomaly can be detected by notifier.
+    // (7) Test topic anomaly can be detected by notifier.
     anomalyNotifier.resetAlert(KafkaAnomalyType.TOPIC_ANOMALY);
     TopicReplicationFactorAnomaly topicReplicationFactorAnomaly = new TopicReplicationFactorAnomaly();
     topicReplicationFactorAnomaly.configure(parameterConfigOverrides);