import org.apache.helix.PropertyKey; //导入依赖的package包/类
private void refresh() {
  LOGGER.info("HelixBrokerRoutingTable.onExternalViewChange");

  PropertyKey externalViews = keyBuilder.externalViews();
  HelixDataAccessor helixDataAccessor = _helixManager.getHelixDataAccessor();
  List<ExternalView> externalViewList = helixDataAccessor.getChildValues(externalViews);

  List<InstanceConfig> instanceConfigList = helixDataAccessor.getChildValues(keyBuilder.instanceConfigs());

  Set<String> servingClusterList = getServingDataResource(externalViewList);
  for (ExternalView externalView : externalViewList) {
    String resourceName = externalView.getResourceName();
    if (servingClusterList.contains(resourceName)) {
      LOGGER.info("Trying to update ExternalView for data resource : " + resourceName + ", ExternalView: "
          + externalView);
      _helixExternalViewBasedRouting.markDataResourceOnline(
          resourceName,
          HelixHelper.getExternalViewForResouce(_helixManager.getClusterManagmentTool(),
              _helixManager.getClusterName(), resourceName), instanceConfigList);
    }
  }
}
 

import org.apache.helix.PropertyKey; //导入依赖的package包/类
public void removeListenersFromController(GenericHelixController controller) {
  PropertyKey.Builder keyBuilder = new PropertyKey.Builder(_manager.getClusterName());
  /**
   * reset generic-controller
   */
  _manager.removeListener(keyBuilder.instanceConfigs(), controller);
  _manager.removeListener(keyBuilder.resourceConfigs(), controller);
  _manager.removeListener(keyBuilder.liveInstances(), controller);
  _manager.removeListener(keyBuilder.idealStates(), controller);
  _manager.removeListener(keyBuilder.controller(), controller);

  /**
   * reset controller message listener and unregister all message handlers
   */
  _manager.removeListener(keyBuilder.controllerMessages(), _messagingService.getExecutor());
}
 

import org.apache.helix.PropertyKey; //导入依赖的package包/类
public ParticipantManager(HelixManager manager, ZkClient zkclient, int sessionTimeout,
    LiveInstanceInfoProvider liveInstanceInfoProvider, List<PreConnectCallback> preConnectCallbacks) {
  _zkclient = zkclient;
  _manager = manager;
  _clusterName = manager.getClusterName();
  _instanceName = manager.getInstanceName();
  _keyBuilder = new PropertyKey.Builder(_clusterName);
  _sessionId = manager.getSessionId();
  _sessionTimeout = sessionTimeout;
  _configAccessor = manager.getConfigAccessor();
  _instanceType = manager.getInstanceType();
  _helixAdmin = manager.getClusterManagmentTool();
  _dataAccessor = (ZKHelixDataAccessor) manager.getHelixDataAccessor();
  _messagingService = (DefaultMessagingService) manager.getMessagingService();
  _stateMachineEngine = manager.getStateMachineEngine();
  _liveInstanceInfoProvider = liveInstanceInfoProvider;
  _preConnectCallbacks = preConnectCallbacks;
}
 

import org.apache.helix.PropertyKey; //导入依赖的package包/类
@Override
public boolean setInstanceConfig(String clusterName, String instanceName,
    InstanceConfig newInstanceConfig) {
  String instanceConfigPath = PropertyPathBuilder.getPath(PropertyType.CONFIGS, clusterName,
      HelixConfigScope.ConfigScopeProperty.PARTICIPANT.toString(), instanceName);
  if (!_zkClient.exists(instanceConfigPath)) {
    throw new HelixException(
        "instance" + instanceName + " does not exist in cluster " + clusterName);
  }

  HelixDataAccessor accessor =
      new ZKHelixDataAccessor(clusterName, new ZkBaseDataAccessor<ZNRecord>(_zkClient));
  PropertyKey instanceConfigPropertyKey = accessor.keyBuilder().instanceConfig(instanceName);
  InstanceConfig currentInstanceConfig = accessor.getProperty(instanceConfigPropertyKey);
  if (!newInstanceConfig.getHostName().equals(currentInstanceConfig.getHostName())
      || !newInstanceConfig.getPort().equals(currentInstanceConfig.getPort())) {
    throw new HelixException(
        "Hostname and port cannot be changed, current hostname: " + currentInstanceConfig
            .getHostName() + " and port: " + currentInstanceConfig.getPort()
            + " is different from new hostname: " + newInstanceConfig.getHostName()
            + "and new port: " + newInstanceConfig.getPort());
  }
  return accessor.setProperty(instanceConfigPropertyKey, newInstanceConfig);
}
 

