Add Helix REST guard rails for known rebalance-failure scenarios (CICP-3788)

helix-core/src/main/java/org/apache/helix/util/FeasibilityResult.java

@@ -0,0 +1,85 @@
+package org.apache.helix.util;
+
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+
+/**
+ * Outcome of a {@link RebalanceFeasibilityEvaluator} check: whether a proposed cluster mutation
+ * keeps the cluster rebalanceable, plus the list of {@link FeasibilityViolation}s if not.
+ */
+public class FeasibilityResult {
+  private final List<FeasibilityViolation> violations;
+
+  private FeasibilityResult(List<FeasibilityViolation> violations) {
+    this.violations = violations;
+  }
+
+  /**
+   * @return a feasible result with no violations.
+   */
+  public static FeasibilityResult feasible() {
+    return new FeasibilityResult(Collections.emptyList());
+  }
+
+  /**
+   * @param violations the violations found; must be non-empty for an infeasible result.
+   * @return a result wrapping the given violations.
+   */
+  public static FeasibilityResult of(List<FeasibilityViolation> violations) {
+    return new FeasibilityResult(new ArrayList<>(violations));
+  }
+
+  /**
+   * Combines several check results into one, unioning their violations. The merged result is
+   * feasible only if every input result is feasible.
+   *
+   * @param results the per-check results to combine
+   * @return a single aggregated result
+   */
+  public static FeasibilityResult merge(List<FeasibilityResult> results) {
+    List<FeasibilityViolation> all = new ArrayList<>();
+    for (FeasibilityResult result : results) {
+      all.addAll(result.getViolations());
+    }
+    return all.isEmpty() ? feasible() : of(all);
+  }
+
+  /**
+   * @return {@code true} when there are no violations.
+   */
+  public boolean isFeasible() {
+    return violations.isEmpty();
+  }
+
+  public List<FeasibilityViolation> getViolations() {
+    return Collections.unmodifiableList(violations);
+  }
+
+  @Override
+  public String toString() {
+    if (violations.isEmpty()) {
+      return "FeasibilityResult: feasible";
+    }
+    return "FeasibilityResult: infeasible, violations=" + violations;
+  }
+}

helix-core/src/main/java/org/apache/helix/util/FeasibilityViolation.java

@@ -0,0 +1,98 @@
+package org.apache.helix.util;
+
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+/**
+ * A single reason a proposed cluster mutation would make the cluster un-rebalanceable.
+ * Produced by {@link RebalanceFeasibilityEvaluator} and surfaced to REST callers as part of a
+ * {@link FeasibilityResult}.
+ */
+public class FeasibilityViolation {
+  /**
+   * The class of rebalance invariant that a proposed mutation would break.
+   */
+  public enum Type {
+    /** A partition would retain fewer than its required minimum active replicas. */
+    MIN_ACTIVE_REPLICA,
+    /** An instance would exceed, or fail to declare, a required WAGED capacity dimension. */
+    CAPACITY,
+    /** A partition that was assignable would be left unassigned. */
+    UNASSIGNED_PARTITION
+  }
+
+  private final Type type;
+  private final String resourceName;
+  private final String partitionName;
+  private final String instanceName;
+  private final String detail;
+
+  public FeasibilityViolation(Type type, String resourceName, String partitionName,
+      String instanceName, String detail) {
+    this.type = type;
+    this.resourceName = resourceName;
+    this.partitionName = partitionName;
+    this.instanceName = instanceName;
+    this.detail = detail;
+  }
+
+  public static FeasibilityViolation minActiveReplica(String resourceName, String partitionName,
+      int currentActiveReplicas, int requiredMinActiveReplicas) {
+    String detail = String.format("Partition %s has %d/%d active replicas", partitionName,
+        currentActiveReplicas, requiredMinActiveReplicas);
+    return new FeasibilityViolation(Type.MIN_ACTIVE_REPLICA, resourceName, partitionName, null,
+        detail);
+  }
+
+  public static FeasibilityViolation capacity(String instanceName, String detail) {
+    return new FeasibilityViolation(Type.CAPACITY, null, null, instanceName, detail);
+  }
+
+  public static FeasibilityViolation unassignedPartition(String resourceName,
+      String partitionName) {
+    String detail = String.format("Partition %s would be left unassigned", partitionName);
+    return new FeasibilityViolation(Type.UNASSIGNED_PARTITION, resourceName, partitionName, null,
+        detail);
+  }
+
+  public Type getType() {
+    return type;
+  }
+
+  public String getResourceName() {
+    return resourceName;
+  }
+
+  public String getPartitionName() {
+    return partitionName;
+  }
+
+  public String getInstanceName() {
+    return instanceName;
+  }
+
+  public String getDetail() {
+    return detail;
+  }
+
+  @Override
+  public String toString() {
+    return String.format("%s: %s", type, detail);
+  }
+}

helix-core/src/main/java/org/apache/helix/util/RebalanceFeasibilityEvaluator.java

