import org.apache.helix.model.StateModelDefinition; //导入依赖的package包/类
/**
 * Build OnlineOffline state model definition
 */
public static StateModelDefinition build() {
  StateModelDefinition.Builder builder = new StateModelDefinition.Builder(name);
  // init state
  builder.initialState(States.OFFLINE.name());

  // add states
  builder.addState(States.ONLINE.name(), 20);
  builder.addState(States.OFFLINE.name(), -1);
  for (HelixDefinedState state : HelixDefinedState.values()) {
    builder.addState(state.name(), -1);
  }

  // add transitions

  builder.addTransition(States.ONLINE.name(), States.OFFLINE.name(), 25);
  builder.addTransition(States.OFFLINE.name(), States.ONLINE.name(), 5);
  builder.addTransition(States.OFFLINE.name(), HelixDefinedState.DROPPED.name(), 0);

  // bounds
  builder.dynamicUpperBound(States.ONLINE.name(), "R");

  return builder.build();
}
 

import org.apache.helix.model.StateModelDefinition; //导入依赖的package包/类
/**
 * Build OnlineOffline state model definition
 * @return
 */
public static StateModelDefinition build() {
  StateModelDefinition.Builder builder = new StateModelDefinition.Builder(name);
  // init state
  builder.initialState(States.OFFLINE.name());

  // add states
  builder.addState(States.ONLINE.name(), 20);
  builder.addState(States.OFFLINE.name(), -1);
  for (HelixDefinedState state : HelixDefinedState.values()) {
    builder.addState(state.name(), -1);
  }

  // add transitions

  builder.addTransition(States.ONLINE.name(), States.OFFLINE.name(), 25);
  builder.addTransition(States.OFFLINE.name(), States.ONLINE.name(), 5);
  builder.addTransition(States.OFFLINE.name(), HelixDefinedState.DROPPED.name(), 0);

  // bounds
  builder.dynamicUpperBound(States.ONLINE.name(), "R");

  return builder.build();
}
 

import org.apache.helix.model.StateModelDefinition; //导入依赖的package包/类
/**
 * Update per-instance resource bean
 * @param stateMap partition->state
 * @tags tags instance tags
 * @param stateModelDef
 */
public synchronized void update(Map<Partition, String> stateMap, Set<String> tags,
    StateModelDefinition stateModelDef) {
  if (tags == null || tags.isEmpty()) {
    _tags = ImmutableList.of(ClusterStatusMonitor.DEFAULT_TAG);
  } else {
    _tags = Lists.newArrayList(tags);
    Collections.sort(_tags);
  }

  int cnt = 0;
  for (String state : stateMap.values()) {
    // Skip DROPPED and initial state (e.g. OFFLINE)
    if (state.equalsIgnoreCase(HelixDefinedState.DROPPED.name())
        || state.equalsIgnoreCase(stateModelDef.getInitialState())) {
      continue;
    }
    cnt++;
  }
  _partitions = cnt;
}
 

import org.apache.helix.model.StateModelDefinition; //导入依赖的package包/类
public static void setup() {
  admin = new ZKHelixAdmin(ZK_ADDRESS);
  // create cluster
  echo("Creating cluster: " + CLUSTER_NAME);
  admin.addCluster(CLUSTER_NAME, true);

  // Add nodes to the cluster
  echo("Adding " + NUM_NODES + " participants to the cluster");
  for (int i = 0; i < NUM_NODES; i++) {
    admin.addInstance(CLUSTER_NAME, INSTANCE_CONFIG_LIST.get(i));
    echo("\t Added participant: " + INSTANCE_CONFIG_LIST.get(i).getInstanceName());
  }

  // Add a state model
  StateModelDefinition myStateModel = defineStateModel();
  echo("Configuring StateModel: " + "MyStateModel  with 1 Master and 1 Slave");
  admin.addStateModelDef(CLUSTER_NAME, STATE_MODEL_NAME, myStateModel);

  // Add a resource with 6 partitions and 2 replicas
  echo("Adding a resource MyResource: " + "with 6 partitions and 2 replicas");
  admin.addResource(CLUSTER_NAME, RESOURCE_NAME, NUM_PARTITIONS, STATE_MODEL_NAME, "AUTO");
  // this will set up the ideal state, it calculates the preference list for
  // each partition similar to consistent hashing
  admin.rebalance(CLUSTER_NAME, RESOURCE_NAME, NUM_REPLICAS);
}
 

