import org.apache.helix.model.LiveInstance; //导入依赖的package包/类
@Override
public void onLiveInstanceChange(List<LiveInstance> list, NotificationContext notificationContext) {
  List<InetSocketAddress> addresses = new ArrayList<>();
  HelixAdmin helixAdmin = helixConnection.createClusterManagementTool();
  for (LiveInstance liveInstance : list) {
    InstanceConfig instanceConfig = helixAdmin.getInstanceConfig(
        clusterName,
        liveInstance.getInstanceName());
    InetSocketAddress address = new InetSocketAddress(
        instanceConfig.getHostName(),
        Integer.valueOf(instanceConfig.getPort()));
    addresses.add(address);
  }

  services.set(addresses);
}
 

import org.apache.helix.model.LiveInstance; //导入依赖的package包/类
@Override
public boolean isLeader() {
  if (_instanceType != InstanceType.CONTROLLER
      && _instanceType != InstanceType.CONTROLLER_PARTICIPANT) {
    return false;
  }

  if (!isConnected()) {
    return false;
  }

  try {
    LiveInstance leader = _dataAccessor.getProperty(_keyBuilder.controllerLeader());
    if (leader != null) {
      String leaderName = leader.getInstanceName();
      String sessionId = leader.getSessionId();
      if (leaderName != null && leaderName.equals(_instanceName) && sessionId != null
          && sessionId.equals(_sessionId)) {
        return true;
      }
    }
  } catch (Exception e) {
    // log
  }
  return false;
}
 

import org.apache.helix.model.LiveInstance; //导入依赖的package包/类
@Override
public void process(ClusterEvent event) throws Exception {
  HelixManager manager = event.getAttribute(AttributeName.helixmanager.name());
  ClusterDataCache cache = event.getAttribute(AttributeName.ClusterDataCache.name());
  if (manager == null || cache == null) {
    throw new StageException("Missing attributes in event:" + event
        + ". Requires HelixManager | DataCache");
  }

  HelixManagerProperties properties = manager.getProperties();
  Map<String, LiveInstance> liveInstanceMap = cache.getLiveInstances();
  for (LiveInstance liveInstance : liveInstanceMap.values()) {
    String participantVersion = liveInstance.getHelixVersion();
    if (!properties.isParticipantCompatible(participantVersion)) {
      String errorMsg =
          "incompatible participant. pipeline will not continue. " + "controller: "
              + manager.getInstanceName() + ", controllerVersion: " + properties.getVersion()
              + ", minimumSupportedParticipantVersion: "
              + properties.getProperty("miminum_supported_version.participant")
              + ", participant: " + liveInstance.getInstanceName() + ", participantVersion: "
              + participantVersion;
      LOG.error(errorMsg);
      throw new StageException(errorMsg);
    }
  }
}
 

import org.apache.helix.model.LiveInstance; //导入依赖的package包/类
private void verifyP2PMessage(String dbName, String instance, String expectedState, String expectedTriggerHost) {
  ClusterDataCache dataCache = new ClusterDataCache(CLUSTER_NAME);
  dataCache.refresh(_accessor);

  Map<String, LiveInstance> liveInstanceMap = dataCache.getLiveInstances();
  LiveInstance liveInstance = liveInstanceMap.get(instance);

  Map<String, CurrentState> currentStateMap = dataCache.getCurrentState(instance, liveInstance.getSessionId());
  Assert.assertNotNull(currentStateMap);
  CurrentState currentState = currentStateMap.get(dbName);
  Assert.assertNotNull(currentState);
  Assert.assertEquals(currentState.getPartitionStateMap().size(), PARTITION_NUMBER);

  for (String partition : currentState.getPartitionStateMap().keySet()) {
    String state = currentState.getState(partition);
    Assert.assertEquals(state, expectedState,
        dbName + " Partition " + partition + "'s state is different as expected!");
    String triggerHost = currentState.getTriggerHost(partition);
    Assert.assertEquals(triggerHost, expectedTriggerHost,
        "Partition " + partition + "'s transition to Master was not triggered by expected host!");
  }
}
 