@@ -0,0 +1,137 @@
+package org.apache.helix.util;
+
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+
+import org.apache.helix.HelixException;
+import org.apache.helix.controller.rebalancer.util.WagedValidationUtil;
+import org.apache.helix.model.ClusterConfig;
+import org.apache.helix.model.InstanceConfig;
+
+/**
+ * Evaluates whether a proposed cluster mutation (dropping an instance, shrinking capacity,
+ * marking an instance EVACUATE, or onboarding a resource) would leave the cluster in a
+ * rebalanceable state. It asserts post-conditions over a computed assignment that the WAGED /
+ * FULL_AUTO rebalancer would produce for the proposed state, without applying the mutation.
+ *
+ * <p>The assignment computation itself is performed by the existing read-only rebalancer
+ * primitives ({@code HelixUtil#getTargetAssignmentForWagedFullAuto} /
+ * {@code HelixUtil#getIdealAssignmentForFullAuto}); this class is the ZooKeeper-free invariant
+ * checker applied to the result.
+ */
+public class RebalanceFeasibilityEvaluator {
+
+  /**
+   * Checks the minimum-active-replica post-condition for a single resource's computed assignment.
+   *
+   * @param resourceName       the resource being checked
+   * @param partitionStateMap  computed assignment for the resource: partition -> (instance ->
+   *                           state)
+   * @param minActiveReplicas  the resource's required minimum active replicas; {@code -1} means no
+   *                           constraint (the check is skipped)
+   * @param unhealthyStates    states that do NOT count as an active replica (e.g. OFFLINE, ERROR,
+   *                           DROPPED), as defined by
+   *                           {@code InstanceValidationUtil#getUnhealthyStates}
+   * @return a {@link FeasibilityResult} listing every partition that would fall below the threshold
+   */
+  public FeasibilityResult checkMinActiveReplicas(String resourceName,
+      Map<String, Map<String, String>> partitionStateMap, int minActiveReplicas,
+      Set<String> unhealthyStates) {
+    // -1 means the resource declares no min-active-replica constraint; nothing to enforce.
+    if (minActiveReplicas < 0) {
+      return FeasibilityResult.feasible();
+    }
+
+    List<FeasibilityViolation> violations = new ArrayList<>();
+    for (Map.Entry<String, Map<String, String>> partitionEntry : partitionStateMap.entrySet()) {
+      String partitionName = partitionEntry.getKey();
+      int activeReplicas = 0;
+      for (String state : partitionEntry.getValue().values()) {
+        if (!unhealthyStates.contains(state)) {
+          activeReplicas++;
+        }
+      }
+      if (activeReplicas < minActiveReplicas) {
+        violations.add(FeasibilityViolation.minActiveReplica(resourceName, partitionName,
+            activeReplicas, minActiveReplicas));
+      }
+    }
+
+    return violations.isEmpty() ? FeasibilityResult.feasible() : FeasibilityResult.of(violations);
+  }
+
+  /**
+   * Checks the no-unassigned-partition post-condition: every expected partition of a resource must
+   * be assigned to at least one instance in the computed assignment.
+   *
+   * @param resourceName        the resource being checked
+   * @param partitionStateMap   computed assignment for the resource: partition -> (instance ->
+   *                            state)
+   * @param expectedPartitions  the full set of partitions the resource is expected to have
+   * @return a {@link FeasibilityResult} listing every partition that would be left unassigned
+   */
+  public FeasibilityResult checkNoUnassignedPartitions(String resourceName,
+      Map<String, Map<String, String>> partitionStateMap, Set<String> expectedPartitions) {
+    List<FeasibilityViolation> violations = new ArrayList<>();
+    for (String partition : expectedPartitions) {
+      Map<String, String> stateByInstance = partitionStateMap.get(partition);
+      if (stateByInstance == null || stateByInstance.isEmpty()) {
+        violations.add(FeasibilityViolation.unassignedPartition(resourceName, partition));
+      }
+    }
+
+    return violations.isEmpty() ? FeasibilityResult.feasible() : FeasibilityResult.of(violations);
+  }
+
+  /**
+   * Checks the WAGED capacity post-condition for a set of proposed instance configs: every
+   * instance must declare all of the cluster's required capacity keys. Reuses the canonical
+   * {@code WagedValidationUtil#validateAndGetInstanceCapacity} so the verdict matches what the
+   * rebalancer would accept. A cluster with no capacity keys configured (non-WAGED) is a no-op.
+   *
+   * @param clusterConfig    the (proposed) cluster config defining required capacity keys
+   * @param instanceConfigs  the (proposed) instance configs to validate
+   * @return a {@link FeasibilityResult} listing every instance that would break capacity validation
+   */
+  public FeasibilityResult checkInstanceCapacities(ClusterConfig clusterConfig,
+      List<InstanceConfig> instanceConfigs) {
+    List<String> requiredCapacityKeys = clusterConfig.getInstanceCapacityKeys();
+    // No capacity keys configured (non-WAGED cluster): nothing to enforce.
+    if (requiredCapacityKeys == null || requiredCapacityKeys.isEmpty()) {
+      return FeasibilityResult.feasible();
+    }
+
+    List<FeasibilityViolation> violations = new ArrayList<>();
+    for (InstanceConfig instanceConfig : instanceConfigs) {
+      try {
+        WagedValidationUtil.validateAndGetInstanceCapacity(clusterConfig, instanceConfig);
+      } catch (HelixException e) {
+        violations.add(
+            FeasibilityViolation.capacity(instanceConfig.getInstanceName(), e.getMessage()));
+      }
+    }
+
+    return violations.isEmpty() ? FeasibilityResult.feasible() : FeasibilityResult.of(violations);
+  }
+}