import org.apache.helix.model.HelixConfigScope; //导入依赖的package包/类
/**
 * Get ClusterConfig of the given cluster.
 *
 * @param clusterName
 *
 * @return
 */
public ClusterConfig getClusterConfig(String clusterName) {
  if (!ZKUtil.isClusterSetup(clusterName, zkClient)) {
    throw new HelixException("fail to get config. cluster: " + clusterName + " is NOT setup.");
  }

  HelixConfigScope scope =
      new HelixConfigScopeBuilder(ConfigScopeProperty.CLUSTER).forCluster(clusterName).build();
  ZNRecord record = getConfigZnRecord(scope);

  if (record == null) {
    LOG.warn("No config found at " + scope.getZkPath());
    return null;
  }

  return new ClusterConfig(record);
}
 

import org.apache.helix.model.HelixConfigScope; //导入依赖的package包/类
private void updateResourceConfig(String clusterName, String resourceName,
    ResourceConfig resourceConfig, boolean overwrite) {
  if (!ZKUtil.isClusterSetup(clusterName, zkClient)) {
    throw new HelixException("fail to setup config. cluster: " + clusterName + " is NOT setup.");
  }

  HelixConfigScope scope =
      new HelixConfigScopeBuilder(ConfigScopeProperty.RESOURCE).forCluster(clusterName)
          .forResource(resourceName).build();
  String zkPath = scope.getZkPath();

  if (overwrite) {
    ZKUtil.createOrReplace(zkClient, zkPath, resourceConfig.getRecord(), true);
  } else {
    ZKUtil.createOrUpdate(zkClient, zkPath, resourceConfig.getRecord(), true, true);
  }
}
 

import org.apache.helix.model.HelixConfigScope; //导入依赖的package包/类
/**
 * Get instance config for given resource in given cluster.
 *
 * @param clusterName
 * @param instanceName
 *
 * @return
 */
public InstanceConfig getInstanceConfig(String clusterName, String instanceName) {
  if (!ZKUtil.isInstanceSetup(zkClient, clusterName, instanceName, InstanceType.PARTICIPANT)) {
    throw new HelixException(
        "fail to get config. instance: " + instanceName + " is NOT setup in cluster: "
            + clusterName);
  }

  HelixConfigScope scope =
      new HelixConfigScopeBuilder(ConfigScopeProperty.PARTICIPANT).forCluster(clusterName)
          .forParticipant(instanceName).build();
  ZNRecord record = getConfigZnRecord(scope);

  if (record == null) {
    LOG.warn("No config found at " + scope.getZkPath());
    return null;
  }

  return new InstanceConfig(record);
}
 

import org.apache.helix.model.HelixConfigScope; //导入依赖的package包/类
private void updateInstanceConfig(String clusterName, String instanceName,
    InstanceConfig instanceConfig, boolean overwrite) {
  if (!ZKUtil.isClusterSetup(clusterName, zkClient)) {
    throw new HelixException("fail to setup config. cluster: " + clusterName + " is NOT setup.");
  }

  HelixConfigScope scope =
      new HelixConfigScopeBuilder(ConfigScopeProperty.PARTICIPANT).forCluster(clusterName)
          .forParticipant(instanceName).build();
  String zkPath = scope.getZkPath();

  if (overwrite) {
    ZKUtil.createOrReplace(zkClient, zkPath, instanceConfig.getRecord(), true);
  } else {
    ZKUtil.createOrUpdate(zkClient, zkPath, instanceConfig.getRecord(), true, true);
  }
}
 

import org.apache.helix.model.HelixConfigScope; //导入依赖的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.model.HelixConfigScope; //导入依赖的package包/类
/**
 * get configs
 * @param type config-scope-type, e.g. CLUSTER, RESOURCE, etc.
 * @param scopeArgsCsv csv-formatted scope-args, e.g myCluster,testDB
 * @param keysCsv csv-formatted keys. e.g. k1,k2
 * @return json-formated key-value pairs, e.g. {k1=v1,k2=v2}
 */
