Release for helix 1.4.4-dev-20250422

helix-core/src/main/java/org/apache/helix/HelixAdmin.java

@@ -801,12 +801,24 @@ Map<String, Boolean> validateInstancesForWagedRebalance(String clusterName,
    * Return if instance operation 'Evacuate' is finished.
    * @param clusterName
    * @param instancesNames
-   * @return Return true if there is no current state nor pending message on the instance.
+   * @return Return true if there is no FULL_AUTO or CUSTOMIZED resources in the current state nor
+   * any pending message on the instance.
    */
   default boolean isEvacuateFinished(String clusterName, String instancesNames) {
     throw new UnsupportedOperationException("isEvacuateFinished is not implemented.");
   }
 
+  /**
+   * Check to see if instance is drained.
+   * @param clusterName
+   * @param instanceName
+   * @return Return true if there is no FULL_AUTO or CUSTOMIZED resources in the current state nor
+   * any pending message on the instance.
+   */
+  default boolean isInstanceDrained(String clusterName, String instanceName) {
+    throw new UnsupportedOperationException("isInstanceDrained is not implemented.");
+  }
+
   /**
    * Check to see if swapping between two instances can be completed. Either the swapOut or
    * swapIn instance can be passed in.

helix-core/src/main/java/org/apache/helix/controller/dataproviders/BaseControllerDataProvider.java

@@ -1049,6 +1049,9 @@ public void notifyDataChange(HelixConstants.ChangeType changeType, String pathCh
 
   private void updateOfflineInstanceHistory(HelixDataAccessor accessor) {
     if (!_updateInstanceOfflineTime) {
+      // Clean up entries for nodes that have been removed from the cluster. This prevents a stale offline time from
+      // being used when the node is re-added to the cluster but before it updates its offline time.
+      _instanceOfflineTimeMap.keySet().retainAll(_allInstanceConfigCache.getPropertyMap().keySet());
       return;
     }
     List<String> offlineNodes = new ArrayList<>(_allInstanceConfigCache.getPropertyMap().keySet());

helix-core/src/main/java/org/apache/helix/controller/rebalancer/waged/constraints/ConstraintBasedAlgorithm.java

@@ -143,7 +143,7 @@ private Optional<AssignableNode> getNodeWithHighestPoints(AssignableReplica repl
       return Optional.empty();
     }
 
-    LOG.info("Disabling hard constraint level logging for cluster: {}", clusterContext.getClusterName());
+    LOG.debug("Disabling hard constraint level logging for cluster: {}", clusterContext.getClusterName());
     removeFullLoggingForCluster();
 
     return candidateNodes.parallelStream().map(node -> new HashMap.SimpleEntry<>(node,

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

@@ -10,8 +10,6 @@
 import org.apache.helix.controller.pipeline.AsyncWorkerType;
 import org.apache.helix.model.ClusterConfig;
 import org.apache.helix.model.InstanceConfig;
-import org.apache.helix.model.LiveInstance;
-import org.apache.helix.model.ParticipantHistory;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import static org.apache.helix.util.RebalanceUtil.scheduleOnDemandPipeline;

helix-core/src/main/java/org/apache/helix/manager/zk/ZKHelixAdmin.java

@@ -211,7 +211,7 @@ public void addInstance(String clusterName, InstanceConfig instanceConfig) {
     }
 
     List<InstanceConfig> matchingLogicalIdInstances =
-        InstanceUtil.findInstancesWithMatchingLogicalId(_configAccessor, clusterName,
+        InstanceUtil.findInstancesWithMatchingLogicalId(_baseDataAccessor, clusterName,
             instanceConfig);
     if (matchingLogicalIdInstances.size() > 1) {
       throw new HelixException(
@@ -224,7 +224,8 @@ public void addInstance(String clusterName, InstanceConfig instanceConfig) {
     InstanceConstants.InstanceOperation attemptedInstanceOperation =
         instanceConfig.getInstanceOperation().getOperation();
     try {
-      InstanceUtil.validateInstanceOperationTransition(_configAccessor, clusterName, instanceConfig,
+      InstanceUtil.validateInstanceOperationTransition(_baseDataAccessor, clusterName,
+          instanceConfig,
           InstanceConstants.InstanceOperation.UNKNOWN, attemptedInstanceOperation);
     } catch (HelixException e) {
       instanceConfig.setInstanceOperation(InstanceConstants.InstanceOperation.UNKNOWN);
@@ -458,13 +459,18 @@ public void setInstanceOperation(String clusterName, String instanceName,
   }
 
   @Override
-  public boolean isEvacuateFinished(String clusterName, String instanceName) {
-    if (!instanceHasFullAutoCurrentStateOrMessage(clusterName, instanceName)) {
-      InstanceConfig config = getInstanceConfig(clusterName, instanceName);
-      return config != null && config.getInstanceOperation().getOperation()
-          .equals(InstanceConstants.InstanceOperation.EVACUATE);
+  public boolean isEvacuateFinished(String clusterName, String instanceNames) {
+    InstanceConfig config = getInstanceConfig(clusterName, instanceNames);
+    if (config == null || config.getInstanceOperation().getOperation() !=
+        InstanceConstants.InstanceOperation.EVACUATE ) {
+      return false;
     }
-    return false;
+    return !instanceHasCurrentStateOrMessage(clusterName, instanceNames);
+  }
+
+  @Override
+  public boolean isInstanceDrained(String clusterName, String instanceName) {
+    return !instanceHasCurrentStateOrMessage(clusterName, instanceName);
   }
 
   /**
@@ -616,7 +622,7 @@ public boolean canCompleteSwap(String clusterName, String instanceName) {
     }
 
     List<InstanceConfig> swappingInstances =
-        InstanceUtil.findInstancesWithMatchingLogicalId(_configAccessor, clusterName,
+        InstanceUtil.findInstancesWithMatchingLogicalId(_baseDataAccessor, clusterName,
             instanceConfig);
     if (swappingInstances.size() != 1) {
       logger.warn(
@@ -655,7 +661,7 @@ public boolean completeSwapIfPossible(String clusterName, String instanceName,
     }
 
     List<InstanceConfig> swappingInstances =
-        InstanceUtil.findInstancesWithMatchingLogicalId(_configAccessor, clusterName,
+        InstanceUtil.findInstancesWithMatchingLogicalId(_baseDataAccessor, clusterName,
             instanceConfig);
     if (swappingInstances.size() != 1) {
       logger.warn(
@@ -720,7 +726,7 @@ public boolean completeSwapIfPossible(String clusterName, String instanceName,
 
   @Override
   public boolean isReadyForPreparingJoiningCluster(String clusterName, String instanceName) {
-    if (!instanceHasFullAutoCurrentStateOrMessage(clusterName, instanceName)) {
+    if (!instanceHasCurrentStateOrMessage(clusterName, instanceName)) {
       InstanceConfig config = getInstanceConfig(clusterName, instanceName);
       return config != null && INSTANCE_OPERATION_TO_EXCLUDE_FROM_ASSIGNMENT.contains(
           config.getInstanceOperation().getOperation());
@@ -756,13 +762,14 @@ public boolean forceKillInstance(String clusterName, String instanceName, String
   }
 
   /**
-   * Return true if Instance has any current state or pending message. Otherwise, return false if instance is offline,
+   * Return true if instance has any resource with FULL_AUTO or CUSTOMIZED rebalance mode in current state or
+   * if instance has any pending message. Otherwise, return false if instance is offline,
    * instance has no active session, or if instance is online but has no current state or pending message.
    * @param clusterName
    * @param instanceName
    * @return
    */
-  private boolean instanceHasFullAutoCurrentStateOrMessage(String clusterName,
+  private boolean instanceHasCurrentStateOrMessage(String clusterName,
       String instanceName) {
     HelixDataAccessor accessor = new ZKHelixDataAccessor(clusterName, _baseDataAccessor);
     PropertyKey.Builder keyBuilder = accessor.keyBuilder();
@@ -802,13 +809,14 @@ private boolean instanceHasFullAutoCurrentStateOrMessage(String clusterName,
       return true;
     }
 
-    // Get set of FULL_AUTO resources
+    // Get set of FULL_AUTO and CUSTOMIZED resources
     List<IdealState> idealStates = accessor.getChildValues(keyBuilder.idealStates(), true);
-    Set<String> fullAutoResources = idealStates != null ? idealStates.stream()
-        .filter(idealState -> idealState.getRebalanceMode() == RebalanceMode.FULL_AUTO)
+    Set<String> resources = idealStates != null ? idealStates.stream()
+        .filter(idealState -> idealState.getRebalanceMode() == RebalanceMode.FULL_AUTO ||
+            idealState.getRebalanceMode() == RebalanceMode.CUSTOMIZED)
         .map(IdealState::getResourceName).collect(Collectors.toSet()) : Collections.emptySet();
 
-    return currentStates.stream().anyMatch(fullAutoResources::contains);
+    return currentStates.stream().anyMatch(resources::contains);
   }
 
   @Override