import org.apache.helix.manager.zk.ZKHelixManager; //导入依赖的package包/类
public void start() throws Exception {
  helixManager =
      new ZKHelixManager(clusterName, instanceName, InstanceType.PARTICIPANT, ZK_ADDR);
  {
    // hack to set sessionTimeout
    Field sessionTimeout = ZKHelixManager.class.getDeclaredField("_sessionTimeout");
    sessionTimeout.setAccessible(true);
    sessionTimeout.setInt(helixManager, 1000);
  }

  StateMachineEngine stateMach = helixManager.getStateMachineEngine();
  stateMach.registerStateModelFactory("MasterSlave", new MyStateModelFactory(helixManager));
  helixManager.connect();

  StatusPrinter statusPrinter = new StatusPrinter();
  statusPrinter.registerWith(helixManager);
}
 

import org.apache.helix.manager.zk.ZKHelixManager; //导入依赖的package包/类
public static void main(String[] args) throws Exception {
  CommandLine cmd = processCommandLineArgs(args);
  String zkAddress = cmd.getOptionValue(zkAddr);
  String clusterName = cmd.getOptionValue(cluster);
  String instance = cmd.getOptionValue(instanceName);
  String stateModelName = cmd.getOptionValue(stateModel);

  HelixManager manager =
      new ZKHelixManager(clusterName, instance, InstanceType.PARTICIPANT, zkAddress);

  StateMachineEngine stateMach = manager.getStateMachineEngine();
  stateMach.registerStateModelFactory(stateModelName, new AgentStateModelFactory());

  Runtime.getRuntime().addShutdownHook(new HelixAgentShutdownHook(manager));

  try {
    manager.connect();
    Thread.currentThread().join();
  } catch (Exception e) {
    LOG.error(e.toString());
  } finally {
    if (manager != null && manager.isConnected()) {
      manager.disconnect();
    }
  }
}
 

import org.apache.helix.manager.zk.ZKHelixManager; //导入依赖的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.manager.zk.ZKHelixManager; //导入依赖的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.manager.zk.ZKHelixManager; //导入依赖的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.manager.zk.ZKHelixManager; //导入依赖的package包/类
private HelixManager createParticipant(String clusterName, String instanceName) {
  HelixManager participant =
      new ZKHelixManager(clusterName, instanceName, InstanceType.PARTICIPANT, ZK_ADDR);
  participant.getStateMachineEngine().registerStateModelFactory("OnlineOffline",
      new MockStateModelFactory());
  return participant;
}
 

import org.apache.helix.manager.zk.ZKHelixManager; //导入依赖的package包/类
/**
 * @return returns true if
 */
public boolean start() throws Exception {
  // auto create cluster and allow nodes to automatically join the cluster
  admin =
      HelixManagerFactory.getZKHelixManager(cluster, "service-discovery",
          InstanceType.ADMINISTRATOR, zkAddress);
  admin.connect();
  admin.getClusterManagmentTool().addCluster(cluster, false);
  HelixConfigScope scope =
      new HelixConfigScopeBuilder(ConfigScopeProperty.CLUSTER).forCluster(cluster).build();

  Map<String, String> properties = new HashMap<String, String>();
  properties.put(ZKHelixManager.ALLOW_PARTICIPANT_AUTO_JOIN, String.valueOf(true));
  admin.getClusterManagmentTool().setConfig(scope, properties);
  switch (mode) {
  case POLL:
    startBackgroundTask();
    break;
  case WATCH:
    setupWatcher();
    break;
  case NONE:// dont monitor changes, supports only registration

  }
  refreshCache();
  return true;
}
 

import org.apache.helix.manager.zk.ZKHelixManager; //导入依赖的package包/类
public MinionStarter(String zkAddress, String helixClusterName, Configuration config) throws Exception {
  _helixClusterName = helixClusterName;
  _config = config;
  _instanceId = config.getString(CommonConstants.Helix.Instance.INSTANCE_ID_KEY,
      CommonConstants.Minion.INSTANCE_PREFIX + NetUtil.getHostAddress() + "_"
          + CommonConstants.Minion.DEFAULT_HELIX_PORT);
  _helixManager = new ZKHelixManager(_helixClusterName, _instanceId, InstanceType.PARTICIPANT, zkAddress);
  _taskExecutorRegistry = new TaskExecutorRegistry();
}
 