public String getConfig(ConfigScopeProperty type, String scopeArgsCsv, String keysCsv) {
  // ConfigScope scope = new ConfigScopeBuilder().build(scopesStr);

  String[] scopeArgs = scopeArgsCsv.split("[\\s,]");
  HelixConfigScope scope = new HelixConfigScopeBuilder(type, scopeArgs).build();

  String[] keys = keysCsv.split("[\\s,]");
  // parse keys
  // String[] keys = keysStr.split("[\\s,]");
  // Set<String> keysSet = new HashSet<String>(Arrays.asList(keys));

  Map<String, String> keyValueMap = _admin.getConfig(scope, Arrays.asList(keys));
  ZNRecord record = new ZNRecord(type.toString());
  // record.setMapField(scopesStr, propertiesMap);
  record.getSimpleFields().putAll(keyValueMap);
  ZNRecordSerializer serializer = new ZNRecordSerializer();
  return new String(serializer.serialize(record));
}
 

import org.apache.helix.model.HelixConfigScope; //导入依赖的package包/类
@Transition(to = "ONLINE", from = "OFFLINE")
public void onBecomeOnlineFromOffline(Message message, NotificationContext context)
    throws Exception {
  LOG.debug(_workerId + " becomes ONLINE from OFFLINE for " + _partition);
  ConfigAccessor clusterConfig = context.getManager().getConfigAccessor();
  HelixManager manager = context.getManager();
  HelixConfigScope clusterScope =
      new HelixConfigScopeBuilder(ConfigScopeProperty.CLUSTER).forCluster(
          manager.getClusterName()).build();
  String json = clusterConfig.get(clusterScope, message.getResourceName());
  Dag.Node node = Dag.Node.fromJson(json);
  Set<String> parentIds = node.getParentIds();
  String resourceName = message.getResourceName();
  int numPartitions = node.getNumPartitions();
  Task task = _taskFactory.createTask(resourceName, parentIds, manager, _taskResultStore);
  manager.addExternalViewChangeListener(task);

  LOG.debug("Starting task for " + _partition + "...");
  int partitionNum = Integer.parseInt(_partition.split("_")[1]);
  task.execute(resourceName, numPartitions, partitionNum);
  LOG.debug("Task for " + _partition + " done");
}
 

import org.apache.helix.model.HelixConfigScope; //导入依赖的package包/类
private static void addConfiguration(ClusterSetup setup, String baseDir, String clusterName,
    String instanceName) throws IOException {
  Map<String, String> properties = new HashMap<String, String>();
  HelixConfigScopeBuilder builder = new HelixConfigScopeBuilder(ConfigScopeProperty.PARTICIPANT);
  HelixConfigScope instanceScope =
      builder.forCluster(clusterName).forParticipant(instanceName).build();
  properties.put("change_log_dir", baseDir + instanceName + "/translog");
  properties.put("file_store_dir", baseDir + instanceName + "/filestore");
  properties.put("check_point_dir", baseDir + instanceName + "/checkpoint");
  setup.getClusterManagementTool().setConfig(instanceScope, properties);
  FileUtils.deleteDirectory(new File(properties.get("change_log_dir")));
  FileUtils.deleteDirectory(new File(properties.get("file_store_dir")));
  FileUtils.deleteDirectory(new File(properties.get("check_point_dir")));
  new File(properties.get("change_log_dir")).mkdirs();
  new File(properties.get("file_store_dir")).mkdirs();
  new File(properties.get("check_point_dir")).mkdirs();
}
 

import org.apache.helix.model.HelixConfigScope; //导入依赖的package包/类
public static void createHelixClusterIfNeeded(String helixClusterName, String zkPath) {
  final HelixAdmin admin = new ZKHelixAdmin(zkPath);

  if (admin.getClusters().contains(helixClusterName)) {
    LOGGER.info(
        "cluster already exist, skipping it.. ********************************************* ");
    return;
  }

  LOGGER.info("Creating a new cluster, as the helix cluster : " + helixClusterName
      + " was not found ********************************************* ");
  admin.addCluster(helixClusterName, false);

  LOGGER.info("Enable mirror maker machines auto join.");
  final HelixConfigScope scope = new HelixConfigScopeBuilder(ConfigScopeProperty.CLUSTER)
      .forCluster(helixClusterName).build();

  final Map<String, String> props = new HashMap<String, String>();
  props.put(ZKHelixManager.ALLOW_PARTICIPANT_AUTO_JOIN, String.valueOf(true));
  props.put(MessageType.STATE_TRANSITION + "." + HelixTaskExecutor.MAX_THREADS,
      String.valueOf(100));

  admin.setConfig(scope, props);

  LOGGER.info("Adding state model definition named : OnlineOffline generated using : "
      + OnlineOfflineStateModel.class.toString()
      + " ********************************************** ");

  // add state model definition
  admin.addStateModelDef(helixClusterName, "OnlineOffline", OnlineOfflineStateModel.build());
  LOGGER.info("New Cluster setup completed... ********************************************** ");
}
 