import org.apache.helix.model.StateModelDefinition; //导入依赖的package包/类
@Override
public void addStateModelDef(String clusterName, String stateModelDef,
    StateModelDefinition stateModel, boolean recreateIfExists) {
  if (!ZKUtil.isClusterSetup(clusterName, _zkClient)) {
    throw new HelixException("cluster " + clusterName + " is not setup yet");
  }
  String stateModelDefPath = PropertyPathBuilder.stateModelDef(clusterName);
  String stateModelPath = stateModelDefPath + "/" + stateModelDef;
  if (_zkClient.exists(stateModelPath)) {
    if (recreateIfExists) {
      logger.info(
          "Operation.State Model directory exists:" + stateModelPath + ", remove and recreate.");
      _zkClient.deleteRecursive(stateModelPath);
    } else {
      logger.info("Skip the operation. State Model directory exists:" + stateModelPath);
      return;
    }
  }

  HelixDataAccessor accessor =
      new ZKHelixDataAccessor(clusterName, new ZkBaseDataAccessor<ZNRecord>(_zkClient));
  Builder keyBuilder = accessor.keyBuilder();
  accessor.setProperty(keyBuilder.stateModelDef(stateModelDef), stateModel);
}
 

import org.apache.helix.model.StateModelDefinition; //导入依赖的package包/类
@Override
public boolean createStateModelDef(StateModelDefinition stateModelDef) {
  String path = PropertyPathBuilder.stateModelDef(_clusterName, stateModelDef.getId());
  HelixProperty property =
      getProperty(new PropertyKey.Builder(_clusterName).stateModelDef(stateModelDef.getId()));

  // Set new StateModelDefinition if it is different from old one.
  if (property != null) {
    // StateModelDefinition need to be updated
    if (!new StateModelDefinition(property.getRecord()).equals(stateModelDef)) {
      return stateModelDef.isValid() && _baseDataAccessor
          .set(path, stateModelDef.getRecord(), AccessOption.PERSISTENT);
    }
  } else {
    // StateModeDefinition does not exist
    return stateModelDef.isValid() && _baseDataAccessor
        .create(path, stateModelDef.getRecord(), AccessOption.PERSISTENT);
  }
  // StateModelDefinition exists but not need to be updated
  return true;
}
 

import org.apache.helix.model.StateModelDefinition; //导入依赖的package包/类
/**
 * Compute the best state for all partitions.
 * This is the default implementation, subclasses should re-implement
 * this method if its logic to generate bestpossible map for each partition is different from the default one here.
 *
 * @param cache
 * @param idealState
 * @param resource
 * @param currentStateOutput
 *          Provides the current state and pending state transitions for all partitions
 * @return
 */
@Override
public ResourceAssignment computeBestPossiblePartitionState(ClusterDataCache cache,
    IdealState idealState, Resource resource, CurrentStateOutput currentStateOutput) {
  if (LOG.isDebugEnabled()) {
    LOG.debug("Processing resource:" + resource.getResourceName());
  }
  String stateModelDefName = idealState.getStateModelDefRef();
  StateModelDefinition stateModelDef = cache.getStateModelDef(stateModelDefName);
  ResourceAssignment partitionMapping = new ResourceAssignment(resource.getResourceName());
  for (Partition partition : resource.getPartitions()) {
    Map<String, String> currentStateMap =
        currentStateOutput.getCurrentStateMap(resource.getResourceName(), partition);
    Set<String> disabledInstancesForPartition =
        cache.getDisabledInstancesForPartition(resource.getResourceName(), partition.toString());
    List<String> preferenceList = getPreferenceList(partition, idealState,
        Collections.unmodifiableSet(cache.getLiveInstances().keySet()));
    Map<String, String> bestStateForPartition =
        computeBestPossibleStateForPartition(cache.getLiveInstances().keySet(), stateModelDef, preferenceList,
            currentStateMap, disabledInstancesForPartition, idealState);
    partitionMapping.addReplicaMap(partition, bestStateForPartition);
  }
  return partitionMapping;
}
 