import org.apache.helix.model.LiveInstance; //导入依赖的package包/类
@Test public void testAddedFieldsInCurrentState() {
  String instanceName = PARTICIPANT_PREFIX + "_" + _startPort;
  HelixDataAccessor accessor = _manager.getHelixDataAccessor();
  LiveInstance liveInstance =
      accessor.getProperty(accessor.keyBuilder().liveInstance(instanceName));
  CurrentState currentState = accessor.getProperty(accessor.keyBuilder()
      .currentState(instanceName, liveInstance.getSessionId(), WorkflowGenerator.DEFAULT_TGT_DB));
  // Test start time should happen after test start time
  Assert.assertTrue(
      currentState.getStartTime(WorkflowGenerator.DEFAULT_TGT_DB + "_0") >= _testStartTime);

  // Test end time is always larger than start time
  Assert.assertTrue(
      currentState.getEndTime(WorkflowGenerator.DEFAULT_TGT_DB + "_0") >= currentState
          .getStartTime(WorkflowGenerator.DEFAULT_TGT_DB + "_0"));

  // Final state is MASTER, so SLAVE will be the previous state
  Assert.assertEquals(currentState.getPreviousState(WorkflowGenerator.DEFAULT_TGT_DB + "_0"),
      "SLAVE");
}
 

import org.apache.helix.model.LiveInstance; //导入依赖的package包/类
private Map<String, Map<String, String>> getMapping(final Map<String, List<String>> listResult) {
  final Map<String, Map<String, String>> mapResult = new HashMap<String, Map<String, String>>();
  ClusterDataCache cache = new ClusterDataCache();
  MockAccessor accessor = new MockAccessor();
  Builder keyBuilder = accessor.keyBuilder();
  for (String node : _liveNodes) {
    LiveInstance liveInstance = new LiveInstance(node);
    liveInstance.setSessionId("testSession");
    accessor.setProperty(keyBuilder.liveInstance(node), liveInstance);
  }
  cache.refresh(accessor);

  IdealState is = new IdealState("resource");
  for (String partition : _partitions) {
    List<String> preferenceList = listResult.get(partition);
    Map<String, String> currentStateMap = _currentMapping.get(partition);
    Set<String> disabled = Collections.emptySet();
    Map<String, String> assignment = new AutoRebalancer()
        .computeBestPossibleStateForPartition(cache.getLiveInstances().keySet(), _stateModelDef, preferenceList,
            currentStateMap, disabled, is);
    mapResult.put(partition, assignment);
  }
  return mapResult;
}
 

import org.apache.helix.model.LiveInstance; //导入依赖的package包/类
@GET
@Path("{clusterId}/controller")
public Response getClusterController(@PathParam("clusterId") String clusterId) {
  HelixDataAccessor dataAccessor = getDataAccssor(clusterId);
  Map<String, Object> controllerInfo = new HashMap<>();
  controllerInfo.put(Properties.id.name(), clusterId);

  LiveInstance leader = dataAccessor.getProperty(dataAccessor.keyBuilder().controllerLeader());
  if (leader != null) {
    controllerInfo.put(ClusterProperties.controller.name(), leader.getInstanceName());
    controllerInfo.putAll(leader.getRecord().getSimpleFields());
  } else {
    controllerInfo.put(ClusterProperties.controller.name(), "No Lead Controller!");
  }

  return JSONRepresentation(controllerInfo);
}
 

