Add percentage-based offline instance thresholds for maintenance mode

helix-core/src/main/java/org/apache/helix/controller/stages/BestPossibleStateCalcStage.java

@@ -358,41 +358,54 @@ public Object call() {
   // if yes, auto enable maintenance mode, and use the maintenance rebalancer for this pipeline.
   private boolean validateInstancesUnableToAcceptOnlineReplicasLimit(final ResourceControllerDataProvider cache,
       final HelixManager manager) {
-    int maxInstancesUnableToAcceptOnlineReplicas =
-        cache.getClusterConfig().getMaxOfflineInstancesAllowed();
-    if (maxInstancesUnableToAcceptOnlineReplicas >= 0) {
-      // Instead of only checking the offline instances, we consider how many instances in the cluster
-      // are not assignable and live. This is because some instances may be online but have an unassignable
-      // InstanceOperation such as EVACUATE, and DISABLE. We will exclude SWAP_IN and UNKNOWN instances from
-      // they should not account against the capacity of the cluster.
-      int instancesUnableToAcceptOnlineReplicas = cache.getInstanceConfigMap().entrySet().stream()
-          .filter(instanceEntry -> !InstanceConstants.UNROUTABLE_INSTANCE_OPERATIONS.contains(
-              instanceEntry.getValue().getInstanceOperation().getOperation()))
-          .collect(Collectors.toSet())
-          .size() - cache.getEnabledLiveInstances().size();
-      if (instancesUnableToAcceptOnlineReplicas > maxInstancesUnableToAcceptOnlineReplicas) {
-        String errMsg = String.format(
-            "Instances unable to take ONLINE replicas count %d greater than allowed count %d. Put cluster %s into "
-                + "maintenance mode.", instancesUnableToAcceptOnlineReplicas,
-            maxInstancesUnableToAcceptOnlineReplicas, cache.getClusterName());
-        if (manager != null) {
-          if (manager.getHelixDataAccessor()
-              .getProperty(manager.getHelixDataAccessor().keyBuilder().maintenance()) == null) {
-            manager.getClusterManagmentTool()
-                .autoEnableMaintenanceMode(manager.getClusterName(), true, errMsg,
-                    MaintenanceSignal.AutoTriggerReason.MAX_INSTANCES_UNABLE_TO_ACCEPT_ONLINE_REPLICAS);
-            LogUtil.logWarn(logger, _eventId, errMsg);
-          }
-        } else {
-          LogUtil.logError(logger, _eventId, "Failed to put cluster " + cache.getClusterName()
-              + " into maintenance mode, HelixManager is not set!");
-        }
+    ClusterConfig clusterConfig = cache.getClusterConfig();
+    int absoluteThreshold = clusterConfig.getMaxOfflineInstancesAllowed();
+    int percentageThreshold = clusterConfig.getMaxOfflineInstancesAllowedPercentage();
 
-        // Enable maintenance mode in cache so the maintenance rebalancer is used for this pipeline
-        cache.enableMaintenanceMode();
+    // Early exit if neither threshold is configured
+    if (absoluteThreshold < 0 && percentageThreshold < 0) {
+      return true;
+    }
 
-        return false;
+    // Instead of only checking the offline instances, we consider how many instances in the cluster
+    // are not assignable and live. This is because some instances may be online but have an unassignable
+    // InstanceOperation such as EVACUATE, and DISABLE. We will exclude SWAP_IN and UNKNOWN instances
+    // as they should not account against the capacity of the cluster.
+    int routableInstanceCount = (int) cache.getInstanceConfigMap().entrySet().stream()
+        .filter(instanceEntry -> !InstanceConstants.UNROUTABLE_INSTANCE_OPERATIONS.contains(
+            instanceEntry.getValue().getInstanceOperation().getOperation()))
+        .count();
+    int instancesUnableToAcceptOnlineReplicas =
+        routableInstanceCount - cache.getEnabledLiveInstances().size();
+
+    int effectiveThreshold = ClusterConfig.resolveEffectiveThreshold(
+        absoluteThreshold, percentageThreshold, routableInstanceCount);
+
+    if (effectiveThreshold >= 0
+        && instancesUnableToAcceptOnlineReplicas > effectiveThreshold) {
+      String errMsg = String.format(
+          "Instances unable to take ONLINE replicas count %d greater than effective allowed count %d "
+              + "(absolute=%d, percentage=%d%% of %d routable). Put cluster %s into maintenance mode.",
+          instancesUnableToAcceptOnlineReplicas, effectiveThreshold,
+          absoluteThreshold, percentageThreshold, routableInstanceCount,
+          cache.getClusterName());
+      if (manager != null) {
+        if (manager.getHelixDataAccessor()
+            .getProperty(manager.getHelixDataAccessor().keyBuilder().maintenance()) == null) {
+          manager.getClusterManagmentTool()
+              .autoEnableMaintenanceMode(manager.getClusterName(), true, errMsg,
+                  MaintenanceSignal.AutoTriggerReason.MAX_INSTANCES_UNABLE_TO_ACCEPT_ONLINE_REPLICAS);
+          LogUtil.logWarn(logger, _eventId, errMsg);
+        }
+      } else {
+        LogUtil.logError(logger, _eventId, "Failed to put cluster " + cache.getClusterName()
+            + " into maintenance mode, HelixManager is not set!");
       }
+
+      // Enable maintenance mode in cache so the maintenance rebalancer is used for this pipeline
+      cache.enableMaintenanceMode();
+
+      return false;
     }
     return true;
   }

helix-core/src/main/java/org/apache/helix/controller/stages/MaintenanceRecoveryStage.java

@@ -24,12 +24,14 @@
 
 import org.apache.helix.HelixDefinedState;
 import org.apache.helix.HelixManager;
+import org.apache.helix.constants.InstanceConstants;
 import org.apache.helix.controller.LogUtil;
 import org.apache.helix.controller.common.PartitionStateMap;
 import org.apache.helix.controller.dataproviders.ResourceControllerDataProvider;
 import org.apache.helix.controller.pipeline.AbstractAsyncBaseStage;
 import org.apache.helix.controller.pipeline.AsyncWorkerType;
 import org.apache.helix.model.BuiltInStateModelDefinitions;
+import org.apache.helix.model.ClusterConfig;
 import org.apache.helix.model.IdealState;
 import org.apache.helix.model.MaintenanceSignal;
 import org.apache.helix.model.Partition;
@@ -84,18 +86,34 @@ public void execute(final ClusterEvent event) throws Exception {
     case MAX_OFFLINE_INSTANCES_EXCEEDED:
     case MAX_INSTANCES_UNABLE_TO_ACCEPT_ONLINE_REPLICAS:
       // Check on the number of offline/disabled instances
-      int numOfflineInstancesForAutoExit =
-          cache.getClusterConfig().getNumOfflineInstancesForAutoExit();
-      if (numOfflineInstancesForAutoExit < 0) {
-        return; // Config is not set, no auto-exit
+      ClusterConfig clusterConfig = cache.getClusterConfig();
+      int absoluteExitThreshold = clusterConfig.getNumOfflineInstancesForAutoExit();
+      int percentageExitThreshold = clusterConfig.getNumOfflineInstancesForAutoExitPercentage();
+
+      if (absoluteExitThreshold < 0 && percentageExitThreshold < 0) {
+        return; // Neither config is set, no auto-exit
       }
+
+      // Compute routable instance count for percentage resolution (same filter as entry logic)
+      int routableInstanceCount = (int) cache.getInstanceConfigMap().entrySet().stream()
+          .filter(instanceEntry -> !InstanceConstants.UNROUTABLE_INSTANCE_OPERATIONS.contains(
+              instanceEntry.getValue().getInstanceOperation().getOperation()))
+          .count();
+
       // Get the count of all instances that are either offline or disabled
       int offlineDisabledCount =
           cache.getAssignableInstances().size() - cache.getEnabledLiveInstances().size();
-      shouldExitMaintenance = offlineDisabledCount <= numOfflineInstancesForAutoExit;
+
+      int effectiveExitThreshold = ClusterConfig.resolveEffectiveThreshold(
+          absoluteExitThreshold, percentageExitThreshold, routableInstanceCount);
+
+      shouldExitMaintenance =
+          effectiveExitThreshold >= 0 && offlineDisabledCount <= effectiveExitThreshold;
       reason = String.format(
-          "Auto-exiting maintenance mode for cluster %s; Num. of offline/disabled instances is %d, less than or equal to the exit threshold %d",
-          event.getClusterName(), offlineDisabledCount, numOfflineInstancesForAutoExit);
+          "Auto-exiting maintenance mode for cluster %s; Num. of offline/disabled instances is %d, "
+              + "less than or equal to effective exit threshold %d (absolute=%d, percentage=%d%% of %d routable)",
+          event.getClusterName(), offlineDisabledCount, effectiveExitThreshold,
+          absoluteExitThreshold, percentageExitThreshold, routableInstanceCount);
       break;
     case MAX_PARTITION_PER_INSTANCE_EXCEEDED:
       IntermediateStateOutput intermediateStateOutput =

helix-core/src/main/java/org/apache/helix/model/ClusterConfig.java

@@ -79,6 +79,10 @@ public enum ClusterConfigProperty {
     //     to make it clear that it includes both offline and non-assignable instances
     MAX_OFFLINE_INSTANCES_ALLOWED,
     NUM_OFFLINE_INSTANCES_FOR_AUTO_EXIT, // For auto-exiting maintenance mode
+    // Percentage-based alternatives for maintenance mode thresholds (0-100).
+    // When both absolute and percentage are set, the stricter (lower effective count) wins.
+    MAX_OFFLINE_INSTANCES_ALLOWED_PERCENTAGE,
+    NUM_OFFLINE_INSTANCES_FOR_AUTO_EXIT_PERCENTAGE,
 
     TARGET_EXTERNALVIEW_ENABLED,
     @Deprecated // ERROR_OR_RECOVERY_PARTITION_THRESHOLD_FOR_LOAD_BALANCE will take
@@ -592,6 +596,114 @@ public int getNumOfflineInstancesForAutoExit() {
         .getIntField(ClusterConfigProperty.NUM_OFFLINE_INSTANCES_FOR_AUTO_EXIT.name(), -1);
   }
 
+  /**
+   * Set the max offline instances allowed as a percentage (0-100) of total routable instances.
+   * When both percentage and absolute thresholds are set, the stricter (lower effective count) wins.
+   * -1 disables the percentage-based entry threshold.
+   * @param maxOfflineInstancesAllowedPercentage percentage threshold (0-100) or -1 to disable
+   */
+  public void setMaxOfflineInstancesAllowedPercentage(int maxOfflineInstancesAllowedPercentage) {
+    if (maxOfflineInstancesAllowedPercentage < -1 || maxOfflineInstancesAllowedPercentage > 100) {
+      throw new HelixException(
+          "Max offline instances allowed percentage must be between 0 and 100, or -1 to disable. Got: "
+              + maxOfflineInstancesAllowedPercentage);
+    }
+    int exitPercentage = getNumOfflineInstancesForAutoExitPercentage();
+    if (exitPercentage >= 0 && maxOfflineInstancesAllowedPercentage >= 0) {
+      if (maxOfflineInstancesAllowedPercentage < exitPercentage) {
+        throw new HelixException(
+            "Entry percentage threshold must be greater than or equal to exit percentage threshold! "
+                + "Entry: " + maxOfflineInstancesAllowedPercentage + ", Exit: " + exitPercentage);
+      }
+    }
+    _record.setIntField(
+        ClusterConfigProperty.MAX_OFFLINE_INSTANCES_ALLOWED_PERCENTAGE.name(),
+        maxOfflineInstancesAllowedPercentage);
+  }
+
+  /**
+   * Get the max offline instances allowed percentage for the cluster.
+   * @return percentage (0-100) or -1 if not set
+   */
+  public int getMaxOfflineInstancesAllowedPercentage() {
+    return _record.getIntField(
+        ClusterConfigProperty.MAX_OFFLINE_INSTANCES_ALLOWED_PERCENTAGE.name(), -1);
+  }
+
+  /**
+   * Sets the percentage-based offline instances threshold for auto-exit (0-100).
+   * The percentage is computed against total routable instances at runtime.
+   * When both percentage and absolute exit thresholds are set, the stricter (lower) wins.
+   * If a percentage-based entry threshold is also set, exit percentage must be <= entry percentage.
+   * -1 disables the percentage-based auto-exit threshold.
+   * @param autoExitPercentage percentage threshold (0-100) or -1 to disable
+   */
+  public void setNumOfflineInstancesForAutoExitPercentage(int autoExitPercentage)
+      throws HelixException {
+    if (autoExitPercentage < -1 || autoExitPercentage > 100) {
+      throw new HelixException(
+          "Num offline instances for auto exit percentage must be between 0 and 100, or -1 to disable. Got: "
+              + autoExitPercentage);
+    }
+    int entryPercentage = getMaxOfflineInstancesAllowedPercentage();
+    if (entryPercentage >= 0 && autoExitPercentage >= 0) {
+      if (autoExitPercentage > entryPercentage) {
+        throw new HelixException(
+            "Auto-exit percentage threshold must be less than or equal to entry percentage threshold! "
+                + "Exit: " + autoExitPercentage + ", Entry: " + entryPercentage);
+      }
+    }
+    _record.setIntField(
+        ClusterConfigProperty.NUM_OFFLINE_INSTANCES_FOR_AUTO_EXIT_PERCENTAGE.name(),
+        autoExitPercentage);
+  }
+
+  /**
+   * Returns the percentage-based offline instances threshold for auto-exit.
+   * @return percentage (0-100) or -1 if not set
+   */
+  public int getNumOfflineInstancesForAutoExitPercentage() {
+    return _record.getIntField(
+        ClusterConfigProperty.NUM_OFFLINE_INSTANCES_FOR_AUTO_EXIT_PERCENTAGE.name(), -1);
+  }
+
+  /**
+   * Resolves the effective threshold given an absolute threshold, a percentage threshold,
+   * and a total instance count. The stricter (lower non-negative) value wins.
+   * <ul>
+   *   <li>If both are less than 0 (disabled), returns -1.</li>
+   *   <li>If only one is set (>= 0), that value is used (percentage is converted to count).</li>
+   *   <li>If both are set, the minimum of the two effective counts is returned.</li>
+   * </ul>
+   * Percentage conversion uses integer division (truncation toward zero), which is conservative
+   * for both entry (triggers sooner) and exit (requires more recovery).
+   *
+   * @param absoluteThreshold the absolute count threshold (-1 if not set)
+   * @param percentageThreshold the percentage threshold (0-100, -1 if not set)
+   * @param totalRoutableCount the total routable instance count to compute percentage against
+   * @return the effective threshold count, or -1 if neither is set
+   */
+  public static int resolveEffectiveThreshold(int absoluteThreshold, int percentageThreshold,
+      int totalRoutableCount) {
+    if (absoluteThreshold < 0 && percentageThreshold < 0) {
+      return -1;
+    }
+
+    int effectivePercentage = -1;
+    if (percentageThreshold >= 0) {
+      effectivePercentage =
+          (totalRoutableCount > 0) ? (int) ((long) totalRoutableCount * percentageThreshold / 100) : 0;
+    }
+
+    if (absoluteThreshold < 0) {
+      return effectivePercentage;
+    }
+    if (effectivePercentage < 0) {
+      return absoluteThreshold;
+    }
+    return Math.min(absoluteThreshold, effectivePercentage);
+  }
+
   /**
    * Set the resource prioritization field. It should be Integer field and sortable.
    * IMPORTANT: The sorting order is DESCENDING order, which means the larger number will have

helix-core/src/test/java/org/apache/helix/integration/controller/TestClusterMaintenanceMode.java

@@ -430,6 +430,86 @@ public void testMaintenanceHistory() throws Exception {
     Assert.assertNull(lastHistoryEntry.get("AUTO_TRIGGER_REASON"));
   }
 
+  /**
+   * Test that auto-exit works with percentage-based threshold.
+   * With 3 total instances and exit percentage of 33%, the effective exit threshold is
+   * 3 * 33 / 100 = 0 (integer truncation). So the cluster should auto-exit only when
+   * all instances are back online (0 offline).
+   */
+  @Test(dependsOnMethods = "testMaintenanceHistory")
+  public void testAutoExitMaintenanceModeWithPercentage() throws Exception {
+    // First, exit any existing maintenance mode
+    _gSetupTool.getClusterManagementTool().manuallyEnableMaintenanceMode(CLUSTER_NAME, false, null,
+        null);
+    TestHelper.verify(() -> _dataAccessor.getProperty(_keyBuilder.maintenance()) == null, 2000L);
+
+    // Stop the extra instance added in testMaintenanceModeAddNewInstance so we have a
+    // predictable instance count (_numNodes) for percentage calculations
+    if (_newInstance != null && _newInstance.isConnected()) {
+      _newInstance.syncStop();
+    }
+
+    // Bring all original instances back up
+    for (int i = 0; i < _numNodes; i++) {
+      if (!_participants[i].isConnected()) {
+        String instanceName = PARTICIPANT_PREFIX + "_" + (_startPort + i);
+        _participants[i] = new MockParticipantManager(ZK_ADDR, CLUSTER_NAME, instanceName);
+        _participants[i].syncStart();
+      }
+    }
+    // Total routable instances = _numNodes (3) + 1 (_newInstance, offline but still registered).
+    // _newInstance is stopped but its InstanceConfig still exists in the cluster, so it counts
+    // as a routable instance. Total routable = _numNodes + 1 = 4.
+    int totalRegisteredRoutable = _numNodes + 1;
+
+    // Set percentage-based exit config.
+    // Use absolute entry threshold of 1 for reliable entry, and percentage-based exit.
+    // 24% of 4 routable = 0 (integer truncation). So exit only when 0 offline/disabled.
+    ClusterConfig clusterConfig = _manager.getConfigAccessor().getClusterConfig(CLUSTER_NAME);
+    clusterConfig.setMaxOfflineInstancesAllowed(1);
+    clusterConfig.setNumOfflineInstancesForAutoExit(-1); // Disable absolute exit
+    clusterConfig.setNumOfflineInstancesForAutoExitPercentage(24);
+    _manager.getConfigAccessor().setClusterConfig(CLUSTER_NAME, clusterConfig);
+
+    // Kill 2 instances to trigger auto-enter (2 > 1)
+    for (int i = 0; i < 2; i++) {
+      _participants[i].syncStop();
+    }
+    TestHelper.verify(
+        () -> _dataAccessor.getProperty(_keyBuilder.maintenance()) != null, TIMEOUT);
+
+    // Bring up 1 instance (1 original still offline + _newInstance offline = 2 offline).
+    // Effective exit threshold = 24% of 4 = 0. 2 > 0, so should NOT auto-exit.
+    String instanceName = PARTICIPANT_PREFIX + "_" + (_startPort + 0);
+    _participants[0] = new MockParticipantManager(ZK_ADDR, CLUSTER_NAME, instanceName);
+    _participants[0].syncStart();
+    // Give some time for the pipeline to run and verify maintenance is NOT exited
+    Thread.sleep(2000);
+    MaintenanceSignal maintenanceSignal = _dataAccessor.getProperty(_keyBuilder.maintenance());
+    Assert.assertNotNull(maintenanceSignal, "Cluster should still be in maintenance");
+
+    // Bring up the last original instance AND _newInstance so all are online (0 offline).
+    // 0 <= 0, so should auto-exit.
+    instanceName = PARTICIPANT_PREFIX + "_" + (_startPort + 1);
+    _participants[1] = new MockParticipantManager(ZK_ADDR, CLUSTER_NAME, instanceName);
+    _participants[1].syncStart();
+    _newInstance =
+        new MockParticipantManager(ZK_ADDR, CLUSTER_NAME, _newInstance.getInstanceName());
+    _newInstance.syncStart();
+    TestHelper.verify(
+        () -> _dataAccessor.getChildNames(_keyBuilder.liveInstances()).size() == totalRegisteredRoutable,
+        2000L);
+
+    // Verify cluster auto-exited maintenance
+    TestHelper.verify(() -> _dataAccessor.getProperty(_keyBuilder.maintenance()) == null, TIMEOUT);
+
+    // Clean up: reset configs
+    clusterConfig = _manager.getConfigAccessor().getClusterConfig(CLUSTER_NAME);
+    clusterConfig.setMaxOfflineInstancesAllowed(-1);
+    clusterConfig.setNumOfflineInstancesForAutoExitPercentage(-1);
+    _manager.getConfigAccessor().setClusterConfig(CLUSTER_NAME, clusterConfig);
+  }
+
   /**
    * Convert a String representation of a Map into a Map object for verification purposes.
    * @param value