import org.apache.helix.manager.zk.ZKHelixManager; //导入依赖的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 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));
  //we need only one segment to be loaded at a time
  props.put(MessageType.STATE_TRANSITION + "." + HelixTaskExecutor.MAX_THREADS, String.valueOf(1));

  admin.setConfig(scope, props);

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

  admin.addStateModelDef(helixClusterName,
      PinotHelixSegmentOnlineOfflineStateModelGenerator.PINOT_SEGMENT_ONLINE_OFFLINE_STATE_MODEL,
      PinotHelixSegmentOnlineOfflineStateModelGenerator.generatePinotStateModelDefinition());

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

  admin.addStateModelDef(helixClusterName,
      PinotHelixBrokerResourceOnlineOfflineStateModelGenerator.PINOT_BROKER_RESOURCE_ONLINE_OFFLINE_STATE_MODEL,
      PinotHelixBrokerResourceOnlineOfflineStateModelGenerator.generatePinotStateModelDefinition());

  LOGGER.info("Adding empty ideal state for Broker!");
  HelixHelper.updateResourceConfigsFor(new HashMap<String, String>(), CommonConstants.Helix.BROKER_RESOURCE_INSTANCE,
      helixClusterName, admin);
  IdealState idealState =
      PinotTableIdealStateBuilder.buildEmptyIdealStateForBrokerResource(admin, helixClusterName);
  admin.setResourceIdealState(helixClusterName, CommonConstants.Helix.BROKER_RESOURCE_INSTANCE, idealState);
  LOGGER.info("New Cluster setup completed... ********************************************** ");
}
 

import org.apache.helix.manager.zk.ZKHelixManager; //导入依赖的package包/类
@Test
public void simpleIntegrationTest() throws Exception {
  // Logger.getRootLogger().setLevel(Level.INFO);
  String className = TestHelper.getTestClassName();
  String methodName = TestHelper.getTestMethodName();
  String clusterName = className + "_" + methodName;
  int n = 2;

  System.out.println("START " + clusterName + " at " + new Date(System.currentTimeMillis()));

  TestHelper.setupCluster(clusterName, ZK_ADDR, 12918, // participant port
      "localhost", // participant name prefix
      "TestDB", // resource name prefix
      1, // resources
      4, // partitions per resource
      n, // number of nodes
      2, // replicas
      "MasterSlave", true); // do rebalance

  HelixManager[] distributedControllers = new HelixManager[n];
  for (int i = 0; i < n; i++) {
    int port = 12918 + i;
    distributedControllers[i] =
        new ZKHelixManager(clusterName, "localhost_" + port, InstanceType.CONTROLLER_PARTICIPANT,
            ZK_ADDR);
    distributedControllers[i].getStateMachineEngine().registerStateModelFactory("MasterSlave",
        new MockMSModelFactory());
    distributedControllers[i].connect();
  }

  boolean result =
      ClusterStateVerifier.verifyByZkCallback(new BestPossAndExtViewZkVerifier(ZK_ADDR,
          clusterName));
  Assert.assertTrue(result);

  // disconnect first distributed-controller, and verify second takes leadership
  distributedControllers[0].disconnect();

  // verify leader changes to localhost_12919
  Thread.sleep(100);
  result =
      ClusterStateVerifier.verifyByZkCallback(new BestPossAndExtViewZkVerifier(ZK_ADDR,
          clusterName));
  Assert.assertTrue(result);

  ZKHelixDataAccessor accessor =
      new ZKHelixDataAccessor(clusterName, new ZkBaseDataAccessor<ZNRecord>(_gZkClient));
  PropertyKey.Builder keyBuilder = accessor.keyBuilder();
  Assert.assertNull(accessor.getProperty(keyBuilder.liveInstance("localhost_12918")));
  LiveInstance leader = accessor.getProperty(keyBuilder.controllerLeader());
  Assert.assertNotNull(leader);
  Assert.assertEquals(leader.getId(), "localhost_12919");

  // clean up
  distributedControllers[1].disconnect();
  Assert.assertNull(accessor.getProperty(keyBuilder.liveInstance("localhost_12919")));
  Assert.assertNull(accessor.getProperty(keyBuilder.controllerLeader()));

  System.out.println("START " + clusterName + " at " + new Date(System.currentTimeMillis()));
}
 