import org.apache.helix.model.StateModelDefinition; //导入依赖的package包/类
protected Map<String, String> computeBestPossibleStateForPartition(Set<String> liveInstances,
    StateModelDefinition stateModelDef, List<String> preferenceList,
    Map<String, String> currentStateMap, Set<String> disabledInstancesForPartition,
    IdealState idealState) {

  if (currentStateMap == null) {
    currentStateMap = Collections.emptyMap();
  }

  // (1) If the partition is removed from IS or the IS is deleted.
  // Transit to DROPPED no matter the instance is disabled or not.
  if (preferenceList == null) {
    return computeBestPossibleMapForDroppedResource(currentStateMap);
  }

  // (2) If resource disabled altogether, transit to initial-state (e.g. OFFLINE) if it's not in ERROR.
  if (!idealState.isEnabled()) {
    return computeBestPossibleMapForDisabledResource(currentStateMap, stateModelDef);
  }

  return computeBestPossibleMap(preferenceList, stateModelDef, currentStateMap, liveInstances,
      disabledInstancesForPartition);
}
 

import org.apache.helix.model.StateModelDefinition; //导入依赖的package包/类
public static int getStateCount(String state, StateModelDefinition stateModelDef, int liveAndEnabledSize,
    int preferenceListSize) {
  String num = stateModelDef.getNumInstancesPerState(state);
  int stateCount = -1;
  if ("N".equals(num)) {
    stateCount = liveAndEnabledSize;
  } else if ("R".equals(num)) {
    stateCount = preferenceListSize;
  } else {
    try {
      stateCount = Integer.parseInt(num);
    } catch (Exception e) {
      LOG.error("Invalid count for state:" + state + " ,count=" + num);
    }
  }

  return stateCount;
}
 

import org.apache.helix.model.StateModelDefinition; //导入依赖的package包/类
private Map<String, Map<String, String>> computeIdealPartitionState(ClusterDataCache cache,
    IdealState idealState) {
  String stateModelDefName = idealState.getStateModelDefRef();
  StateModelDefinition stateModelDef = cache.getStateModelDef(stateModelDefName);

  Map<String, Map<String, String>> idealPartitionState =
      new HashMap<String, Map<String, String>>();

  Set<String> liveEnabledInstances = new HashSet<String>(cache.getLiveInstances().keySet());
  liveEnabledInstances.removeAll(cache.getDisabledInstances());

  for (String partition : idealState.getPartitionSet()) {
    List<String> preferenceList = AbstractRebalancer
        .getPreferenceList(new Partition(partition), idealState, liveEnabledInstances);
    Map<String, String> idealMapping =
        HelixUtil.computeIdealMapping(preferenceList, stateModelDef, liveEnabledInstances);
    idealPartitionState.put(partition, idealMapping);
  }

  return idealPartitionState;
}
 

