import org.apache.helix.participant.statemachine.StateModelFactory; //导入依赖的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.statemachine.StateModelFactory; //导入依赖的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.statemachine.StateModelFactory; //导入依赖的package包/类
@Override
public boolean registerStateModelFactory(String stateModelName,
    StateModelFactory<? extends StateModel> factory, String factoryName) {
  if (stateModelName == null || factory == null || factoryName == null) {
    throw new HelixException("stateModelDef|stateModelFactory|factoryName cannot be null");
  }

  logger.info("Register state model factory for state model " + stateModelName
      + " using factory name " + factoryName + " with " + factory);

  if (!_stateModelFactoryMap.containsKey(stateModelName)) {
    _stateModelFactoryMap.put(stateModelName,
        new ConcurrentHashMap<String, StateModelFactory<? extends StateModel>>());
  }

  if (_stateModelFactoryMap.get(stateModelName).containsKey(factoryName)) {
    logger.warn("stateModelFactory for " + stateModelName + " using factoryName " + factoryName
        + " has already been registered.");
    return false;
  }

  _stateModelFactoryMap.get(stateModelName).put(factoryName, factory);
  sendNopMessage();
  return true;
}
 

import org.apache.helix.participant.statemachine.StateModelFactory; //导入依赖的package包/类
@Override
public void reset() {
  for (Map<String, StateModelFactory<? extends StateModel>> ftyMap : _stateModelFactoryMap
      .values()) {
    for (StateModelFactory<? extends StateModel> stateModelFactory : ftyMap.values()) {
      for (String resourceName : stateModelFactory.getResourceSet()) {
        for (String partitionKey : stateModelFactory.getPartitionSet(resourceName)) {
          StateModel stateModel = stateModelFactory.getStateModel(resourceName, partitionKey);
          stateModel.reset();
          String initialState = _stateModelParser.getInitialState(stateModel.getClass());
          stateModel.updateState(initialState);
          // TODO probably should update the state on ZK. Shi confirm what needs
          // to be done here.
        }
      }
    }
  }
}
 

import org.apache.helix.participant.statemachine.StateModelFactory; //导入依赖的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.statemachine.StateModelFactory; //导入依赖的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.statemachine.StateModelFactory; //导入依赖的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.statemachine.StateModelFactory; //导入依赖的package包/类
public HelixStateMachineEngine(HelixManager manager) {
  _stateModelParser = new StateModelParser();
  _manager = manager;

  _stateModelFactoryMap =
      new ConcurrentHashMap<String, Map<String, StateModelFactory<? extends StateModel>>>();
  _stateModelDefs = new ConcurrentHashMap<String, StateModelDefinition>();
}
 

import org.apache.helix.participant.statemachine.StateModelFactory; //导入依赖的package包/类
@Override
public StateModelFactory<? extends StateModel> getStateModelFactory(String stateModelName,
    String factoryName) {
  if (!_stateModelFactoryMap.containsKey(stateModelName)) {
    return null;
  }
  return _stateModelFactoryMap.get(stateModelName).get(factoryName);
}
 

import org.apache.helix.participant.statemachine.StateModelFactory; //导入依赖的package包/类
public HelixStateTransitionHandler(StateModelFactory<? extends StateModel> stateModelFactory,
    StateModel stateModel, Message message, NotificationContext context,
    CurrentState currentStateDelta) {
  super(message, context);
  _stateModel = stateModel;
  _statusUpdateUtil = new StatusUpdateUtil();
  _transitionMethodFinder = new StateModelParser();
  _currentStateDelta = currentStateDelta;
  _manager = _notificationContext.getManager();
  _stateModelFactory = stateModelFactory;
}
 

import org.apache.helix.participant.statemachine.StateModelFactory; //导入依赖的package包/类
/**
 * check state-model factory contains state-models same as in expect-state-model map
 * @param fty
 * @param expectStateModelMap
 */
static void checkStateModelMap(StateModelFactory<? extends StateModel> fty,
    Map<String, String> expectStateModelMap) {
  Assert.assertEquals(fty.getPartitionSet("TestDB0").size(), expectStateModelMap.size());
  for (String partition : fty.getPartitionSet("TestDB0")) {
    StateModel stateModel = fty.getStateModel("TestDB0", partition);
    String actualState = stateModel.getCurrentState();
    String expectState = expectStateModelMap.get(partition);
    LOG.debug(partition + " actual state: " + actualState + ", expect state: " + expectState);
    Assert.assertEquals(actualState, expectState, "partition: " + partition
        + " should be in state: " + expectState + " but was " + actualState);
  }
}
 

