Refine the WAGED rebalancer related interfaces for integration (apach…

…e#431)

* Refine the WAGED rebalancer related interfaces and initial integrate with the BestPossibleStateCalStage.

- Modify the BestPossibleStateCalStage logic to plugin the WAGED rebalancer.
- Refine ClusterModel to integrate with the ClusterDataDetector implementation.
- Enabling getting the changed details for Cluster Config in the change detector. Which is required by the WAGED rebalancer.

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

@@ -0,0 +1,43 @@
+package org.apache.helix;
+
+/*
+ * 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.
+ */
+
+/**
+ * Exception thrown by Helix due to rebalance failures.
+ */
+public class HelixRebalanceException extends Exception {
+  enum RebalanceFailureType {
+    INVALID_CLUSTER_STATUS,
+    INVALID_REBALANCER_STATUS,
+    FAILED_TO_CALCULATE,
+    UNKNOWN_FAILURE
+  }
+
+  private final RebalanceFailureType _type;
+
+  public HelixRebalanceException(String message, RebalanceFailureType type, Throwable cause) {
+    super(String.format("%s. Failure Type: %s", message, type.name()), cause);
+    _type = type;
+  }
+
+  public RebalanceFailureType getFailureType() {
+    return _type;
+  }
+}

helix-core/src/main/java/org/apache/helix/controller/changedetector/ResourceChangeDetector.java

@@ -20,15 +20,17 @@
  */
 
 import com.google.common.collect.Sets;
+import org.apache.helix.HelixConstants;
+import org.apache.helix.HelixProperty;
+import org.apache.helix.controller.dataproviders.ResourceControllerDataProvider;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
 import java.util.Collection;
 import java.util.Collections;
 import java.util.HashMap;
 import java.util.HashSet;
 import java.util.Map;
-import org.apache.helix.HelixConstants;
-import org.apache.helix.HelixException;
-import org.apache.helix.HelixProperty;
-import org.apache.helix.controller.dataproviders.ResourceControllerDataProvider;
 
 /**
  * ResourceChangeDetector implements ChangeDetector. It caches resource-related metadata from
@@ -37,6 +39,7 @@
  * WARNING: the methods of this class are not thread-safe.
  */
 public class ResourceChangeDetector implements ChangeDetector {
+  private static final Logger LOG = LoggerFactory.getLogger(ResourceChangeDetector.class.getName());
 
   private ResourceChangeSnapshot _oldSnapshot; // snapshot for previous pipeline run
   private ResourceChangeSnapshot _newSnapshot; // snapshot for this pipeline run
@@ -108,10 +111,13 @@ private void clearCachedComputation() {
       return snapshot.getResourceConfigMap();
     case LIVE_INSTANCE:
       return snapshot.getLiveInstances();
+    case CONFIG:
+      return Collections.emptyMap();
     default:
-      throw new HelixException(String.format(
-          "ResourceChangeDetector cannot compute the names of changes for the given ChangeType: %s",
-          changeType));
+      LOG.warn(
+          "ResourceChangeDetector cannot compute the names of changes for the given ChangeType: {}",
+          changeType);
+      return Collections.emptyMap();
     }
   }
 

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

@@ -19,10 +19,13 @@
  * under the License.
  */
 
-import org.apache.helix.HelixException;
 import org.apache.helix.HelixManager;
+import org.apache.helix.HelixRebalanceException;
+import org.apache.helix.controller.changedetector.ResourceChangeDetector;
 import org.apache.helix.controller.dataproviders.ResourceControllerDataProvider;
-import org.apache.helix.controller.rebalancer.GlobalRebalancer;
+import org.apache.helix.controller.rebalancer.DelayedAutoRebalancer;
+import org.apache.helix.controller.rebalancer.internal.MappingCalculator;
+import org.apache.helix.controller.rebalancer.waged.constraints.ConstraintsRebalanceAlgorithm;
 import org.apache.helix.controller.stages.CurrentStateOutput;
 import org.apache.helix.model.IdealState;
 import org.apache.helix.model.Resource;
@@ -36,23 +39,57 @@
  * A placeholder before we have the implementation.
  * Weight-Aware Globally-Even Distribute Rebalancer.
  *
- * @see <a href="Design Document">https://github.com/apache/helix/wiki/Design-Proposal---Weight-Aware-Globally-Even-Distribute-Rebalancer</a>
+ * @see <a href="https://github.com/apache/helix/wiki/Design-Proposal---Weight-Aware-Globally-Even-Distribute-Rebalancer">
+ * Design Document
+ * </a>
  */
-public class WagedRebalancer implements GlobalRebalancer<ResourceControllerDataProvider> {
+public class WagedRebalancer {
   private static final Logger LOG = LoggerFactory.getLogger(WagedRebalancer.class);
 
-  @Override
-  public void init(HelixManager manager) { }
+  // --------- The following fields are placeholders and need replacement. -----------//
+  // TODO Shall we make the metadata store a static threadlocal object as well to avoid reinitialization?
+  private final AssignmentMetadataStore _assignmentMetadataStore;
+  private final RebalanceAlgorithm _rebalanceAlgorithm;
+  // ------------------------------------------------------------------------------------//
 
-  @Override
-  public Map<String, IdealState> computeNewIdealState(CurrentStateOutput currentStateOutput,
-      ResourceControllerDataProvider clusterData, Map<String, Resource> resourceMap)
-      throws HelixException {
-    return new HashMap<>();
+  // The cluster change detector is a stateful object. Make it static to avoid unnecessary
+  // reinitialization.
+  private static final ThreadLocal<ResourceChangeDetector> CHANGE_DETECTOR_THREAD_LOCAL =
+      new ThreadLocal<>();
+  private final MappingCalculator<ResourceControllerDataProvider> _mappingCalculator;
+
+  private ResourceChangeDetector getChangeDetector() {
+    if (CHANGE_DETECTOR_THREAD_LOCAL.get() == null) {
+      CHANGE_DETECTOR_THREAD_LOCAL.set(new ResourceChangeDetector());
+    }
+    return CHANGE_DETECTOR_THREAD_LOCAL.get();
+  }
+
+  public WagedRebalancer(HelixManager helixManager) {
+    // TODO init the metadata store according to their requirement when integrate, or change to final static method if possible.
+    _assignmentMetadataStore = new AssignmentMetadataStore();
+    // TODO init the algorithm according to the requirement when integrate.
+    _rebalanceAlgorithm = new ConstraintsRebalanceAlgorithm();
+
+    // Use the mapping calculator in DelayedAutoRebalancer for calculating the final assignment
+    // output.
+    // This calculator will translate the best possible assignment into an applicable state mapping
+    // based on the current states.
+    // TODO abstract and separate the mapping calculator logic from the DelayedAutoRebalancer
+    _mappingCalculator = new DelayedAutoRebalancer();
   }
 
-  @Override
-  public RebalanceFailureReason getFailureReason() {
-    return new RebalanceFailureReason(RebalanceFailureType.UNKNOWN_FAILURE);
+  /**
+   * Compute the new IdealStates for all the resources input. The IdealStates include both the new
+   * partition assignment (in the listFiles) and the new replica state mapping (in the mapFields).
+   * @param clusterData        The Cluster status data provider.
+   * @param resourceMap        A map containing all the rebalancing resources.
+   * @param currentStateOutput The present Current State of the cluster.
+   * @return A map containing the computed new IdealStates.
+   */
+  public Map<String, IdealState> computeNewIdealStates(ResourceControllerDataProvider clusterData,
+      Map<String, Resource> resourceMap, final CurrentStateOutput currentStateOutput)
+      throws HelixRebalanceException {
+    return new HashMap<>();
   }
 }