import org.apache.helix.model.StateModelDefinition; //导入依赖的package包/类
@Test
public void testDisableResource() {
  String className = TestHelper.getTestClassName();
  String methodName = TestHelper.getTestMethodName();
  String clusterName = className + "_" + methodName;
  System.out.println("START " + clusterName + " at " + new Date(System.currentTimeMillis()));
  HelixAdmin admin = new ZKHelixAdmin(_gZkClient);
  admin.addCluster(clusterName, true);
  Assert.assertTrue(ZKUtil.isClusterSetup(clusterName, _gZkClient), "Cluster should be setup");
  String resourceName = "TestDB";
  admin.addStateModelDef(clusterName, "MasterSlave", new StateModelDefinition(
      StateModelConfigGenerator.generateConfigForMasterSlave()));
  admin.addResource(clusterName, resourceName, 4, "MasterSlave");
  admin.enableResource(clusterName, resourceName, false);
  BaseDataAccessor<ZNRecord> baseAccessor = new ZkBaseDataAccessor<ZNRecord>(_gZkClient);
  HelixDataAccessor accessor = new ZKHelixDataAccessor(clusterName, baseAccessor);
  PropertyKey.Builder keyBuilder = accessor.keyBuilder();
  IdealState idealState = accessor.getProperty(keyBuilder.idealStates(resourceName));
  Assert.assertFalse(idealState.isEnabled());
  admin.enableResource(clusterName, resourceName, true);
  idealState = accessor.getProperty(keyBuilder.idealStates(resourceName));
  Assert.assertTrue(idealState.isEnabled());
  System.out.println("END " + clusterName + " at " + new Date(System.currentTimeMillis()));
}
 

import org.apache.helix.model.StateModelDefinition; //导入依赖的package包/类
protected void setupStateModel(String clusterName) {
  ZKHelixDataAccessor accessor =
      new ZKHelixDataAccessor(clusterName, new ZkBaseDataAccessor<ZNRecord>(_gZkClient));
  Builder keyBuilder = accessor.keyBuilder();

  StateModelDefinition masterSlave =
      new StateModelDefinition(StateModelConfigGenerator.generateConfigForMasterSlave());
  accessor.setProperty(keyBuilder.stateModelDef(masterSlave.getId()), masterSlave);

  StateModelDefinition leaderStandby =
      new StateModelDefinition(StateModelConfigGenerator.generateConfigForLeaderStandby());
  accessor.setProperty(keyBuilder.stateModelDef(leaderStandby.getId()), leaderStandby);

  StateModelDefinition onlineOffline =
      new StateModelDefinition(StateModelConfigGenerator.generateConfigForOnlineOffline());
  accessor.setProperty(keyBuilder.stateModelDef(onlineOffline.getId()), onlineOffline);

}
 

import org.apache.helix.model.StateModelDefinition; //导入依赖的package包/类
public void createCluster(String clusterName) {
  _helixAdmin.addCluster(clusterName);
  ZNRecord masterSlave = StateModelConfigGenerator.generateConfigForMasterSlave();
  _helixAdmin.addStateModelDef(clusterName, masterSlave.getId(), new StateModelDefinition(
      masterSlave));
  ZNRecord onlineOffline = StateModelConfigGenerator.generateConfigForOnlineOffline();
  _helixAdmin.addStateModelDef(clusterName, onlineOffline.getId(), new StateModelDefinition(
      onlineOffline));
  ZNRecord leaderStandby = StateModelConfigGenerator.generateConfigForLeaderStandby();
  _helixAdmin.addStateModelDef(clusterName, leaderStandby.getId(), new StateModelDefinition(
      leaderStandby));
}
 

import org.apache.helix.model.StateModelDefinition; //导入依赖的package包/类
private static StateModelDefinition defineStateModel() {
  StateModelDefinition.Builder builder = new StateModelDefinition.Builder(STATE_MODEL_NAME);
  // Add states and their rank to indicate priority. Lower the rank higher the
  // priority
  builder.addState(MASTER, 1);
  builder.addState(SLAVE, 2);
  builder.addState(OFFLINE);
  builder.addState(DROPPED);
  // Set the initial state when the node starts
  builder.initialState(OFFLINE);

  // Add transitions between the states.
  builder.addTransition(OFFLINE, SLAVE);
  builder.addTransition(SLAVE, OFFLINE);
  builder.addTransition(SLAVE, MASTER);
  builder.addTransition(MASTER, SLAVE);
  builder.addTransition(OFFLINE, DROPPED);

  // set constraints on states.
  // static constraint
  builder.upperBound(MASTER, 1);
  // dynamic constraint, R means it should be derived based on the replication
  // factor.
  builder.dynamicUpperBound(SLAVE, "R");

  StateModelDefinition statemodelDefinition = builder.build();
  return statemodelDefinition;
}
 