import org.apache.helix.PropertyKey; //导入依赖的package包/类
public static Map<String, String> readZkChildrenAsBytesMap(ZkClient zkclient, PropertyKey propertyKey) {
  BaseDataAccessor<byte[]> baseAccessor = new ZkBaseDataAccessor<byte[]>(zkclient);
  String parentPath = propertyKey.getPath();
  List<String> childNames = baseAccessor.getChildNames(parentPath, 0);
  if (childNames == null) {
    return null;
  }
  List<String> paths = new ArrayList<String>();
  for (String childName : childNames) {
    paths.add(parentPath + "/" + childName);
  }
  List<byte[]> values = baseAccessor.get(paths, null, 0);
  Map<String, String> ret = new HashMap<String, String>();
  for (int i = 0; i < childNames.size(); i++) {
    ret.put(childNames.get(i), new String(values.get(i)));
  }
  return ret;
}
 

import org.apache.helix.PropertyKey; //导入依赖的package包/类
@Override
public HelixProperty.Stat getPropertyStat(PropertyKey key) {
  PropertyType type = key.getType();
  String path = key.getPath();
  int options = constructOptions(type);
  try {
    Stat stat = _baseDataAccessor.getStat(path, options);
    if (stat != null) {
      return new HelixProperty.Stat(stat.getVersion(), stat.getCtime(), stat.getMtime());
    }
  } catch (ZkNoNodeException e) {

  }

  return null;
}
 

import org.apache.helix.PropertyKey; //导入依赖的package包/类
@Override
public List<HelixProperty.Stat> getPropertyStats(List<PropertyKey> keys) {
  if (keys == null || keys.size() == 0) {
    return Collections.emptyList();
  }

  List<HelixProperty.Stat> propertyStats = new ArrayList<>(keys.size());
  List<String> paths = new ArrayList<>(keys.size());
  for (PropertyKey key : keys) {
    paths.add(key.getPath());
  }
  Stat[] zkStats = _baseDataAccessor.getStats(paths, 0);

  for (int i = 0; i < keys.size(); i++) {
    Stat zkStat = zkStats[i];
    HelixProperty.Stat propertyStat = null;
    if (zkStat != null) {
      propertyStat =
          new HelixProperty.Stat(zkStat.getVersion(), zkStat.getCtime(), zkStat.getMtime());
    }
    propertyStats.add(propertyStat);
  }

  return propertyStats;
}
 

import org.apache.helix.PropertyKey; //导入依赖的package包/类
private void updateOfflineInstanceHistory(HelixDataAccessor accessor) {
  List<String> offlineNodes = new ArrayList<>(_instanceConfigMap.keySet());
  offlineNodes.removeAll(_liveInstanceMap.keySet());
  _instanceOfflineTimeMap = new HashMap<>();

  for (String instance : offlineNodes) {
    Builder keyBuilder = accessor.keyBuilder();
    PropertyKey propertyKey = keyBuilder.participantHistory(instance);
    ParticipantHistory history = accessor.getProperty(propertyKey);
    if (history == null) {
      history = new ParticipantHistory(instance);
    }
    if (history.getLastOfflineTime() == ParticipantHistory.ONLINE) {
      history.reportOffline();
      // persist history back to ZK.
      if (!accessor.setProperty(propertyKey, history)) {
        LOG.error("Fails to persist participant online history back to ZK!");
      }
    }
    _instanceOfflineTimeMap.put(instance, history.getLastOfflineTime());
  }
  _updateInstanceOfflineTime = false;
}
 

import org.apache.helix.PropertyKey; //导入依赖的package包/类
private void syncSessionToController(HelixManager manager) {
  if (_lastSessionSyncTime == null ||
          System.currentTimeMillis() - _lastSessionSyncTime > SESSION_SYNC_INTERVAL) { // > delay since last sync
    HelixDataAccessor accessor = manager.getHelixDataAccessor();
    PropertyKey key = new Builder(manager.getClusterName()).controllerMessage(SESSION_SYNC);
    if (accessor.getProperty(key) == null) {
      LOG.info(String.format("Participant %s syncs session with controller", manager.getInstanceName()));
      Message msg = new Message(MessageType.PARTICIPANT_SESSION_CHANGE, SESSION_SYNC);
      msg.setSrcName(manager.getInstanceName());
      msg.setTgtSessionId("*");
      msg.setMsgState(MessageState.NEW);
      msg.setMsgId(SESSION_SYNC);

      Criteria cr = new Criteria();
      cr.setRecipientInstanceType(InstanceType.CONTROLLER);
      cr.setSessionSpecific(false);

      manager.getMessagingService().send(cr, msg);
      _lastSessionSyncTime = System.currentTimeMillis();
    }
  }
}
 

import org.apache.helix.PropertyKey; //导入依赖的package包/类
/**
 * Request a state change for a specific task.
 *
 * @param accessor  connected Helix data accessor
 * @param instance  the instance serving the task
 * @param sessionId the current session of the instance
 * @param resource  the job name
 * @param partition the task partition name
 * @param state     the requested state
 * @return true if the request was persisted, false otherwise
 */
