import org.apache.helix.participant.StateMachineEngine; //导入依赖的package包/类
public static void addFakeBrokerInstancesToAutoJoinHelixCluster(String helixClusterName, String zkServer, int numInstances, boolean isSingleTenant) throws Exception {
  for (int i = 0; i < numInstances; ++i) {
    final String brokerId = "Broker_localhost_" + i;
    final HelixManager helixZkManager =
        HelixManagerFactory.getZKHelixManager(helixClusterName, brokerId, InstanceType.PARTICIPANT, zkServer);
    final StateMachineEngine stateMachineEngine = helixZkManager.getStateMachineEngine();
    final StateModelFactory<?> stateModelFactory = new EmptyBrokerOnlineOfflineStateModelFactory();
    stateMachineEngine.registerStateModelFactory(EmptyBrokerOnlineOfflineStateModelFactory.getStateModelDef(),
        stateModelFactory);
    helixZkManager.connect();
    if (isSingleTenant) {
      helixZkManager.getClusterManagmentTool().addInstanceTag(helixClusterName, brokerId, ControllerTenantNameBuilder.getBrokerTenantNameForTenant(ControllerTenantNameBuilder.DEFAULT_TENANT_NAME));
    } else {
      helixZkManager.getClusterManagmentTool().addInstanceTag(helixClusterName, brokerId, UNTAGGED_BROKER_INSTANCE);
    }
    Thread.sleep(1000);
  }
}
 

import org.apache.helix.participant.StateMachineEngine; //导入依赖的package包/类
public static void addFakeDataInstancesToAutoJoinHelixCluster(String helixClusterName, String zkServer, int numInstances, boolean isSingleTenant) throws Exception {
  for (int i = 0; i < numInstances; ++i) {
    final String instanceId = "Server_localhost_" + i;

    final HelixManager helixZkManager =
        HelixManagerFactory.getZKHelixManager(helixClusterName, instanceId, InstanceType.PARTICIPANT, zkServer);
    final StateMachineEngine stateMachineEngine = helixZkManager.getStateMachineEngine();
    final StateModelFactory<?> stateModelFactory = new EmptySegmentOnlineOfflineStateModelFactory();
    stateMachineEngine.registerStateModelFactory(EmptySegmentOnlineOfflineStateModelFactory.getStateModelDef(),
        stateModelFactory);
    helixZkManager.connect();
    if (isSingleTenant) {
      helixZkManager.getClusterManagmentTool().addInstanceTag(helixClusterName, instanceId,
          TableNameBuilder.OFFLINE_TABLE_NAME_BUILDER.forTable(ControllerTenantNameBuilder.DEFAULT_TENANT_NAME));
      helixZkManager.getClusterManagmentTool().addInstanceTag(helixClusterName, instanceId,
          TableNameBuilder.REALTIME_TABLE_NAME_BUILDER.forTable(ControllerTenantNameBuilder.DEFAULT_TENANT_NAME));
    } else {
      helixZkManager.getClusterManagmentTool().addInstanceTag(helixClusterName, instanceId, UNTAGGED_SERVER_INSTANCE);
    }
  }
}
 

import org.apache.helix.participant.StateMachineEngine; //导入依赖的package包/类
public void start() throws Exception {
  manager =
      HelixManagerFactory.getZKHelixManager(clusterName, instanceName, InstanceType.PARTICIPANT,
          zkConnectString);

  stateModelFactory = new BootstrapHandler();
  // genericStateMachineHandler = new StateMachineEngine();
  // genericStateMachineHandler.registerStateModelFactory("MasterSlave", stateModelFactory);

  StateMachineEngine stateMach = manager.getStateMachineEngine();
  stateMach.registerStateModelFactory("MasterSlave", stateModelFactory);

  manager.getMessagingService().registerMessageHandlerFactory(
      MessageType.STATE_TRANSITION.name(), stateMach);
  manager.getMessagingService().registerMessageHandlerFactory(
      MessageType.USER_DEFINE_MSG.name(), new CustomMessageHandlerFactory());
  manager.connect();
}
 