import org.apache.helix.model.HelixConfigScope; //导入依赖的package包/类
public void disableAutoBalancing() {
  HelixConfigScope scope =
      new HelixConfigScopeBuilder(ConfigScopeProperty.CLUSTER).forCluster(_helixClusterName)
          .build();
  Map<String, String> properties = new HashMap<String, String>();
  properties.put(AUTO_BALANCING, DISABLE);
  _helixAdmin.setConfig(scope, properties);
}
 

import org.apache.helix.model.HelixConfigScope; //导入依赖的package包/类
public void enableAutoBalancing() {
  HelixConfigScope scope =
      new HelixConfigScopeBuilder(ConfigScopeProperty.CLUSTER).forCluster(_helixClusterName)
          .build();
  Map<String, String> properties = new HashMap<String, String>();
  properties.put(AUTO_BALANCING, ENABLE);
  _helixAdmin.setConfig(scope, properties);
}
 

import org.apache.helix.model.HelixConfigScope; //导入依赖的package包/类
public boolean isAutoBalancingEnabled() {
  HelixConfigScope scope =
      new HelixConfigScopeBuilder(ConfigScopeProperty.CLUSTER).forCluster(_helixClusterName)
          .build();
  Map<String, String> config = _helixAdmin.getConfig(scope, Arrays.asList(AUTO_BALANCING));
  if (config.containsKey(AUTO_BALANCING) && config.get(AUTO_BALANCING).equals(DISABLE)) {
    return false;
  }
  return true;
}
 

import org.apache.helix.model.HelixConfigScope; //导入依赖的package包/类
/**
 * Create a Helix cluster with the cluster name specified using
 * {@link GobblinYarnConfigurationKeys#HELIX_CLUSTER_NAME_KEY} if it does not exist.
 */
private void createGobblinYarnHelixCluster() {
  ClusterSetup clusterSetup =
      new ClusterSetup(this.config.getString(GobblinYarnConfigurationKeys.ZK_CONNECTION_STRING_KEY));
  String clusterName = this.config.getString(GobblinYarnConfigurationKeys.HELIX_CLUSTER_NAME_KEY);
  // Create the cluster and overwrite if it already exists
  clusterSetup.addCluster(clusterName, true);
  // Helix 0.6.x requires a configuration property to have the form key=value.
  String autoJoinConfig = ZKHelixManager.ALLOW_PARTICIPANT_AUTO_JOIN + "=true";
  clusterSetup.setConfig(HelixConfigScope.ConfigScopeProperty.CLUSTER, clusterName, autoJoinConfig);
  LOGGER.info("Created Helix cluster " + clusterName);
}
 

import org.apache.helix.model.HelixConfigScope; //导入依赖的package包/类
private void setUpHelixCluster(String zookeeperQuorum, String clusterName) {
  ZkClient zkClient = ZKClientPool.getZkClient(zookeeperQuorum);
  HelixAdmin helixAdmin = new ZKHelixAdmin(zkClient);
  try {
    if(!ImmutableSet.copyOf(helixAdmin.getClusters()).contains(clusterName)) {
      ClusterSetup helixClusterSetUp = new ClusterSetup(zkClient);
      helixClusterSetUp.addCluster(clusterName, false);
      helixClusterSetUp.setConfig(HelixConfigScope.ConfigScopeProperty.CLUSTER, clusterName,
          "allowParticipantAutoJoin=true");
    }
  } finally {
    zkClient.close();
  }
}
 

import org.apache.helix.model.HelixConfigScope; //导入依赖的package包/类
public static void deleteResourcePropertyFromHelix(HelixAdmin admin, String clusterName, String resourceName,
    String configKey) {
  final List<String> keys = new ArrayList<String>();
  keys.add(configKey);

  final HelixConfigScope scope = getResourceScopeFor(clusterName, resourceName);
  admin.removeConfig(scope, keys);
}
 