private static boolean setRequestedState(HelixDataAccessor accessor, String instance,
    String sessionId, String resource, String partition, TaskPartitionState state) {
  LOG.debug(
      String.format("Requesting a state transition to %s for partition %s.", state, partition));
  try {
    PropertyKey.Builder keyBuilder = accessor.keyBuilder();
    PropertyKey key = keyBuilder.currentState(instance, sessionId, resource);
    CurrentState currStateDelta = new CurrentState(resource);
    currStateDelta.setRequestedState(partition, state.name());

    return accessor.updateProperty(key, currStateDelta);
  } catch (Exception e) {
    LOG.error(String
        .format("Error when requesting a state transition to %s for partition %s.", state,
            partition), e);
    return false;
  }
}
 

import org.apache.helix.PropertyKey; //导入依赖的package包/类
private void setupCluster(String clusterName, List<String> instanceNames, String dbName,
    int replica, int partitions, int bucketSize) {
  _gSetupTool.addCluster(clusterName, true);
  _gSetupTool.addInstancesToCluster(clusterName,
      instanceNames.toArray(new String[instanceNames.size()]));

  // add a bucketized resource
  ZKHelixDataAccessor accessor = new ZKHelixDataAccessor(clusterName, _baseAccessor);
  PropertyKey.Builder keyBuilder = accessor.keyBuilder();
  ZNRecord idealStateRec =
      DefaultIdealStateCalculator.calculateIdealState(instanceNames, partitions, replica - 1,
          dbName,
          "MASTER", "SLAVE");
  IdealState idealState = new IdealState(idealStateRec);
  idealState.setBucketSize(bucketSize);
  idealState.setStateModelDefRef("MasterSlave");
  idealState.setRebalanceMode(IdealState.RebalanceMode.CUSTOMIZED);
  idealState.setReplicas(Integer.toString(replica));
  accessor.setProperty(keyBuilder.idealStates(dbName), idealState);

}
 

import org.apache.helix.PropertyKey; //导入依赖的package包/类
@Test()
public void testControllerLeaderHistory() throws Exception {
  HelixManager manager = HelixManagerFactory
      .getZKHelixManager(CLUSTER_NAME, "admin", InstanceType.ADMINISTRATOR, ZK_ADDR);
  manager.connect();

  PropertyKey.Builder keyBuilder = new PropertyKey.Builder(CLUSTER_NAME);
  PropertyKey propertyKey = keyBuilder.controllerLeaderHistory();
  LeaderHistory leaderHistory = manager.getHelixDataAccessor().getProperty(propertyKey);
  Assert.assertNotNull(leaderHistory);
  List<String> list = leaderHistory.getRecord().getListField("HISTORY");
  Assert.assertEquals(list.size(), 1);

  for (int i = 0; i <= 12; i++) {
    _controller.syncStop();
    _controller = new ClusterControllerManager(ZK_ADDR, CLUSTER_NAME, "Controller-" + i);
    _controller.syncStart();
  }

  leaderHistory = manager.getHelixDataAccessor().getProperty(propertyKey);
  Assert.assertNotNull(leaderHistory);
  list = leaderHistory.getRecord().getListField("HISTORY");
  Assert.assertEquals(list.size(), 10);
  manager.disconnect();
}
 

import org.apache.helix.PropertyKey; //导入依赖的package包/类
/**
 * trigger dummy message handler and verify it's invoked
 * @param manager
 * @param handledMsgSet
 * @throws Exception
 */
private static void checkDummyMsgHandler(HelixManager manager,
    final Set<String> handledMsgSet) throws Exception {

  final Message aMsg = newMsg();
  HelixDataAccessor accessor = manager.getHelixDataAccessor();
  PropertyKey.Builder keyBuilder = accessor.keyBuilder();
  accessor.setProperty(keyBuilder.message(manager.getInstanceName(), aMsg.getId()), aMsg);
  boolean result = TestHelper.verify(new TestHelper.Verifier() {

    @Override
    public boolean verify() throws Exception {

      return handledMsgSet.contains(aMsg.getId());
    }
  }, 5 * 1000);
  Assert.assertTrue(result);
}
 