import org.apache.helix.participant.StateMachineEngine; //导入依赖的package包/类
public void start() throws Exception {
  manager =
      HelixManagerFactory.getZKHelixManager(clusterName, instanceName, InstanceType.PARTICIPANT,
          zkConnectString);

  if ("MasterSlave".equalsIgnoreCase(stateModelType)) {
    stateModelFactory = new MasterSlaveStateModelFactory(this.instanceName, delay);
  } else if ("OnlineOffline".equalsIgnoreCase(stateModelType)) {
    stateModelFactory = new OnlineOfflineStateModelFactory(this.instanceName, delay);
  } else if ("LeaderStandby".equalsIgnoreCase(stateModelType)) {
    stateModelFactory = new LeaderStandbyStateModelFactory(this.instanceName, delay);
  }
  // genericStateMachineHandler = new StateMachineEngine();
  // genericStateMachineHandler.registerStateModelFactory(stateModelType,
  // stateModelFactory);

  StateMachineEngine stateMach = manager.getStateMachineEngine();
  stateMach.registerStateModelFactory(stateModelType, stateModelFactory);
  manager.connect();
  manager.getMessagingService().registerMessageHandlerFactory(
      MessageType.STATE_TRANSITION.name(), stateMach);
}
 

import org.apache.helix.participant.StateMachineEngine; //导入依赖的package包/类
@Override
public void run() {
  try {
    StateMachineEngine stateMach = getStateMachineEngine();
    DistClusterControllerStateModelFactory lsModelFactory =
        new DistClusterControllerStateModelFactory(_zkAddress);
    stateMach.registerStateModelFactory("LeaderStandby", lsModelFactory);

    connect();
    _startCountDown.countDown();
    _stopCountDown.await();
  } catch (Exception e) {
    LOG.error("exception running controller-manager", e);
  } finally {
    _startCountDown.countDown();
    disconnect();
    _waitStopFinishCountDown.countDown();
  }
}
 

import org.apache.helix.participant.StateMachineEngine; //导入依赖的package包/类
public HelixManager start() throws Exception {
  HelixManager manager = null;
  // zk cluster manager
  if (_clusterMangerType.equalsIgnoreCase("zk")) {
    manager =
        HelixManagerFactory.getZKHelixManager(_clusterName, _instanceName,
            InstanceType.PARTICIPANT, _zkConnectString);
  } else {
    throw new IllegalArgumentException("Unsupported cluster manager type:" + _clusterMangerType);
  }

  MockOnlineOfflineStateModelFactory stateModelFactory2 =
      new MockOnlineOfflineStateModelFactory(_transDelayInMs, _resourceName, _resourceTag,
          _instanceName);
  // genericStateMachineHandler = new StateMachineEngine();
  StateMachineEngine stateMach = manager.getStateMachineEngine();
  stateMach.registerStateModelFactory("OnlineOffline", stateModelFactory2);

  manager.connect();
  //manager.getMessagingService().registerMessageHandlerFactory(MessageType.STATE_TRANSITION.name(), genericStateMachineHandler);
  return manager;
}
 

import org.apache.helix.participant.StateMachineEngine; //导入依赖的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.participant.StateMachineEngine; //导入依赖的package包/类
protected void startParticipantsWithStuckTaskStateModelFactory() {

    Map<String, TaskFactory> taskFactoryReg = new HashMap<String, TaskFactory>();
    taskFactoryReg.put(MockTask.TASK_COMMAND, new TaskFactory() {
      @Override public Task createNewTask(TaskCallbackContext context) {
        return new MockTask(context);
      }
    });

    List<String> instances = _setupTool.getClusterManagementTool().getInstancesInCluster(CLUSTER_NAME);

    _participants[0] = new MockParticipantManager(ZK_ADDR, CLUSTER_NAME, instances.get(0));
    StateMachineEngine stateMachine = _participants[0].getStateMachineEngine();
    stateMachine.registerStateModelFactory("Task",
        new MockTaskStateModelFactory(_participants[0], taskFactoryReg));
    _participants[0].syncStart();
    _blockedParticipant = _participants[0];

    _participants[1] = new MockParticipantManager(ZK_ADDR, CLUSTER_NAME, instances.get(1));
    stateMachine = _participants[1].getStateMachineEngine();
    stateMachine.registerStateModelFactory("Task",
        new TaskStateModelFactory(_participants[1], taskFactoryReg));
    _participants[1].syncStart();
    _normalParticipant = _participants[1];
  }
 