import org.apache.helix.model.LiveInstance; //导入依赖的package包/类
StringRepresentation getSchedulerTasksRepresentation() throws JsonGenerationException,
    JsonMappingException, IOException {
  String clusterName = (String) getRequest().getAttributes().get("clusterName");
  String instanceName = (String) getRequest().getAttributes().get("instanceName");
  ZkClient zkClient = (ZkClient) getContext().getAttributes().get(RestAdminApplication.ZKCLIENT);
  ClusterSetup setupTool = new ClusterSetup(zkClient);
  List<String> instances =
      setupTool.getClusterManagementTool().getInstancesInCluster(clusterName);

  HelixDataAccessor accessor =
      ClusterRepresentationUtil.getClusterDataAccessor(zkClient, clusterName);
  LiveInstance liveInstance =
      accessor.getProperty(accessor.keyBuilder().liveInstance(instanceName));
  String sessionId = liveInstance.getSessionId();

  StringRepresentation representation = new StringRepresentation("");// (ClusterRepresentationUtil.ObjectToJson(instanceConfigs),
                                                                     // MediaType.APPLICATION_JSON);

  return representation;
}
 

import org.apache.helix.model.LiveInstance; //导入依赖的package包/类
/**
 * Check if an instance can safely dropped from helix cluster. Instance should not be dropped if:
 * - It is a live instance.
 * - Any idealstate includes the instance.
 *
 * @param instanceName: Name of the instance to be dropped.
 * @return
 */
public boolean isInstanceDroppable(String instanceName) {
  // Check if this instance is live
  HelixDataAccessor helixDataAccessor = _helixZkManager.getHelixDataAccessor();
  LiveInstance liveInstance = helixDataAccessor.getProperty(_keyBuilder.liveInstance(instanceName));
  if (liveInstance != null) {
    return false;
  }

  // Check if any idealstate contains information on this instance
  for (String resourceName : getAllResources()) {
    IdealState resourceIdealState = _helixAdmin.getResourceIdealState(_helixClusterName, resourceName);
    for (String partition : resourceIdealState.getPartitionSet()) {
      for (String instance : resourceIdealState.getInstanceSet(partition)) {
        if (instance.equals(instanceName)) {
          return false;
        }
      }
    }
  }
  return true;
}
 

import org.apache.helix.model.LiveInstance; //导入依赖的package包/类
/**
 * Update the liveness states of existing instances based on the input.
 * @param liveInstances the list of instances that are up.
 */
private void updateInstanceLiveness(List<LiveInstance> liveInstances) {
  synchronized (notificationLock) {
    Set<String> liveInstancesSet = new HashSet<>();
    for (LiveInstance liveInstance : liveInstances) {
      liveInstancesSet.add(liveInstance.getInstanceName());
    }
    for (String instanceName : allInstances) {
      if (liveInstancesSet.contains(instanceName)) {
        instanceNameToAmbryDataNode.get(instanceName).setState(HardwareState.AVAILABLE);
      } else {
        instanceNameToAmbryDataNode.get(instanceName).setState(HardwareState.UNAVAILABLE);
      }
    }
  }
}
 

import org.apache.helix.model.LiveInstance; //导入依赖的package包/类
@Override
public void onLiveInstanceChange(final List<LiveInstance> liveInstances,
    NotificationContext changeContext) {
  LOGGER.info("AutoRebalanceLiveInstanceChangeListener.onLiveInstanceChange() wakes up!");
  _delayedScheuler.schedule(new Runnable() {
    @Override
    public void run() {
      try {
        rebalanceCurrentCluster(_helixMirrorMakerManager.getCurrentLiveInstances());
      } catch (Exception e) {
        LOGGER.error("Got exception during rebalance the whole cluster! ", e);
      }
    }
  }, _delayedAutoReblanceTimeInSeconds, TimeUnit.SECONDS);
}
 

import org.apache.helix.model.LiveInstance; //导入依赖的package包/类
@SuppressWarnings("unused")
private static boolean isAnyWorkerDown(List<LiveInstance> liveInstances,
    Map<String, Set<TopicPartition>> instanceToTopicPartitionMap) {
  Set<String> removedInstances =
      getRemovedInstanceSet(getLiveInstanceName(liveInstances),
          instanceToTopicPartitionMap.keySet());
  return !removedInstances.isEmpty();
}
 