import org.apache.helix.manager.zk.ZKHelixManager; //导入依赖的package包/类
@Test
public void simpleIntegrationTest() throws Exception {
  // Logger.getRootLogger().setLevel(Level.INFO);
  String className = TestHelper.getTestClassName();
  String methodName = TestHelper.getTestMethodName();
  String clusterName = className + "_" + methodName;
  int n = 1;

  System.out.println("START " + clusterName + " at " + new Date(System.currentTimeMillis()));

  TestHelper.setupCluster(clusterName, ZK_ADDR, 12918, // participant port
      "localhost", // participant name prefix
      "TestDB", // resource name prefix
      1, // resources
      4, // partitions per resource
      n, // number of nodes
      1, // replicas
      "MasterSlave", true); // do rebalance

  HelixManager participant =
      new ZKHelixManager(clusterName, "localhost_12918", InstanceType.PARTICIPANT, ZK_ADDR);
  participant.getStateMachineEngine().registerStateModelFactory("MasterSlave",
      new MockMSModelFactory());
  participant.connect();

  HelixManager controller =
      new ZKHelixManager(clusterName, "controller_0", InstanceType.CONTROLLER, ZK_ADDR);
  controller.connect();

  boolean result =
      ClusterStateVerifier.verifyByZkCallback(new BestPossAndExtViewZkVerifier(ZK_ADDR,
          clusterName));
  Assert.assertTrue(result);

  // cleanup
  controller.disconnect();
  participant.disconnect();

  // verify all live-instances and leader nodes are gone
  ZKHelixDataAccessor accessor =
      new ZKHelixDataAccessor(clusterName, new ZkBaseDataAccessor<ZNRecord>(_gZkClient));
  PropertyKey.Builder keyBuilder = accessor.keyBuilder();
  Assert.assertNull(accessor.getProperty(keyBuilder.liveInstance("localhost_12918")));
  Assert.assertNull(accessor.getProperty(keyBuilder.controllerLeader()));

  System.out.println("END " + clusterName + " at " + new Date(System.currentTimeMillis()));
}
 

import org.apache.helix.manager.zk.ZKHelixManager; //导入依赖的package包/类
@Test
public void testSyncSessionToController() throws Exception {
  System.out.println("START testSyncSessionToController at " + new Date(System.currentTimeMillis()));

  String clusterName = getShortClassName();
  MockParticipantManager[] participants = new MockParticipantManager[5];
  int resourceNb = 10;
  TestHelper.setupCluster(clusterName, ZK_ADDR, 12918, // participant port
      "localhost", // participant name prefix
      "TestDB", // resource name prefix
      resourceNb, // resources
      1, // partitions per resource
      5, // number of nodes
      1, // replicas
      "MasterSlave", true); // do rebalance

  ClusterControllerManager controller =
      new ClusterControllerManager(ZK_ADDR, clusterName, "controller_0");
  controller.syncStart();

  // start participants
  for (int i = 0; i < 5; i++) {
    String instanceName = "localhost_" + (12918 + i);

    participants[i] = new MockParticipantManager(ZK_ADDR, clusterName, instanceName);
    participants[i].syncStart();
  }

  ZKHelixManager zkHelixManager = new ZKHelixManager(clusterName, "controllerMessageListener", InstanceType.CONTROLLER, ZK_ADDR);
  zkHelixManager.connect();
  MockMessageListener mockMessageListener = new MockMessageListener();
  zkHelixManager.addControllerMessageListener(mockMessageListener);

  PropertyKey.Builder keyBuilder = new PropertyKey.Builder(clusterName);
  ZkBaseDataAccessor<ZNRecord> accessor = new ZkBaseDataAccessor<ZNRecord>(_gZkClient);
  String path = keyBuilder.liveInstance("localhost_12918").getPath();
  Stat stat = new Stat();
  ZNRecord data = accessor.get(path, stat, 2);
  data.getSimpleFields().put("SESSION_ID", "invalid-id");
  accessor.set(path, data, 2);
  Thread.sleep(2000);

  Assert.assertTrue(mockMessageListener.isSessionSyncMessageSent());
}
 

import org.apache.helix.manager.zk.ZKHelixManager; //导入依赖的package包/类
@Test
public void testInstanceAutoJoin() throws Exception {
  HelixManager manager = _participants[0];
  HelixDataAccessor accessor = manager.getHelixDataAccessor();

  _setupTool.addResourceToCluster(CLUSTER_NAME, db2, 60, "OnlineOffline", RebalanceMode.FULL_AUTO
      + "");

  _setupTool.rebalanceStorageCluster(CLUSTER_NAME, db2, 1);
  String instance2 = "localhost_279699";
  // StartCMResult result = TestHelper.startDummyProcess(ZK_ADDR, CLUSTER_NAME, instance2);
  MockParticipantManager newParticipant =
      new MockParticipantManager(ZK_ADDR, CLUSTER_NAME, instance2);
  newParticipant.syncStart();

  Thread.sleep(500);
  // Assert.assertFalse(result._thread.isAlive());
  Assert.assertTrue(null == manager.getHelixDataAccessor().getProperty(
      accessor.keyBuilder().liveInstance(instance2)));

  ConfigScope scope = new ConfigScopeBuilder().forCluster(CLUSTER_NAME).build();

  manager.getConfigAccessor().set(scope, ZKHelixManager.ALLOW_PARTICIPANT_AUTO_JOIN, "true");

  newParticipant = new MockParticipantManager(ZK_ADDR, CLUSTER_NAME, instance2);
  newParticipant.syncStart();

  Thread.sleep(500);
  // Assert.assertTrue(result._thread.isAlive() || result2._thread.isAlive());
  for (int i = 0; i < 20; i++) {
    if (null == manager.getHelixDataAccessor().getProperty(
        accessor.keyBuilder().liveInstance(instance2))) {
      Thread.sleep(100);
    } else
      break;
  }
  Assert.assertTrue(null != manager.getHelixDataAccessor().getProperty(
      accessor.keyBuilder().liveInstance(instance2)));

  newParticipant.syncStop();
}
 