import org.apache.helix.participant.statemachine.StateModelFactory; //导入依赖的package包/类
public static void addFakeDataInstancesToAutoJoinHelixCluster(String helixClusterName, String zkServer,
    int numInstances, boolean isSingleTenant, int adminPort)
    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.tableNameWithType(ControllerTenantNameBuilder.DEFAULT_TENANT_NAME));
      helixZkManager.getClusterManagmentTool()
          .addInstanceTag(helixClusterName, instanceId,
              TableNameBuilder.REALTIME.tableNameWithType(ControllerTenantNameBuilder.DEFAULT_TENANT_NAME));
    } else {
      helixZkManager.getClusterManagmentTool().addInstanceTag(helixClusterName, instanceId, UNTAGGED_SERVER_INSTANCE);
    }
    HelixConfigScope scope =
        new HelixConfigScopeBuilder(HelixConfigScope.ConfigScopeProperty.PARTICIPANT, helixClusterName)
            .forParticipant(instanceId).build();
    Map<String, String> props = new HashMap<>();
    props.put(CommonConstants.Helix.Instance.ADMIN_PORT_KEY, String.valueOf(adminPort + i));
    helixZkManager.getClusterManagmentTool().setConfig(scope, props);
  }
}
 

import org.apache.helix.participant.statemachine.StateModelFactory; //导入依赖的package包/类
public HelixBrokerStarter(String helixClusterName, String zkServer, Configuration pinotHelixProperties)
    throws Exception {
  _liveInstancesListener = new LiveInstancesChangeListenerImpl(helixClusterName);

  _pinotHelixProperties = DefaultHelixBrokerConfig.getDefaultBrokerConf(pinotHelixProperties);
  final String brokerId =
      _pinotHelixProperties.getString(
              "instanceId",
              CommonConstants.Helix.PREFIX_OF_BROKER_INSTANCE
                      + NetUtil.getHostAddress()
                      + "_"
                      + _pinotHelixProperties.getInt(CommonConstants.Helix.KEY_OF_BROKER_QUERY_PORT,
                  CommonConstants.Helix.DEFAULT_BROKER_QUERY_PORT));

  _pinotHelixProperties.addProperty("pinot.broker.id", brokerId);
  RoutingTableBuilder defaultOfflineRoutingTableBuilder =
      getRoutingTableBuilder(_pinotHelixProperties.subset(DEFAULT_OFFLINE_ROUTING_TABLE_BUILDER_KEY));
  RoutingTableBuilder defaultRealtimeRoutingTableBuilder =
      getRoutingTableBuilder(_pinotHelixProperties.subset(DEFAULT_REALTIME_ROUTING_TABLE_BUILDER_KEY));
  Map<String, RoutingTableBuilder> tableToRoutingTableBuilderMap =
      getTableToRoutingTableBuilderMap(_pinotHelixProperties.subset(ROUTING_TABLE_BUILDER_KEY));

  // Remove all white-spaces from the list of zkServers (if any).
  String zkServers = zkServer.replaceAll("\\s+", "");

  _zkClient =
      new ZkClient(getZkAddressForBroker(zkServers, helixClusterName),
          ZkClient.DEFAULT_SESSION_TIMEOUT, ZkClient.DEFAULT_CONNECTION_TIMEOUT, new ZNRecordSerializer());
  _propertyStore = new ZkHelixPropertyStore<ZNRecord>(new ZkBaseDataAccessor<ZNRecord>(_zkClient), "/", null);
  _helixExternalViewBasedRouting =
      new HelixExternalViewBasedRouting(defaultOfflineRoutingTableBuilder, defaultRealtimeRoutingTableBuilder,
          tableToRoutingTableBuilderMap, _propertyStore);

  // _brokerServerBuilder = startBroker();
  _brokerServerBuilder = startBroker(_pinotHelixProperties);
  _helixManager =
      HelixManagerFactory.getZKHelixManager(helixClusterName, brokerId, InstanceType.PARTICIPANT, zkServers);
  final StateMachineEngine stateMachineEngine = _helixManager.getStateMachineEngine();
  final StateModelFactory<?> stateModelFactory =
      new BrokerResourceOnlineOfflineStateModelFactory(_helixManager, _helixExternalViewBasedRouting);
  stateMachineEngine.registerStateModelFactory(BrokerResourceOnlineOfflineStateModelFactory.getStateModelDef(),
      stateModelFactory);
  _helixManager.connect();
  _helixAdmin = _helixManager.getClusterManagmentTool();
  _helixBrokerRoutingTable = new HelixBrokerRoutingTable(_helixExternalViewBasedRouting, brokerId, _helixManager);
  addInstanceTagIfNeeded(helixClusterName, brokerId);
  _helixManager.addExternalViewChangeListener(_helixBrokerRoutingTable);
  _helixManager.addInstanceConfigChangeListener(_helixBrokerRoutingTable);
  _helixManager.addLiveInstanceChangeListener(_liveInstancesListener);

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

import org.apache.helix.participant.statemachine.StateModelFactory; //导入依赖的package包/类
@Override
public boolean removeStateModelFactory(String stateModelDef,
    StateModelFactory<? extends StateModel> factory) {
  throw new UnsupportedOperationException("Remove not yet supported");
}
 

import org.apache.helix.participant.statemachine.StateModelFactory; //导入依赖的package包/类
/**
 * test drop resource should remove all state models
 * @throws Exception
 */
@Test
public void testDrop() 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

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

  MockParticipantManager[] participants = new MockParticipantManager[n];
  for (int i = 0; i < n; i++) {
    final String instanceName = "localhost_" + (12918 + i);

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

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

  // check state-models in state-machine
  HelixStateMachineEngine stateMachine =
      (HelixStateMachineEngine) participants[0].getStateMachineEngine();
  StateModelFactory<? extends StateModel> fty = stateMachine.getStateModelFactory("MasterSlave");
  Map<String, String> expectStateModelMap = new TreeMap<String, String>();
  expectStateModelMap.put("TestDB0_0", "SLAVE");
  expectStateModelMap.put("TestDB0_1", "MASTER");
  expectStateModelMap.put("TestDB0_2", "SLAVE");
  expectStateModelMap.put("TestDB0_3", "MASTER");
  checkStateModelMap(fty, expectStateModelMap);

  // drop resource
  HelixAdmin admin = new ZKHelixAdmin(_gZkClient);
  admin.dropResource(clusterName, "TestDB0");

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

  // check state models have been dropped also
  Assert.assertTrue(fty.getPartitionSet("TestDB0").isEmpty(),
      "All state-models should be dropped, but was " + fty.getPartitionSet("TestDB0"));

  // cleanup
  controller.syncStop();
  for (int i = 0; i < n; i++) {
    participants[i].syncStop();
  }

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

import org.apache.helix.participant.statemachine.StateModelFactory; //导入依赖的package包/类
@Test()
public void testCMTaskExecutor() throws Exception {
  System.out.println("START TestCMTaskExecutor");
  String msgId = "TestMessageId";
  Message message = new Message(MessageType.TASK_REPLY, msgId);

  message.setMsgId(msgId);
  message.setSrcName("cm-instance-0");
  message.setTgtName("cm-instance-1");
  message.setTgtSessionId("1234");
  message.setFromState("Offline");
  message.setToState("Slave");
  message.setPartitionName("TestDB_0");
  message.setResourceName("TestDB");
  message.setStateModelDef("MasterSlave");

  MockManager manager = new MockManager("clusterName");
  HelixDataAccessor accessor = manager.getHelixDataAccessor();
  StateModelDefinition stateModelDef =
      new StateModelDefinition(StateModelConfigGenerator.generateConfigForMasterSlave());
  Builder keyBuilder = accessor.keyBuilder();
  accessor.setProperty(keyBuilder.stateModelDef("MasterSlave"), stateModelDef);

  MockHelixTaskExecutor executor = new MockHelixTaskExecutor();
  MockMasterSlaveStateModel stateModel = new MockMasterSlaveStateModel();
  executor.registerMessageHandlerFactory(MessageType.TASK_REPLY.name(),
      new AsyncCallbackService());

  NotificationContext context = new NotificationContext(manager);
  CurrentState currentStateDelta = new CurrentState("TestDB");
  currentStateDelta.setState("TestDB_0", "OFFLINE");

  StateModelFactory<MockMasterSlaveStateModel> stateModelFactory = new StateModelFactory<MockMasterSlaveStateModel>() {

    @Override
    public MockMasterSlaveStateModel createNewStateModel(String resource, String partitionName) {
      // TODO Auto-generated method stub
      return new MockMasterSlaveStateModel();
    }

  };
  HelixStateTransitionHandler handler =
      new HelixStateTransitionHandler(stateModelFactory, stateModel, message, context,
          currentStateDelta);

  HelixTask task = new HelixTask(message, context, handler, executor);
  executor.scheduleTask(task);
  for (int i = 0; i < 10; i++) {
    if (!executor.isDone(task.getTaskId())) {
      Thread.sleep(500);
    }
  }
  AssertJUnit.assertTrue(stateModel.stateModelInvoked);
  System.out.println("END TestCMTaskExecutor");
}
 

import org.apache.helix.participant.statemachine.StateModelFactory; //导入依赖的package包/类
@Test()
public void testInvocationAnnotated() throws Exception {
  System.out.println("START TestCMTaskHandler.testInvocationAnnotated() at "
      + new Date(System.currentTimeMillis()));
  HelixTaskExecutor executor = new HelixTaskExecutor();
  Message message = new Message(MessageType.STATE_TRANSITION, "Some unique id");
  message.setSrcName("cm-instance-0");
  message.setTgtSessionId("1234");
  message.setFromState("Offline");
  message.setToState("Slave");
  message.setPartitionName("TestDB_0");
  message.setMsgId("Some unique message id");
  message.setResourceName("TestDB");
  message.setTgtName("localhost");
  message.setStateModelDef("MasterSlave");
  message.setStateModelFactoryName(HelixConstants.DEFAULT_STATE_MODEL_FACTORY);
  MockStateModelAnnotated stateModel = new MockStateModelAnnotated();
  NotificationContext context;

  MockManager manager = new MockManager("clusterName");
  HelixDataAccessor accessor = manager.getHelixDataAccessor();

  StateModelDefinition stateModelDef =
      new StateModelDefinition(StateModelConfigGenerator.generateConfigForMasterSlave());
  Builder keyBuilder = accessor.keyBuilder();
  accessor.setProperty(keyBuilder.stateModelDef("MasterSlave"), stateModelDef);

  context = new NotificationContext(manager);

  CurrentState currentStateDelta = new CurrentState("TestDB");
  currentStateDelta.setState("TestDB_0", "OFFLINE");

  StateModelFactory<MockStateModelAnnotated> stateModelFactory =
      new StateModelFactory<MockStateModelAnnotated>() {

        @Override
        public MockStateModelAnnotated createNewStateModel(String resource, String partitionName) {
          // TODO Auto-generated method stub
          return new MockStateModelAnnotated();
        }

      };

  HelixStateTransitionHandler stHandler =
      new HelixStateTransitionHandler(stateModelFactory, stateModel, message, context,
          currentStateDelta);

  HelixTask handler = new HelixTask(message, context, stHandler, executor);
  handler.call();
  AssertJUnit.assertTrue(stateModel.stateModelInvoked);
  System.out.println("END TestCMTaskHandler.testInvocationAnnotated() at "
      + new Date(System.currentTimeMillis()));
}
 

import org.apache.helix.participant.statemachine.StateModelFactory; //导入依赖的package包/类
public HelixBrokerStarter(String brokerHost, String helixClusterName, String zkServer, Configuration pinotHelixProperties)
    throws Exception {
  LOGGER.info("Starting Pinot broker");

  _liveInstancesListener = new LiveInstancesChangeListenerImpl(helixClusterName);

  _pinotHelixProperties = DefaultHelixBrokerConfig.getDefaultBrokerConf(pinotHelixProperties);

  if (brokerHost == null) {
    brokerHost = NetUtil.getHostAddress();
  }

  final String brokerId =
      _pinotHelixProperties.getString(
              CommonConstants.Helix.Instance.INSTANCE_ID_KEY,
              CommonConstants.Helix.PREFIX_OF_BROKER_INSTANCE
                      + brokerHost
                      + "_"
                      + _pinotHelixProperties.getInt(CommonConstants.Helix.KEY_OF_BROKER_QUERY_PORT,
                  CommonConstants.Helix.DEFAULT_BROKER_QUERY_PORT));

  _pinotHelixProperties.addProperty(CommonConstants.Broker.CONFIG_OF_BROKER_ID, brokerId);
  setupHelixSystemProperties();

  // Remove all white-spaces from the list of zkServers (if any).
  String zkServers = zkServer.replaceAll("\\s+", "");

  LOGGER.info("Connecting Helix components");
  // Connect spectator Helix manager.
  _spectatorHelixManager =
      HelixManagerFactory.getZKHelixManager(helixClusterName, brokerId, InstanceType.SPECTATOR, zkServers);
  _spectatorHelixManager.connect();
  _helixAdmin = _spectatorHelixManager.getClusterManagmentTool();
  _propertyStore = _spectatorHelixManager.getHelixPropertyStore();
  _helixExternalViewBasedRouting = new HelixExternalViewBasedRouting(_propertyStore, _spectatorHelixManager,
      pinotHelixProperties.subset(ROUTING_TABLE_PARAMS_SUBSET_KEY));
  _brokerServerBuilder = startBroker(_pinotHelixProperties);
  _metricsRegistry = _brokerServerBuilder.getMetricsRegistry();
  ClusterChangeMediator clusterChangeMediator =
      new ClusterChangeMediator(_helixExternalViewBasedRouting, _brokerServerBuilder.getBrokerMetrics());
  _spectatorHelixManager.addExternalViewChangeListener(clusterChangeMediator);
  _spectatorHelixManager.addInstanceConfigChangeListener(clusterChangeMediator);
  _spectatorHelixManager.addLiveInstanceChangeListener(_liveInstancesListener);

  // Connect participant Helix manager.
  _helixManager =
      HelixManagerFactory.getZKHelixManager(helixClusterName, brokerId, InstanceType.PARTICIPANT, zkServers);
  StateMachineEngine stateMachineEngine = _helixManager.getStateMachineEngine();
  StateModelFactory<?> stateModelFactory =
      new BrokerResourceOnlineOfflineStateModelFactory(_spectatorHelixManager, _propertyStore, _helixExternalViewBasedRouting);
  stateMachineEngine.registerStateModelFactory(BrokerResourceOnlineOfflineStateModelFactory.getStateModelDef(),
      stateModelFactory);
  _helixManager.connect();
  addInstanceTagIfNeeded(helixClusterName, brokerId);

  // Register the service status handler
  ServiceStatus.setServiceStatusCallback(
      new ServiceStatus.MultipleCallbackServiceStatusCallback(ImmutableList.of(
          new ServiceStatus.IdealStateAndCurrentStateMatchServiceStatusCallback(_helixManager, helixClusterName, brokerId),
          new ServiceStatus.IdealStateAndExternalViewMatchServiceStatusCallback(_helixManager, helixClusterName, brokerId)
      )));

  _brokerServerBuilder.getBrokerMetrics().addCallbackGauge(
      "helix.connected", new Callable<Long>() {
        @Override
        public Long call() throws Exception {
          return _helixManager.isConnected() ? 1L : 0L;
        }
      });

  _helixManager.addPreConnectCallback(
      new PreConnectCallback() {
        @Override
        public void onPreConnect() {
          _brokerServerBuilder.getBrokerMetrics()
          .addMeteredGlobalValue(BrokerMeter.HELIX_ZOOKEEPER_RECONNECTS, 1L);
        }
      });
}
 

import org.apache.helix.participant.statemachine.StateModelFactory; //导入依赖的package包/类
/**
 * Register a default state model factory for a state model definition
 * A state model definition could be, for example:
 * "MasterSlave", "OnlineOffline", "LeaderStandby", etc.
 * @param stateModelDef
 * @param factory
 * @return
 */
public boolean registerStateModelFactory(String stateModelDef,
    StateModelFactory<? extends StateModel> factory);
 

import org.apache.helix.participant.statemachine.StateModelFactory; //导入依赖的package包/类
/**
 * Remove the default state model factory for a state model definition
 * @param stateModelDef
 * @param factory
 * @return
 */
public boolean removeStateModelFactory(String stateModelDef,
    StateModelFactory<? extends StateModel> factory);