Fix ignored broker failure anomalies when self healing is disabled (#2270)

bsandeep23 · web-flow · commit 70e51ecebab5 · 2025-04-28T09:32:09.000-07:00
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
@@ -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 @@ synchronized void detectBrokerFailures(boolean skipReportingIfNotUpdated) {
       }
       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 @@ private boolean tooManyFailedBrokers(int failedBrokerCount, int aliveBrokerCount
            || (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 @@ private void reportBrokerFailures() {
       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
@@ -350,7 +350,7 @@ public void run() {
         _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 void run() {
           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 @@ private void handleAnomalyInProgress() throws Exception {
       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 @@ private void processAnomalyInProgress(AnomalyType anomalyType) throws Exception
             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 @@ private AnomalyNotificationResult notifyAnomalyInProgress(AnomalyType anomalyTyp
      * 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 @@ private void postProcessAnomalyInProgress(long delayMs) {
           } 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
@@ -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 boolean fixable() {
     return _fixable;
   }
 
+  public int brokerFailureCheckWithDelayRetryCount() {
+    return _brokerFailureCheckWithDelayRetryCount;
+  }
+
   @Override
   public boolean fix() throws KafkaCruiseControlException {
     boolean hasProposalsToFix = false;
@@ -90,6 +95,19 @@ public String toString() {
     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 void configure(Map<String, ?> configs) {
       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
@@ -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 AnomalyNotificationResult onBrokerFailure(BrokerFailures brokerFailures)
       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 void configure(Map<String, ?> config) {
     _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/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
diff --git a/docs/wiki/User Guide/Configurations.md b/docs/wiki/User Guide/Configurations.md