import org.apache.helix.participant.StateMachineEngine; //导入依赖的package包/类
protected void startParticipantsWithStuckTaskStateModelFactory() {
  Map<String, TaskFactory> taskFactoryReg = new HashMap<String, TaskFactory>();
  taskFactoryReg.put(MockTask.TASK_COMMAND, new TaskFactory() {
    @Override public Task createNewTask(TaskCallbackContext context) {
      return new MockTask(context);
    }
  });

  // start dummy participants
  for (int i = 0; i < _numNodes; i++) {
    String instanceName = PARTICIPANT_PREFIX + "_" + (_startPort + i);
    _participants[i] = new MockParticipantManager(ZK_ADDR, CLUSTER_NAME, instanceName);

    // Register a Task state model factory.
    StateMachineEngine stateMachine = _participants[i].getStateMachineEngine();
    stateMachine.registerStateModelFactory("Task",
        new MockTaskStateModelFactory(_participants[i], taskFactoryReg));
    _participants[i].syncStart();
  }
}
 

import org.apache.helix.participant.StateMachineEngine; //导入依赖的package包/类
protected void startParticipant(int i) {
  Map<String, TaskFactory> taskFactoryReg = new HashMap<String, TaskFactory>();
  taskFactoryReg.put(MockTask.TASK_COMMAND, new TaskFactory() {
    @Override public Task createNewTask(TaskCallbackContext context) {
      return new MockTask(context);
    }
  });
  String instanceName = PARTICIPANT_PREFIX + "_" + (_startPort + i);
  _participants[i] = new MockParticipantManager(ZK_ADDR, CLUSTER_NAME, instanceName);

  // Register a Task state model factory.
  StateMachineEngine stateMachine = _participants[i].getStateMachineEngine();
  stateMachine.registerStateModelFactory("Task",
      new TaskStateModelFactory(_participants[i], taskFactoryReg));
  _participants[i].syncStart();
}
 

import org.apache.helix.participant.StateMachineEngine; //导入依赖的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.participant.StateMachineEngine; //导入依赖的package包/类
public static void addFakeBrokerInstancesToAutoJoinHelixCluster(String helixClusterName, String zkServer,
    int numInstances, boolean isSingleTenant)
    throws Exception {
  for (int i = 0; i < numInstances; ++i) {
    final String brokerId = "Broker_localhost_" + i;
    final HelixManager helixZkManager =
        HelixManagerFactory.getZKHelixManager(helixClusterName, brokerId, InstanceType.PARTICIPANT, zkServer);
    final StateMachineEngine stateMachineEngine = helixZkManager.getStateMachineEngine();
    final StateModelFactory<?> stateModelFactory = new EmptyBrokerOnlineOfflineStateModelFactory();
    stateMachineEngine
        .registerStateModelFactory(EmptyBrokerOnlineOfflineStateModelFactory.getStateModelDef(), stateModelFactory);
    helixZkManager.connect();
    if (isSingleTenant) {
      helixZkManager.getClusterManagmentTool().addInstanceTag(helixClusterName, brokerId,
          ControllerTenantNameBuilder.getBrokerTenantNameForTenant(ControllerTenantNameBuilder.DEFAULT_TENANT_NAME));
    } else {
      helixZkManager.getClusterManagmentTool().addInstanceTag(helixClusterName, brokerId, UNTAGGED_BROKER_INSTANCE);
    }
  }
}
 

import org.apache.helix.participant.StateMachineEngine; //导入依赖的package包/类
/**
 * Initialize the participant by registering via the {@link HelixManager} as a participant to the associated Helix
 * cluster.
 * @param hostName the hostname to use when registering as a participant.
 * @param port the port to use when registering as a participant.
 * @param ambryHealthReports {@link List} of {@link AmbryHealthReport} to be registered to the participant.
 * @throws IOException if there is an error connecting to the Helix cluster.
 */
@Override
public void initialize(String hostName, int port, List<AmbryHealthReport> ambryHealthReports) throws IOException {
  logger.info("Initializing participant");
  instanceName = ClusterMapUtils.getInstanceName(hostName, port);
  manager = helixFactory.getZKHelixManager(clusterName, instanceName, InstanceType.PARTICIPANT, zkConnectStr);
  StateMachineEngine stateMachineEngine = manager.getStateMachineEngine();
  stateMachineEngine.registerStateModelFactory(LeaderStandbySMD.name, new AmbryStateModelFactory());
  registerHealthReportTasks(stateMachineEngine, ambryHealthReports);
  try {
    manager.connect();
  } catch (Exception e) {
    throw new IOException("Exception while connecting to the Helix manager", e);
  }
  for (AmbryHealthReport ambryHealthReport : ambryHealthReports) {
    manager.getHealthReportCollector().addHealthReportProvider((HealthReportProvider) ambryHealthReport);
  }
}
 