import org.apache.helix.model.HelixConfigScope; //导入依赖的package包/类
private void setShuttingDownStatus(boolean shuttingDown) {
  HelixConfigScope scope =
      new HelixConfigScopeBuilder(ConfigScopeProperty.PARTICIPANT, _helixClusterName).forParticipant(_instanceId)
          .build();
  Map<String, String> propToUpdate = new HashMap<String, String>();
  propToUpdate.put(CommonConstants.Helix.IS_SHUTDOWN_IN_PROGRESS, String.valueOf(shuttingDown));
  _helixAdmin.setConfig(scope, propToUpdate);
}
 

import org.apache.helix.model.HelixConfigScope; //导入依赖的package包/类
public void createTable(String clusterName, String dbName, String table, String tableSpec) {
  LOG.info("Creating table:" + table + " with table_spec: " + tableSpec + " in dbName:" + dbName);
  IdealState dbIdealState = _helixAdmin.getResourceIdealState(clusterName, dbName);
  int numPartitions = dbIdealState.getNumPartitions();
  String dbTableName = dbName + "." + table;
  AutoModeISBuilder builder = new AutoModeISBuilder(dbTableName);
  builder.setRebalancerMode(RebalanceMode.SEMI_AUTO);
  builder.setNumPartitions(numPartitions);
  builder.setNumReplica(Integer.parseInt(dbIdealState.getReplicas()));
  builder.setStateModel("OnlineOffline");
  builder.setStateModelFactoryName("TableTransitionHandlerFactory");
  for (String dbPartitionName : dbIdealState.getPartitionSet()) {
    String tablePartitionName = dbPartitionName + "." + table;
    Set<String> instanceSet = dbIdealState.getInstanceSet(dbPartitionName);
    builder.add(tablePartitionName);
    String[] instanceNames = new String[instanceSet.size()];
    instanceSet.toArray(instanceNames);
    builder.assignPreferenceList(tablePartitionName, instanceNames);
  }
  IdealState idealState = builder.build();

  // before setting the idealstate, set the configuration
  Map<String, String> properties = new HashMap<String, String>();
  properties.put("table_spec", tableSpec);
  properties.put("type", "TABLE");

  HelixConfigScope scope =
      new HelixConfigScopeBuilder(ConfigScopeProperty.RESOURCE).forCluster(clusterName)
          .forResource(dbTableName).build();
  _helixAdmin.setConfig(scope, properties);
  _helixAdmin.setResourceIdealState(clusterName, dbTableName, idealState);
}
 

import org.apache.helix.model.HelixConfigScope; //导入依赖的package包/类
public TableTransitionHandler(Context context, String databaseName, String databasePartitionName, String tableName) {
  _databasePartitionName = databasePartitionName;
  _tableName = tableName;
  _manager = context.getHelixManager();
  _mysqlAdmin = context.getMysqlAdmin();
  HelixConfigScope scope =
      new HelixConfigScopeBuilder(ConfigScopeProperty.RESOURCE).forCluster(
          _manager.getClusterName()).forResource(databaseName + "." + tableName).build();
  _tableDDL = _manager.getConfigAccessor().get(scope, "table_spec");
}
 

import org.apache.helix.model.HelixConfigScope; //导入依赖的package包/类
/**
 * Create a Helix cluster for the Gobblin Cluster application.
 *
 * @param zkConnectionString the ZooKeeper connection string
 * @param clusterName the Helix cluster name
 * @param overwrite true to overwrite exiting cluster, false to reuse existing cluster
 */
public static void createGobblinHelixCluster(String zkConnectionString, String clusterName, boolean overwrite) {
  ClusterSetup clusterSetup = new ClusterSetup(zkConnectionString);
  // Create the cluster and overwrite if it already exists
  clusterSetup.addCluster(clusterName, overwrite);
  // Helix 0.6.x requires a configuration property to have the form key=value.
  String autoJoinConfig = ZKHelixManager.ALLOW_PARTICIPANT_AUTO_JOIN + "=true";
  clusterSetup.setConfig(HelixConfigScope.ConfigScopeProperty.CLUSTER, clusterName, autoJoinConfig);
}
 

import org.apache.helix.model.HelixConfigScope; //导入依赖的package包/类
/**
 * get a single config entry
 * @param scope specification of the entity set to query
 *          (e.g. cluster, resource, participant, etc.)
 * @param key the identifier of the configuration entry
 * @return the configuration entry
 */
public String get(HelixConfigScope scope, String key) {
  Map<String, String> map = get(scope, Arrays.asList(key));
  if (map != null) {
    return map.get(key);
  }
  return null;
}