import org.apache.helix.model.StateModelDefinition; //导入依赖的package包/类
@Override
public StateModelDefinition getStateModelDef(String clusterName,
    String stateModelName) {
  HelixDataAccessor accessor =
      new ZKHelixDataAccessor(clusterName, new ZkBaseDataAccessor<ZNRecord>(_zkClient));
  Builder keyBuilder = accessor.keyBuilder();

  return accessor.getProperty(keyBuilder.stateModelDef(stateModelName));
}
 

import org.apache.helix.model.StateModelDefinition; //导入依赖的package包/类
/**
 * Provides the state model definition for a given state model
 * @param stateModelDefRef
 * @return
 */
public StateModelDefinition getStateModelDef(String stateModelDefRef) {
  if (stateModelDefRef == null) {
    return null;
  }
  return _stateModelDefMap.get(stateModelDefRef);
}
 

import org.apache.helix.model.StateModelDefinition; //导入依赖的package包/类
@Override
public void process(ClusterEvent event) throws Exception {
  ClusterDataCache cache = event.getAttribute(AttributeName.ClusterDataCache.name());
  Map<String, Resource> resourceMap = event.getAttribute(AttributeName.RESOURCES.name());
  CurrentStateOutput currentStateOutput =
      event.getAttribute(AttributeName.CURRENT_STATE.name());
  MessageGenerationOutput messageGenOutput =
      event.getAttribute(AttributeName.MESSAGES_ALL.name());
  if (cache == null || resourceMap == null || currentStateOutput == null
      || messageGenOutput == null) {
    throw new StageException("Missing attributes in event:" + event
        + ". Requires DataCache|RESOURCES|CURRENT_STATE|MESSAGES_ALL");
  }

  MessageSelectionStageOutput output = new MessageSelectionStageOutput();

  for (String resourceName : resourceMap.keySet()) {
    Resource resource = resourceMap.get(resourceName);
    StateModelDefinition stateModelDef = cache.getStateModelDef(resource.getStateModelDefRef());

    Map<String, Integer> stateTransitionPriorities = getStateTransitionPriorityMap(stateModelDef);
    IdealState idealState = cache.getIdealState(resourceName);
    Map<String, Bounds> stateConstraints =
        computeStateConstraints(stateModelDef, idealState, cache);
    for (Partition partition : resource.getPartitions()) {
      List<Message> messages = messageGenOutput.getMessages(resourceName, partition);
      List<Message> selectedMessages = selectMessages(cache.getLiveInstances(),
          currentStateOutput.getCurrentStateMap(resourceName, partition),
          currentStateOutput.getPendingMessageMap(resourceName, partition), messages,
          stateConstraints, stateTransitionPriorities, stateModelDef,
          resource.isP2PMessageEnabled());
      output.addMessages(resourceName, partition, selectedMessages);
    }
  }
  event.addAttribute(AttributeName.MESSAGES_SELECTED.name(), output);
}
 

import org.apache.helix.model.StateModelDefinition; //导入依赖的package包/类
/**
 * TODO: This code is duplicate in multiple places. Can we do it in to one place in the
 * beginning and compute the stateConstraint instance once and re use at other places.
 * Each IdealState must have a constraint object associated with it
 */