import org.apache.helix.participant.StateMachineEngine; //导入依赖的package包/类
/**
 * Register {@link HelixHealthReportAggregatorTask}s for appropriate {@link AmbryHealthReport}s.
 * @param engine the {@link StateMachineEngine} to register the task state model.
 * @param healthReports the {@link List} of {@link AmbryHealthReport}s that may require the registration of
 * corresponding {@link HelixHealthReportAggregatorTask}s.
 */
private void registerHealthReportTasks(StateMachineEngine engine, List<AmbryHealthReport> healthReports) {
  Map<String, TaskFactory> taskFactoryMap = new HashMap<>();
  for (final AmbryHealthReport healthReport : healthReports) {
    if (healthReport.getAggregateIntervalInMinutes() != Utils.Infinite_Time) {
      // register cluster wide aggregation task for the health report
      taskFactoryMap.put(
          String.format("%s_%s", HelixHealthReportAggregatorTask.TASK_COMMAND_PREFIX, healthReport.getReportName()),
          new TaskFactory() {
            @Override
            public Task createNewTask(TaskCallbackContext context) {
              return new HelixHealthReportAggregatorTask(context, healthReport.getAggregateIntervalInMinutes(),
                  healthReport.getReportName(), healthReport.getQuotaStatsFieldName());
            }
          });
    }
  }
  if (!taskFactoryMap.isEmpty()) {
    engine.registerStateModelFactory(TaskConstants.STATE_MODEL_NAME,
        new TaskStateModelFactory(manager, taskFactoryMap));
  }
}
 

import org.apache.helix.participant.StateMachineEngine; //导入依赖的package包/类
public void start() throws Exception {
  _helixZkManager = HelixManagerFactory.getZKHelixManager(_helixClusterName, _instanceId,
      InstanceType.PARTICIPANT, _zkServer);
  StateMachineEngine stateMachineEngine = _helixZkManager.getStateMachineEngine();
  // create a stateModelFactory that returns a statemodel object for each partition.
  TestOnlineOfflineStateModelFactory stateModelFactory =
      new TestOnlineOfflineStateModelFactory(_instanceId, 0);
  stateMachineEngine.registerStateModelFactory("OnlineOffline", stateModelFactory);
  _helixZkManager.connect();
}
 

import org.apache.helix.participant.StateMachineEngine; //导入依赖的package包/类
public void start() throws Exception {
  String zookeeperQuorum = TerrapinUtil.getZKQuorumFromConf(configuration);
  int thriftPort = configuration.getInt(Constants.THRIFT_PORT, Constants.DEFAULT_THRIFT_PORT);

  // Connect to Helix.
  this.helixManager = HelixManagerFactory.getZKHelixManager(
      configuration.getString(Constants.HELIX_CLUSTER, Constants.HELIX_CLUSTER_NAME_DEFAULT),
      TerrapinUtil.getHelixInstanceFromHDFSHost(InetAddress.getLocalHost().getHostName()),
      InstanceType.PARTICIPANT,
      zookeeperQuorum);
  StateMachineEngine stateMach = this.helixManager.getStateMachineEngine();

  // Create state model factory for HDFS.
  Configuration conf = new Configuration();
  conf.set("fs.default.name", configuration.getString(Constants.HDFS_NAMENODE));
  // Setup HDFS short circuit parameters.
  conf.setBoolean("dfs.client.read.shortcircuit", true);
  conf.setInt("dfs.client.read.shortcircuit.streams.cache.size", 5000);
  conf.setInt("dfs.client.read.shortcircuit.buffer.size", 131072);
  conf.set("dfs.domain.socket.path", "/var/run/hadoop-hdfs/dn._PORT");

  FileSystem fs = FileSystem.get(conf);
  this.stateModelFactory = new OnlineOfflineStateModelFactory(
      this.configuration,
      resourcePartitionMap,
      new ReaderFactory(configuration, new HFileSystem(fs)));
  stateMach.registerStateModelFactory("OnlineOffline", this.stateModelFactory);
  this.helixManager.connect();

  // Start up the thrift server for serving.
  startThriftServer(thriftPort);
}
 