import org.apache.helix.manager.zk.ZKHelixManager; //导入依赖的package包/类
public static void createHelixClusterIfNeeded(String helixClusterName, String zkPath, boolean isUpdateStateModel) {
  final HelixAdmin admin = new ZKHelixAdmin(zkPath);
  final String segmentStateModelName = PinotHelixSegmentOnlineOfflineStateModelGenerator.PINOT_SEGMENT_ONLINE_OFFLINE_STATE_MODEL;

  if (admin.getClusters().contains(helixClusterName)) {
    LOGGER.info("cluster already exists ********************************************* ");
    if (isUpdateStateModel) {
      final StateModelDefinition curStateModelDef = admin.getStateModelDef(helixClusterName, segmentStateModelName);
      List<String> states = curStateModelDef.getStatesPriorityList();
      if (states.contains(PinotHelixSegmentOnlineOfflineStateModelGenerator.CONSUMING_STATE)) {
        LOGGER.info("State model {} already updated to contain CONSUMING state", segmentStateModelName);
        return;
      } else {
        LOGGER.info("Updating {} to add states for low level kafka consumers", segmentStateModelName);
        StateModelDefinition newStateModelDef = PinotHelixSegmentOnlineOfflineStateModelGenerator.generatePinotStateModelDefinition();
        ZkClient zkClient = new ZkClient(zkPath);
        zkClient.waitUntilConnected(20, TimeUnit.SECONDS);
        zkClient.setZkSerializer(new ZNRecordSerializer());
        HelixDataAccessor accessor = new ZKHelixDataAccessor(helixClusterName, new ZkBaseDataAccessor<ZNRecord>(zkClient));
        PropertyKey.Builder keyBuilder = accessor.keyBuilder();
        accessor.setProperty(keyBuilder.stateModelDef(segmentStateModelName), newStateModelDef);
        LOGGER.info("Completed updating statemodel {}", segmentStateModelName);
        zkClient.close();
      }
    }
    return;
  }

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

  LOGGER.info("Enable 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));
  //we need only one segment to be loaded at a time
  props.put(MessageType.STATE_TRANSITION + "." + HelixTaskExecutor.MAX_THREADS, String.valueOf(1));

  admin.setConfig(scope, props);

  LOGGER.info("Adding state model {} (with CONSUMED state) generated using {} **********************************************",
      segmentStateModelName , PinotHelixSegmentOnlineOfflineStateModelGenerator.class.toString());

  // If this is a fresh cluster we are creating, then the cluster will see the CONSUMING state in the
  // state model. But then the servers will never be asked to go to that STATE (whether they have the code
  // to handle it or not) unil we complete the feature using low-level kafka consumers and turn the feature on.
  admin.addStateModelDef(helixClusterName, segmentStateModelName,
      PinotHelixSegmentOnlineOfflineStateModelGenerator.generatePinotStateModelDefinition());

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

  admin.addStateModelDef(helixClusterName,
      PinotHelixBrokerResourceOnlineOfflineStateModelGenerator.PINOT_BROKER_RESOURCE_ONLINE_OFFLINE_STATE_MODEL,
      PinotHelixBrokerResourceOnlineOfflineStateModelGenerator.generatePinotStateModelDefinition());

  LOGGER.info("Adding empty ideal state for Broker!");
  HelixHelper.updateResourceConfigsFor(new HashMap<String, String>(), CommonConstants.Helix.BROKER_RESOURCE_INSTANCE,
      helixClusterName, admin);
  IdealState idealState =
      PinotTableIdealStateBuilder.buildEmptyIdealStateForBrokerResource(admin, helixClusterName);
  admin.setResourceIdealState(helixClusterName, CommonConstants.Helix.BROKER_RESOURCE_INSTANCE, idealState);
  initPropertyStorePath(helixClusterName, zkPath);
  LOGGER.info("New Cluster setup completed... ********************************************** ");
}
 

import org.apache.helix.manager.zk.ZKHelixManager; //导入依赖的package包/类
/**
 * Construct a zk-based cluster manager that enforces all types (PARTICIPANT, CONTROLLER, and
 * SPECTATOR) to have a name
 * @param clusterName
 * @param instanceName
 * @param type
 * @param zkAddr
 * @return a HelixManager backed by Zookeeper
 */
public static HelixManager getZKHelixManager(String clusterName, String instanceName,
    InstanceType type, String zkAddr) {
  return new ZKHelixManager(clusterName, instanceName, type, zkAddr);
}