[Testing] Soft delete the instance, and add Orphan deletion

Author: LZD-PratyushBhatt

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

@@ -235,6 +235,17 @@ void addResource(String clusterName, String resourceName, int numPartitions, Str
    */
   void dropInstance(String clusterName, InstanceConfig instanceConfig);
 
+  /**
+   * Soft drop an instance from a cluster. Deletes InstanceConfig immediately to make the instance
+   * non-assignable, then relies on async cleanup to remove the instance runtime path.
+   * Use this for instances with large numbers of pending messages to avoid jute.maxbuffer issues.
+   * @param clusterName
+   * @param instanceConfig
+   */
+  default void softDropInstance(String clusterName, InstanceConfig instanceConfig) {
+    throw new UnsupportedOperationException("softDropInstance is not implemented.");
+  }
+
   /**
    * Purge offline instances that have been offline for longer than the offline duration time
    * from a cluster

helix-core/src/main/java/org/apache/helix/controller/GenericHelixController.java

@@ -89,6 +89,7 @@
 import org.apache.helix.controller.stages.MessageSelectionStage;
 import org.apache.helix.controller.stages.MessageThrottleStage;
 import org.apache.helix.controller.stages.ParticipantDeregistrationStage;
+import org.apache.helix.controller.stages.OrphanedInstanceCleanupStage;
 import org.apache.helix.controller.stages.PersistAssignmentStage;
 import org.apache.helix.controller.stages.ReadClusterDataStage;
 import org.apache.helix.controller.stages.ResourceComputationStage;
@@ -537,6 +538,7 @@ private static PipelineRegistry createDefaultRegistry(String pipelineName) {
       rebalancePipeline.addStage(new PersistAssignmentStage());
       rebalancePipeline.addStage(new TargetExteralViewCalcStage());
       rebalancePipeline.addStage(new ParticipantDeregistrationStage());
+      rebalancePipeline.addStage(new OrphanedInstanceCleanupStage());
 
       // external view generation
       Pipeline externalViewPipeline = new Pipeline(pipelineName);

helix-core/src/main/java/org/apache/helix/controller/pipeline/AsyncWorkerType.java

@@ -34,5 +34,6 @@ public enum AsyncWorkerType {
   MaintenanceRecoveryWorker,
   TaskJobPurgeWorker,
   CustomizedStateViewComputeWorker,
-  ParticipantDeregistrationWorker
+  ParticipantDeregistrationWorker,
+  OrphanedInstanceCleanupWorker
 }

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

@@ -0,0 +1,98 @@
+package org.apache.helix.controller.stages;
+
+/*
+ * 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.List;
+
+import org.apache.helix.AccessOption;
+import org.apache.helix.HelixManager;
+import org.apache.helix.PropertyPathBuilder;
+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.ClusterConfig;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Cleans up orphaned instance runtime paths when InstanceConfig is deleted but the instance
+ * subtree under INSTANCES remains.
+ */
+public class OrphanedInstanceCleanupStage extends AbstractAsyncBaseStage {
+  private static final Logger LOG = LoggerFactory.getLogger(OrphanedInstanceCleanupStage.class);
+
+  @Override
+  public AsyncWorkerType getAsyncWorkerType() {
+    return AsyncWorkerType.OrphanedInstanceCleanupWorker;
+  }
+
+  @Override
+  public void execute(final ClusterEvent event) {
+    HelixManager manager = event.getAttribute(AttributeName.helixmanager.name());
+    ResourceControllerDataProvider cache =
+        event.getAttribute(AttributeName.ControllerDataProvider.name());
+    if (manager == null || cache == null) {
+      return;
+    }
+
+    ClusterConfig clusterConfig = cache.getClusterConfig();
+    if (clusterConfig == null || !clusterConfig.isAsyncInstanceDropEnabled()) {
+      return;
+    }
+
+    String clusterName = cache.getClusterName();
+
+    try {
+      List<String> instanceNames = manager.getHelixDataAccessor().getBaseDataAccessor()
+          .getChildNames(PropertyPathBuilder.instance(clusterName), AccessOption.PERSISTENT);
+      if (instanceNames == null || instanceNames.isEmpty()) {
+        return;
+      }
+
+      for (String instanceName : instanceNames) {
+        String instanceConfigPath = PropertyPathBuilder.instanceConfig(clusterName, instanceName);
+        if (manager.getHelixDataAccessor().getBaseDataAccessor()
+            .exists(instanceConfigPath, AccessOption.PERSISTENT)) {
+          continue;
+        }
+
+        String liveInstancePath = PropertyPathBuilder.liveInstance(clusterName, instanceName);
+        if (manager.getHelixDataAccessor().getBaseDataAccessor()
+            .exists(liveInstancePath, AccessOption.PERSISTENT)) {
+          LOG.warn("Found live instance {} without InstanceConfig in cluster {}. Skip orphan cleanup.",
+              instanceName, clusterName);
+          continue;
+        }
+
+        String instancePath = PropertyPathBuilder.instance(clusterName, instanceName);
+        LOG.info("Cleaning up orphaned instance path {}", instancePath);
+        boolean removed = manager.getHelixDataAccessor().getBaseDataAccessor()
+            .remove(instancePath, AccessOption.PERSISTENT);
+        if (removed) {
+          LOG.info("Removed orphaned instance path {} in cluster {}.", instancePath, clusterName);
+        } else {
+          LOG.warn("Failed to remove orphaned instance path {} in cluster {}. Will retry.", instancePath, clusterName);
+        }
+      }
+    } catch (Exception e) {
+      LOG.warn("OrphanedInstanceCleanupStage failed for cluster {}.", clusterName, e);
+    }
+  }
+}

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

@@ -292,9 +292,6 @@ private void dropInstancePathsRecursively(String clusterName, String instanceNam
       } catch (ZkClientException e) {
         if (retryCnt < 3 && e.getCause() instanceof ZkException && e.getCause()
             .getCause() instanceof KeeperException.NotEmptyException) {
-          // Racing condition with controller's persisting node history, retryable.
-          // We don't need to backoff here as this racing condition only happens once (controller
-          // does not repeatedly write instance history)
           logger.warn("Retrying dropping instance {} with exception {}", instanceName,
               e.getCause().getMessage());
           retryCnt++;
@@ -308,6 +305,27 @@ private void dropInstancePathsRecursively(String clusterName, String instanceNam
     }
   }
 
+  @Override
+  public void softDropInstance(String clusterName, InstanceConfig instanceConfig) {
+    logger.info("Soft drop instance {} from cluster {}.", instanceConfig.getInstanceName(), clusterName);
+    String instanceName = instanceConfig.getInstanceName();
+
+    String instanceConfigPath = PropertyPathBuilder.instanceConfig(clusterName, instanceName);
+    if (!_zkClient.exists(instanceConfigPath)) {
+      throw new HelixException(
+          "Node " + instanceName + " does not exist in config for cluster " + clusterName);
+    }
+
+    String liveInstancePath = PropertyPathBuilder.liveInstance(clusterName, instanceName);
+    if (_zkClient.exists(liveInstancePath)) {
+      throw new HelixException(
+          "Node " + instanceName + " is still alive for cluster " + clusterName + ", can't drop.");
+    }
+
+    _zkClient.delete(instanceConfigPath);
+    logger.info("Deleted InstanceConfig for {}. OrphanedInstanceCleanupStage will clean up INSTANCES path asynchronously.", instanceName);
+  }
+
   /**
    * Please note that the purge function should only be called when there is no new instance
    * joining happening in the cluster. The reason is that current implementation is not thread safe,

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

@@ -170,6 +170,9 @@ public enum ClusterConfigProperty {
     // How long offline nodes will stay in the cluster before they are automatically purged, in milliseconds
     PARTICIPANT_DEREGISTRATION_TIMEOUT,
 
+    // Enables async instance drop cleanup to avoid jute.maxbuffer failures
+    ASYNC_INSTANCE_DROP_ENABLED,
+
     // Allow disabled partitions to remain OFFLINE instead of being reassigned in WAGED rebalancer
     RELAXED_DISABLED_PARTITION_CONSTRAINT,
   }
@@ -1329,4 +1332,12 @@ public void setParticipantDeregistrationTimeout(long timeout) {
   public boolean isParticipantDeregistrationEnabled() {
     return getParticipantDeregistrationTimeout() > -1;
   }
+
+  public boolean isAsyncInstanceDropEnabled() {
+    return _record.getBooleanField(ClusterConfigProperty.ASYNC_INSTANCE_DROP_ENABLED.name(), false);
+  }
+
+  public void setAsyncInstanceDropEnabled(boolean enabled) {
+    _record.setBooleanField(ClusterConfigProperty.ASYNC_INSTANCE_DROP_ENABLED.name(), enabled);
+  }
 }