import org.apache.helix.model.LiveInstance; //导入依赖的package包/类
private static Set<String> getLiveInstanceName(List<LiveInstance> liveInstances) {
  Set<String> liveInstanceNames = new HashSet<String>();
  for (LiveInstance liveInstance : liveInstances) {
    liveInstanceNames.add(liveInstance.getInstanceName());
  }
  return liveInstanceNames;
}
 

import org.apache.helix.model.LiveInstance; //导入依赖的package包/类
@Override
public synchronized void onLiveInstanceChange(final List<LiveInstance> liveInstances,
    NotificationContext changeContext) {
  LOGGER.info("AutoRebalanceLiveInstanceChangeListener.onLiveInstanceChange() wakes up!");
  _delayedScheuler.schedule(new Runnable() {
    @Override
    public void run() {
      try {
        rebalanceCurrentCluster(_helixMirrorMakerManager.getCurrentLiveInstances()); 
      } catch (Exception e) {
        LOGGER.error("Got exception during rebalance the whole cluster! ", e);
      }
    }
  }, _delayedAutoReblanceTimeInSeconds, TimeUnit.SECONDS);
}
 

import org.apache.helix.model.LiveInstance; //导入依赖的package包/类
private String findNewMaster(Map<String, LiveInstance> liveInstancesMap,
    Map<String, InstanceConfig> instanceConfigs, String prevMasterHost, Set<String> slaveSet) {
  String newMaster = null;
  SlaveStatus slaveStatusOfNewMaster = null;
  for (String slave : slaveSet) {
    if (liveInstancesMap.containsKey(slave) && instanceConfigs.get(slave).getInstanceEnabled()) {
      MySQLAdmin admin = new MySQLAdmin(instanceConfigs.get(slave));
      SlaveStatus slaveStatus = admin.getSlaveStatus();
      if (slaveStatus != null) {
        String masterHost = slaveStatus.getString(SlaveStatusAttribute.Master_Host);
        String masterPort = slaveStatus.getString(SlaveStatusAttribute.Master_Port);
        String slaveIOStatus = slaveStatus.getString(SlaveStatusAttribute.Slave_IO_Running);
        String slaveSQLStatus = slaveStatus.getString(SlaveStatusAttribute.Slave_SQL_Running);
        String masterLogFile = slaveStatus.getString(SlaveStatusAttribute.Master_Log_File);
        int readMasterLogPos = slaveStatus.getInt(SlaveStatusAttribute.Read_Master_Log_Pos);
        if ((masterHost + "_" + masterPort).equals(prevMasterHost)
            && "YES".equalsIgnoreCase(slaveIOStatus) && "YES".equalsIgnoreCase(slaveSQLStatus)) {
          if (newMaster == null) {
            newMaster = slave;
            slaveStatusOfNewMaster = slaveStatus;
          } else {
            boolean isThisSlaveAhead =
                masterLogFile.compareTo(slaveStatusOfNewMaster
                    .getString(SlaveStatusAttribute.Master_Log_File)) > 0
                    && readMasterLogPos > slaveStatusOfNewMaster
                        .getInt(SlaveStatusAttribute.Read_Master_Log_Pos);
            if (isThisSlaveAhead) {
              newMaster = slave;
              slaveStatusOfNewMaster = slaveStatus;
            }
          }
        }
      }
    }
  }
  return newMaster;
}
 

import org.apache.helix.model.LiveInstance; //导入依赖的package包/类
public RoutingTable(Collection<ExternalView> externalViews, Collection<InstanceConfig> instanceConfigs,
    Collection<LiveInstance> liveInstances) {
  _resourceInfoMap = new HashMap<>();
  _resourceGroupInfoMap = new HashMap<>();
  _liveInstances = liveInstances;
  _instanceConfigs = instanceConfigs;
  refresh(externalViews, instanceConfigs);
}
 