private Map<String, Bounds> computeStateConstraints(StateModelDefinition stateModelDefinition,
    IdealState idealState, ClusterDataCache cache) {
  Map<String, Bounds> stateConstraints = new HashMap<String, Bounds>();

  List<String> statePriorityList = stateModelDefinition.getStatesPriorityList();
  for (String state : statePriorityList) {
    String numInstancesPerState = stateModelDefinition.getNumInstancesPerState(state);
    int max = -1;
    if ("N".equals(numInstancesPerState)) {
      max = cache.getLiveInstances().size();
    } else if ("R".equals(numInstancesPerState)) {
      // idealState is null when resource has been dropped,
      // R can't be evaluated and ignore state constraints
      //if (idealState != null) {
      //  max = cache.getReplicas(idealState.getResourceName());
      //}
    } else {
      try {
        max = Integer.parseInt(numInstancesPerState);
      } catch (Exception e) {
        // use -1
      }
    }

    if (max > -1) {
      // if state has no constraint, will not put in map
      stateConstraints.put(state, new Bounds(0, max));
    }
  }

  return stateConstraints;
}
 

import org.apache.helix.model.StateModelDefinition; //导入依赖的package包/类
private Map<String, Integer> getStateTransitionPriorityMap(StateModelDefinition stateModelDef) {
  Map<String, Integer> stateTransitionPriorities = new HashMap<String, Integer>();
  List<String> stateTransitionPriorityList = stateModelDef.getStateTransitionPriorityList();
  for (int i = 0; i < stateTransitionPriorityList.size(); i++) {
    stateTransitionPriorities.put(stateTransitionPriorityList.get(i), i);
  }

  return stateTransitionPriorities;
}
 

import org.apache.helix.model.StateModelDefinition; //导入依赖的package包/类
/**
 * Compute the best state for all partitions.
 * This is the default implementation, subclasses should re-implement
 * this method if its logic to generate bestpossible map for each partition is different from the default one here.
 *
 * @param cache
 * @param idealState
 * @param resource
 * @param currentStateOutput Provides the current state and pending state transitions for all partitions
 * @return
 */
@Override
public ResourceAssignment computeBestPossiblePartitionState(ClusterDataCache cache,
    IdealState idealState, Resource resource, CurrentStateOutput currentStateOutput) {
  if (LOG.isDebugEnabled()) {
    LOG.debug("Processing resource:" + resource.getResourceName());
  }

  Set<String> allNodes = cache.getEnabledInstances();
  Set<String> liveNodes = cache.getLiveInstances().keySet();

  ClusterConfig clusterConfig = cache.getClusterConfig();
  long delayTime = getRebalanceDelay(idealState, clusterConfig);
  Set<String> activeNodes = getActiveInstances(allNodes, idealState, liveNodes,
      cache.getInstanceOfflineTimeMap(), cache.getLiveInstances().keySet(),
      cache.getInstanceConfigMap(), delayTime, clusterConfig);

  String stateModelDefName = idealState.getStateModelDefRef();
  StateModelDefinition stateModelDef = cache.getStateModelDef(stateModelDefName);
  ResourceAssignment partitionMapping = new ResourceAssignment(resource.getResourceName());
  for (Partition partition : resource.getPartitions()) {
    Map<String, String> currentStateMap =
        currentStateOutput.getCurrentStateMap(resource.getResourceName(), partition);
    Set<String> disabledInstancesForPartition =
        cache.getDisabledInstancesForPartition(resource.getResourceName(), partition.toString());
    List<String> preferenceList = getPreferenceList(partition, idealState, activeNodes);
    Map<String, String> bestStateForPartition =
        computeBestPossibleStateForPartition(liveNodes, stateModelDef, preferenceList, currentStateMap,
            disabledInstancesForPartition, idealState);

    partitionMapping.addReplicaMap(partition, bestStateForPartition);
  }

  if (LOG.isDebugEnabled()) {
    LOG.debug("Best possible mapping for resource  " + resource.getResourceName() + ": "
        + partitionMapping);
  }

  return partitionMapping;
}