import org.apache.helix.participant.StateMachineEngine; //导入依赖的package包/类
public HelixServerStarter(String helixClusterName, String zkServer, Configuration pinotHelixProperties)
    throws Exception {

  _helixClusterName = helixClusterName;
  _pinotHelixProperties = pinotHelixProperties;

  _instanceId =
      pinotHelixProperties.getString(
          "instanceId",
          CommonConstants.Helix.PREFIX_OF_SERVER_INSTANCE
              + pinotHelixProperties.getString(CommonConstants.Helix.KEY_OF_SERVER_NETTY_HOST,
                  NetUtil.getHostAddress())
              + "_"
              + pinotHelixProperties.getInt(CommonConstants.Helix.KEY_OF_SERVER_NETTY_PORT,
                  CommonConstants.Helix.DEFAULT_SERVER_NETTY_PORT));

  pinotHelixProperties.addProperty("pinot.server.instance.id", _instanceId);
  startServerInstance(pinotHelixProperties);

  // Replace all white-spaces from list of zkServers.
  String zkServers = zkServer.replaceAll("\\s+", "");
  _helixManager =
      HelixManagerFactory.getZKHelixManager(helixClusterName, _instanceId, InstanceType.PARTICIPANT, zkServers);
  final StateMachineEngine stateMachineEngine = _helixManager.getStateMachineEngine();
  _helixManager.connect();
  ZkHelixPropertyStore<ZNRecord> zkPropertyStore = ZkUtils.getZkPropertyStore(_helixManager, helixClusterName);
  final StateModelFactory<?> stateModelFactory =
      new SegmentOnlineOfflineStateModelFactory(helixClusterName, _instanceId,
          _serverInstance.getInstanceDataManager(), new ColumnarSegmentMetadataLoader(), pinotHelixProperties,
          zkPropertyStore);
  stateMachineEngine.registerStateModelFactory(SegmentOnlineOfflineStateModelFactory.getStateModelDef(),
      stateModelFactory);
  _helixAdmin = _helixManager.getClusterManagmentTool();
  addInstanceTagIfNeeded(helixClusterName, _instanceId);
  setShuttingDownStatus(false);

  _serverInstance.getServerMetrics().addCallbackGauge(
      "helix.connected", () -> _helixManager.isConnected() ? 1L : 0L);
}
 

import org.apache.helix.participant.StateMachineEngine; //导入依赖的package包/类
public void start(String stateModelName, StateModelFactory<T> stateModelFactory) {
    try {
        manager = HelixManagerFactory.getZKHelixManager(clusterName, instanceName,
                InstanceType.PARTICIPANT, zkAddr);

        StateMachineEngine stateMachine = manager.getStateMachineEngine();
        stateMachine.registerStateModelFactory(stateModelName, stateModelFactory);

        manager.connect();
    } catch (Exception e) {
        logger.error("failed to connect manager", e);
        throw new RuntimeException("failed to start HelixPartitionManager");
    }
}
 

import org.apache.helix.participant.StateMachineEngine; //导入依赖的package包/类
public void start() throws Exception {
  manager =
      HelixManagerFactory.getZKHelixManager(CLUSTER_NAME, instanceName,
          InstanceType.PARTICIPANT, ZK_ADDRESS);

  MasterSlaveStateModelFactory stateModelFactory =
      new MasterSlaveStateModelFactory(instanceName);

  StateMachineEngine stateMach = manager.getStateMachineEngine();
  stateMach.registerStateModelFactory(STATE_MODEL_NAME, stateModelFactory);
  manager.connect();
}
 

import org.apache.helix.participant.StateMachineEngine; //导入依赖的package包/类
public static void main(String[] args) {
  if (args.length < 3) {
    System.err.println("USAGE: DummyParticipant zkAddress clusterName instanceName");
    System.exit(1);
  }

  String zkAddr = args[0];
  String clusterName = args[1];
  String instanceName = args[2];

  HelixManager manager = null;
  try {
    manager =
        HelixManagerFactory.getZKHelixManager(clusterName, instanceName,
            InstanceType.PARTICIPANT, zkAddr);

    StateMachineEngine stateMach = manager.getStateMachineEngine();
    DummyMSModelFactory msModelFactory = new DummyMSModelFactory();
    stateMach.registerStateModelFactory("MasterSlave", msModelFactory);

    manager.connect();

    Thread.currentThread().join();
  } catch (Exception e) {
    // TODO Auto-generated catch block
    e.printStackTrace();
  } finally {
    if (manager != null) {
      manager.disconnect();
    }
  }
}