import org.apache.helix.model.LiveInstance; //导入依赖的package包/类
@Override
@PreFetch(enabled = false)
public void onLiveInstanceChange(List<LiveInstance> liveInstances,
    NotificationContext changeContext) {
  _routerUpdater.queueEvent(changeContext, ClusterEventType.LiveInstanceChange,
      HelixConstants.ChangeType.LIVE_INSTANCE);
}
 

import org.apache.helix.model.LiveInstance; //导入依赖的package包/类
public void refresh(List<ExternalView> externalViewList, NotificationContext changeContext) {
  HelixDataAccessor accessor = changeContext.getManager().getHelixDataAccessor();
  PropertyKey.Builder keyBuilder = accessor.keyBuilder();

  List<InstanceConfig> configList = accessor.getChildValues(keyBuilder.instanceConfigs());
  List<LiveInstance> liveInstances = accessor.getChildValues(keyBuilder.liveInstances());
  refresh(externalViewList, configList, liveInstances);
}
 

import org.apache.helix.model.LiveInstance; //导入依赖的package包/类
private boolean tryUpdateController(HelixManager manager) {
  HelixDataAccessor accessor = manager.getHelixDataAccessor();
  Builder keyBuilder = accessor.keyBuilder();

  LiveInstance leader = new LiveInstance(manager.getInstanceName());
  try {
    leader.setLiveInstance(ManagementFactory.getRuntimeMXBean().getName());
    leader.setSessionId(manager.getSessionId());
    leader.setHelixVersion(manager.getVersion());
    boolean success = accessor.createControllerLeader(leader);
    if (success) {
      return true;
    } else {
      LOG.info("Unable to become leader probably because some other controller becames the leader");
    }
  } catch (Exception e) {
    LOG.error(
        "Exception when trying to updating leader record in cluster:" + manager.getClusterName()
            + ". Need to check again whether leader node has been created or not", e);
  }

  leader = accessor.getProperty(keyBuilder.controllerLeader());
  if (leader != null) {
    String leaderSessionId = leader.getSessionId();
    LOG.info("Leader exists for cluster: " + manager.getClusterName() + ", currentLeader: "
        + leader.getInstanceName() + ", leaderSessionId: " + leaderSessionId);

    if (leaderSessionId != null && leaderSessionId.equals(manager.getSessionId())) {
      return true;
    }
  }
  return false;
}
 

import org.apache.helix.model.LiveInstance; //导入依赖的package包/类
@Override
public void onLiveInstanceChange(List<LiveInstance> liveInstances,
    NotificationContext changeContext) {
  logger.info("START: Generic GenericClusterController.onLiveInstanceChange() for cluster " + _clusterName);
  if (changeContext == null || changeContext.getType() != Type.CALLBACK) {
    _cache.requireFullRefresh();
    _taskCache.requireFullRefresh();
  }

  if (liveInstances == null) {
    liveInstances = Collections.emptyList();
  }
  _cache.setLiveInstances(liveInstances);
  _taskCache.setLiveInstances(liveInstances);

  // Go though the live instance list and make sure that we are observing them
  // accordingly. The action is done regardless of the paused flag.
  if (changeContext.getType() == NotificationContext.Type.INIT
      || changeContext.getType() == NotificationContext.Type.CALLBACK) {
    checkLiveInstancesObservation(liveInstances, changeContext);
  } else if (changeContext.getType() == NotificationContext.Type.FINALIZE) {
    // on finalize, should remove all message/current-state listeners
    logger.info("remove message/current-state listeners. lastSeenInstances: " + _lastSeenInstances
        + ", lastSeenSessions: " + _lastSeenSessions);
    liveInstances = Collections.emptyList();
    checkLiveInstancesObservation(liveInstances, changeContext);
  }

  pushToEventQueues(ClusterEventType.LiveInstanceChange, changeContext,
      Collections.<String, Object>singletonMap(AttributeName.eventData.name(), liveInstances));

  logger.info(
      "END: Generic GenericClusterController.onLiveInstanceChange() for cluster " + _clusterName);
}