import org.apache.helix.PropertyKey; //导入依赖的package包/类
@Test() public void testInstanceHistory() throws Exception {
  HelixManager manager = HelixManagerFactory
      .getZKHelixManager(CLUSTER_NAME, "admin", InstanceType.ADMINISTRATOR, ZK_ADDR);
  manager.connect();

  PropertyKey.Builder keyBuilder = new PropertyKey.Builder(CLUSTER_NAME);
  PropertyKey propertyKey = keyBuilder.participantHistory(_participants[0].getInstanceName());
  ParticipantHistory history = manager.getHelixDataAccessor().getProperty(propertyKey);
  Assert.assertNotNull(history);
  List<String> list = history.getRecord().getListField("HISTORY");
  Assert.assertEquals(list.size(), 1);

  for (int i = 0; i <= 22; i++) {
    _participants[0].disconnect();
    _participants[0].connect();
  }

  history = manager.getHelixDataAccessor().getProperty(propertyKey);
  Assert.assertNotNull(history);
  list = history.getRecord().getListField("HISTORY");
  Assert.assertEquals(list.size(), 20);
  list = history.getRecord().getListField("OFFLINE");
  Assert.assertEquals(list.size(), 20);
  manager.disconnect();
}
 

import org.apache.helix.PropertyKey; //导入依赖的package包/类
StringRepresentation getInstanceCurrentStateRepresentation(String clusterName,
    String instanceName, String resourceGroup) throws JsonGenerationException,
    JsonMappingException, IOException {
  ZkClient zkClient = (ZkClient) getRequest().getAttributes().get(RestAdminApplication.ZKCLIENT);
  String instanceSessionId =
      ClusterRepresentationUtil.getInstanceSessionId(zkClient, clusterName, instanceName);
  Builder keyBuilder = new PropertyKey.Builder(clusterName);

  String message =
      ClusterRepresentationUtil.getInstancePropertyAsString(zkClient, clusterName,
          keyBuilder.currentState(instanceName, instanceSessionId, resourceGroup),
          MediaType.APPLICATION_JSON);
  StringRepresentation representation =
      new StringRepresentation(message, MediaType.APPLICATION_JSON);
  return representation;
}
 

import org.apache.helix.PropertyKey; //导入依赖的package包/类
private HelixProperty pollForError(HelixDataAccessor accessor, PropertyKey key) {
  final int POLL_TIMEOUT = 5000;
  final int POLL_INTERVAL = 100;
  HelixProperty property = accessor.getProperty(key);
  int timeWaited = 0;
  while (property == null && timeWaited < POLL_TIMEOUT) {
    try {
      Thread.sleep(POLL_INTERVAL);
    } catch (InterruptedException e) {
      return null;
    }
    timeWaited += POLL_INTERVAL;
    property = accessor.getProperty(key);
  }
  return property;
}
 

import org.apache.helix.PropertyKey; //导入依赖的package包/类
StringRepresentation getInstanceRepresentation() throws JsonGenerationException,
    JsonMappingException, IOException {
  String clusterName =
      ResourceUtil.getAttributeFromRequest(getRequest(), ResourceUtil.RequestKey.CLUSTER_NAME);
  String instanceName =
      ResourceUtil.getAttributeFromRequest(getRequest(), ResourceUtil.RequestKey.INSTANCE_NAME);
  Builder keyBuilder = new PropertyKey.Builder(clusterName);
  ZkClient zkclient =
      ResourceUtil.getAttributeFromCtx(getContext(), ResourceUtil.ContextKey.RAW_ZKCLIENT);

  String instanceCfgStr =
      ResourceUtil.readZkAsBytes(zkclient, keyBuilder.instanceConfig(instanceName));
  StringRepresentation representation =
      new StringRepresentation(instanceCfgStr, MediaType.APPLICATION_JSON);

  return representation;
}
 

import org.apache.helix.PropertyKey; //导入依赖的package包/类
StringRepresentation getInstanceStatusUpdateRepresentation(String clusterName,
    String instanceName, String resourceGroup) throws JsonGenerationException,
    JsonMappingException, IOException {
  ZkClient zkClient = (ZkClient) getContext().getAttributes().get(RestAdminApplication.ZKCLIENT);

  String instanceSessionId =
      ClusterRepresentationUtil.getInstanceSessionId(zkClient, clusterName, instanceName);

  Builder keyBuilder = new PropertyKey.Builder(clusterName);
  String message =
      ClusterRepresentationUtil.getInstancePropertiesAsString(zkClient, clusterName,
          keyBuilder.stateTransitionStatus(instanceName, instanceSessionId, resourceGroup),
          // instanceSessionId
          // + "__"
          // + resourceGroup,
          MediaType.APPLICATION_JSON);
  StringRepresentation representation =
      new StringRepresentation(message, MediaType.APPLICATION_JSON);
  return representation;
}
 

import org.apache.helix.PropertyKey; //导入依赖的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.PropertyKey; //导入依赖的package包/类
private ParticipantHistory getHistory() {
  PropertyKey propertyKey = _keyBuilder.participantHistory(_instanceName);
  ParticipantHistory history = _dataAccessor.getProperty(propertyKey);
  if (history == null) {
    history = new ParticipantHistory(_instanceName);
  }
  return history;
}