import org.apache.helix.model.Message; //导入依赖的package包/类
private void sendShutdownRequest() {
  Criteria criteria = new Criteria();
  criteria.setInstanceName("%");
  criteria.setResource("%");
  criteria.setPartition("%");
  criteria.setPartitionState("%");
  criteria.setRecipientInstanceType(InstanceType.CONTROLLER);
  criteria.setSessionSpecific(true);

  Message shutdownRequest = new Message(Message.MessageType.SHUTDOWN,
      HelixMessageSubTypes.APPLICATION_MASTER_SHUTDOWN.toString().toLowerCase() + UUID.randomUUID().toString());
  shutdownRequest.setMsgSubType(HelixMessageSubTypes.APPLICATION_MASTER_SHUTDOWN.toString());
  shutdownRequest.setMsgState(Message.MessageState.NEW);
  shutdownRequest.setTgtSessionId("*");

  int messagesSent = this.helixManager.getMessagingService().send(criteria, shutdownRequest);
  if (messagesSent == 0) {
    LOGGER.error(String.format("Failed to send the %s message to the controller", shutdownRequest.getMsgSubType()));
  }
}
 

import org.apache.helix.model.Message; //导入依赖的package包/类
private void sendTokenFileUpdatedMessage(InstanceType instanceType) {
  Criteria criteria = new Criteria();
  criteria.setInstanceName("%");
  criteria.setResource("%");
  criteria.setPartition("%");
  criteria.setPartitionState("%");
  criteria.setRecipientInstanceType(instanceType);
  if (instanceType == InstanceType.PARTICIPANT) {
    criteria.setDataSource(Criteria.DataSource.LIVEINSTANCES);
  }
  criteria.setSessionSpecific(true);

  Message tokenFileUpdatedMessage = new Message(Message.MessageType.USER_DEFINE_MSG,
      HelixMessageSubTypes.TOKEN_FILE_UPDATED.toString().toLowerCase() + UUID.randomUUID().toString());
  tokenFileUpdatedMessage.setMsgSubType(HelixMessageSubTypes.TOKEN_FILE_UPDATED.toString());
  tokenFileUpdatedMessage.setMsgState(Message.MessageState.NEW);
  if (instanceType == InstanceType.CONTROLLER) {
    tokenFileUpdatedMessage.setTgtSessionId("*");
  }

  int messagesSent = this.helixManager.getMessagingService().send(criteria, tokenFileUpdatedMessage);
  LOGGER.info(String.format("Sent %d token file updated message(s) to the %s", messagesSent, instanceType));
}
 

import org.apache.helix.model.Message; //导入依赖的package包/类
private void sendShutdownRequest() {
  Criteria criteria = new Criteria();
  criteria.setInstanceName("%");
  criteria.setResource("%");
  criteria.setPartition("%");
  criteria.setPartitionState("%");
  criteria.setRecipientInstanceType(InstanceType.PARTICIPANT);
  criteria.setDataSource(Criteria.DataSource.LIVEINSTANCES);
  criteria.setSessionSpecific(true);

  Message shutdownRequest = new Message(Message.MessageType.SHUTDOWN,
      HelixMessageSubTypes.WORK_UNIT_RUNNER_SHUTDOWN.toString().toLowerCase() + UUID.randomUUID().toString());
  shutdownRequest.setMsgSubType(HelixMessageSubTypes.WORK_UNIT_RUNNER_SHUTDOWN.toString());
  shutdownRequest.setMsgState(Message.MessageState.NEW);

  int messagesSent = this.helixManager.getMessagingService().send(criteria, shutdownRequest);
  if (messagesSent == 0) {
    LOGGER.error(String.format("Failed to send the %s message to the participants", shutdownRequest.getMsgSubType()));
  }
}
 

import org.apache.helix.model.Message; //导入依赖的package包/类
@Override
public HelixTaskResult handleMessage() throws InterruptedException {
  String messageSubType = this._message.getMsgSubType();

  if (messageSubType.equalsIgnoreCase(HelixMessageSubTypes.TOKEN_FILE_UPDATED.toString())) {
    LOGGER.info("Handling message " + HelixMessageSubTypes.TOKEN_FILE_UPDATED.toString());

    eventBus.post(new DelegationTokenUpdatedEvent());
    HelixTaskResult helixTaskResult = new HelixTaskResult();
    helixTaskResult.setSuccess(true);
    return helixTaskResult;
  }

  throw new IllegalArgumentException(String.format("Unknown %s message subtype: %s",
      Message.MessageType.USER_DEFINE_MSG.toString(), messageSubType));
}
 

import org.apache.helix.model.Message; //导入依赖的package包/类
@Transition(from = "OFFLINE", to = "ONLINE")
public void onBecomeOnlineFromOffline(Message message,
                                      NotificationContext context) {
  Pair<String, String> hdfsPathAndPartition = getHdfsPathAndPartitionNum(message);
  String hdfsPath = hdfsPathAndPartition.getLeft();
  LOG.info("Opening " + hdfsPath);
  try {
    // TODO(varun): Maybe retry here.
    HColumnDescriptor family = new HColumnDescriptor(Constants.HFILE_COLUMN_FAMILY);
    family.setBlockCacheEnabled(isBlockCacheEnabled);
    Reader r = readerFactory.createHFileReader(hdfsPath, new CacheConfig(conf, family));
    resourcePartitionMap.addReader(
        message.getResourceName(), hdfsPathAndPartition.getRight(), r);
  } catch (IOException e) {
    throw new RuntimeException(e);
  }
}
 

import org.apache.helix.model.Message; //导入依赖的package包/类
@Transition(from = "OFFLINE", to = "ONLINE")
public void onBecomeOnlineFromOffline(Message message, NotificationContext context) {

  try {
    LOGGER.info("BrokerResourceOnlineOfflineStateModel.onBecomeOnlineFromOffline() : " + message);
    Builder keyBuilder = _helixManager.getHelixDataAccessor().keyBuilder();
    String resourceName = message.getPartitionName();
    HelixDataAccessor helixDataAccessor = _helixManager.getHelixDataAccessor();
    List<InstanceConfig> instanceConfigList = helixDataAccessor.getChildValues(keyBuilder.instanceConfigs());

    _helixExternalViewBasedRouting.markDataResourceOnline(
        resourceName,
        HelixHelper.getExternalViewForResouce(_helixManager.getClusterManagmentTool(),
            _helixManager.getClusterName(), resourceName), instanceConfigList);
  } catch (Exception e) {
    LOGGER.error("Caught exception during OFFLINE -> ONLINE transition", e);
    Utils.rethrowException(e);
    throw new AssertionError("Should not reach this");
  }
}
 

import org.apache.helix.model.Message; //导入依赖的package包/类
@Transition(from = "OFFLINE", to = "ONLINE")
public void onBecomeOnlineFromOffline(Message message, NotificationContext context) {
  LOGGER.debug("SegmentOnlineOfflineStateModel.onBecomeOnlineFromOffline() : " + message);
  final TableType tableType = TableNameBuilder.getTableTypeFromTableName(message.getResourceName());
  try {
    switch (tableType) {
      case OFFLINE:
        onBecomeOnlineFromOfflineForOfflineSegment(message, context);
        break;
      case REALTIME:
        onBecomeOnlineFromOfflineForRealtimeSegment(message, context);
        break;

      default:
        throw new RuntimeException("Not supported table Type for onBecomeOnlineFromOffline message: " + message);
    }
  } catch (Exception e) {
    if (LOGGER.isErrorEnabled()) {
      LOGGER.error(
          "Caught exception in state transition for OFFLINE -> ONLINE for partition" + message.getPartitionName()
              + " of table " + message.getResourceName(), e);
    }
    Utils.rethrowException(e);
  }
}
 

import org.apache.helix.model.Message; //导入依赖的package包/类
@Override
public HelixTaskResult handleMessage() throws InterruptedException {
  String messageSubType = this._message.getMsgSubType();

  if (messageSubType.equalsIgnoreCase(org.apache.gobblin.cluster.HelixMessageSubTypes.TOKEN_FILE_UPDATED.toString())) {
    LOGGER.info("Handling message " + org.apache.gobblin.cluster.HelixMessageSubTypes.TOKEN_FILE_UPDATED.toString());

    eventBus.post(new DelegationTokenUpdatedEvent());
    HelixTaskResult helixTaskResult = new HelixTaskResult();
    helixTaskResult.setSuccess(true);
    return helixTaskResult;
  }

  throw new IllegalArgumentException(String
      .format("Unknown %s message subtype: %s", Message.MessageType.USER_DEFINE_MSG.toString(), messageSubType));
}
 

import org.apache.helix.model.Message; //导入依赖的package包/类
/**
 * Create a ZNRecord for a message, which stores the content of the message (stored in
 * simple fields) into the ZNRecord mapFields. In this way, the message update can be
 * merged with the previous status update record in the zookeeper. See ZNRecord.merge()
 * for more details.
 */
ZNRecord createMessageLogRecord(Message message) {
  ZNRecord result = new ZNRecord(getStatusUpdateRecordName(message));
  String mapFieldKey = "MESSAGE " + message.getMsgId();
  result.setMapField(mapFieldKey, new TreeMap<String, String>());

  // Store all the simple fields of the message in the new ZNRecord's map
  // field.
  for (String simpleFieldKey : message.getRecord().getSimpleFields().keySet()) {
    result.getMapField(mapFieldKey).put(simpleFieldKey,
        message.getRecord().getSimpleField(simpleFieldKey));
  }
  if (message.getResultMap() != null) {
    result.setMapField("MessageResult", message.getResultMap());
  }
  return result;
}
 

import org.apache.helix.model.Message; //导入依赖的package包/类
@Transition(to = "TIMED_OUT", from = "RUNNING")
public String onBecomeTimedOutFromRunning(Message msg, NotificationContext context) {
  String taskPartition = msg.getPartitionName();
  if (_taskRunner == null) {
    throw new IllegalStateException(String.format(
        "Invalid state transition. There is no running task for partition %s.", taskPartition));
  }

  TaskResult r = _taskRunner.waitTillDone();
  if (r.getStatus() != TaskResult.Status.CANCELED) {
    throw new IllegalStateException(String.format(
        "Partition %s received a state transition to %s but the result status code is %s.",
        msg.getPartitionName(), msg.getToState(), r.getStatus()));
  }

  timeout_task.cancel(false);

  return r.getInfo();
}
 

import org.apache.helix.model.Message; //导入依赖的package包/类
public synchronized void reportProcessedMessage(Message message,
    ParticipantMessageMonitor.ProcessedMessageState processedMessageState) {
  if (_messageMonitor != null) {  // is participant
    switch (processedMessageState) {
      case DISCARDED:
        _messageMonitor.incrementDiscardedMessages(1);
        _messageMonitor.decrementPendingMessages(1);
        break;
      case FAILED:
        _messageMonitor.incrementFailedMessages(1);
        _messageMonitor.decrementPendingMessages(1);
        break;
      case COMPLETED:
        _messageMonitor.incrementCompletedMessages(1);
        _messageMonitor.decrementPendingMessages(1);
        break;
    }
  }
}
 

import org.apache.helix.model.Message; //导入依赖的package包/类
@Transition(to = "COMPLETED", from = "OFFLINE")
public void onBecomeCompletedFromOffline(Message message, NotificationContext context)
    throws InterruptedException {
  logger.info(_partitionKey + " onBecomeCompletedFromOffline");

  // Construct the inner task message from the mapfields of scheduledTaskQueue resource group
  Map<String, String> messageInfo =
      message.getRecord().getMapField(Message.Attributes.INNER_MESSAGE.toString());
  ZNRecord record = new ZNRecord(_partitionKey);
  record.getSimpleFields().putAll(messageInfo);
  Message taskMessage = new Message(record);
  if (logger.isDebugEnabled()) {
    logger.debug(taskMessage.getRecord().getSimpleFields().toString());
  }
  MessageHandler handler =
      _executor.createMessageHandler(taskMessage, new NotificationContext(null));
  if (handler == null) {
    throw new HelixException("Task message " + taskMessage.getMsgType()
        + " handler not found, task id " + _partitionKey);
  }
  // Invoke the internal handler to complete the task
  handler.handleMessage();
  logger.info(_partitionKey + " onBecomeCompletedFromOffline completed");
}
 

import org.apache.helix.model.Message; //导入依赖的package包/类
public MessageHandler createMessageHandler(Message message, NotificationContext changeContext) {
  String msgType = message.getMsgType().toString();

  MsgHandlerFactoryRegistryItem item = _hdlrFtyRegistry.get(msgType);

  // Fail to find a MessageHandlerFactory for the message
  // we will keep the message and the message will be handled when
  // the corresponding MessageHandlerFactory is registered
  if (item == null) {
    LOG.warn("Fail to find message handler factory for type: " + msgType + " msgId: "
        + message.getMsgId());
    return null;
  }
  MessageHandlerFactory handlerFactory = item.factory();

  // pass the executor to msg-handler since batch-msg-handler needs task-executor to schedule
  // sub-msgs
  changeContext.add(MapKey.TASK_EXECUTOR.toString(), this);
  return handlerFactory.createHandler(message, changeContext);
}
 

import org.apache.helix.model.Message; //导入依赖的package包/类
private Message newMessage(String resourceName, String partitionName, String instanceName,
    String fromState, String toState) {
  String uuid = UUID.randomUUID().toString();
  Message message = new Message(MessageType.STATE_TRANSITION, uuid);
  message.setSrcName("controller");
  message.setTgtName(instanceName);
  message.setMsgState(MessageState.NEW);
  message.setResourceName(resourceName);
  message.setPartitionName(partitionName);
  message.setFromState(fromState);
  message.setToState(toState);
  message.setTgtSessionId("sessionId");
  message.setSrcSessionId("sessionId");
  message.setStateModelDef("MasterSlave");
  message.setStateModelFactoryName("DEFAULT");
  message.setBucketSize(0);
  return message;
}
 

import org.apache.helix.model.Message; //导入依赖的package包/类
public void onBecomeOfflineFromOnline(Message message, NotificationContext context) {
  MockProcess.sleep(_transDelay);

  logger.info(
      "MockStateModel.onBecomeOfflineFromOnline(), resource " + message.getResourceName()
          + ", partition"
          + message.getPartitionName() + ", targetName: " + message.getTgtName());

  try {
    Thread.sleep(1);
  } catch (InterruptedException e) {
    e.printStackTrace();
  }

  verifyMessage(message);
}
 

import org.apache.helix.model.Message; //导入依赖的package包/类
/**
 * Start the ApplicationMaster.
 */
public void start() {
  LOGGER.info("Starting the Gobblin Yarn ApplicationMaster");

  // Add a shutdown hook so the task scheduler gets properly shutdown
  addShutdownHook();

  this.eventBus.register(this);

  try {
    this.helixManager.connect();
    this.helixManager.addLiveInstanceChangeListener(new GobblinLiveInstanceChangeListener());
    this.helixManager.getMessagingService().registerMessageHandlerFactory(
        Message.MessageType.SHUTDOWN.toString(), new ControllerShutdownMessageHandlerFactory());
    this.helixManager.getMessagingService().registerMessageHandlerFactory(
        Message.MessageType.USER_DEFINE_MSG.toString(), new ControllerUserDefinedMessageHandlerFactory()
    );
  } catch (Exception e) {
    LOGGER.error("HelixManager failed to connect", e);
    throw Throwables.propagate(e);
  }

  // Register JVM metrics to collect and report
  registerJvmMetrics();

  // Start metric reporting
  this.jmxReporter.start();

  // Start all the services running in the ApplicationMaster
  this.serviceManager.startAsync();
  this.serviceManager.awaitHealthy();
}
 

import org.apache.helix.model.Message; //导入依赖的package包/类
/**
 * Start this {@link GobblinWorkUnitRunner} instance.
 */
public void start() {
  LOGGER.info(String.format("Starting %s in container %s",
      GobblinWorkUnitRunner.class.getSimpleName(), this.containerId));

  // Add a shutdown hook so the task scheduler gets properly shutdown
  addShutdownHook();

  try {
    this.helixManager.connect();
    this.helixManager.getMessagingService().registerMessageHandlerFactory(
        Message.MessageType.SHUTDOWN.toString(), new ParticipantShutdownMessageHandlerFactory());
    this.helixManager.getMessagingService().registerMessageHandlerFactory(
        Message.MessageType.USER_DEFINE_MSG.toString(), new ParticipantUserDefinedMessageHandlerFactory());
  } catch (Exception e) {
    LOGGER.error("HelixManager failed to connect", e);
    throw Throwables.propagate(e);
  }

  // Register JVM metrics to collect and report
  registerJvmMetrics();
  // Start metric reporting
  this.jmxReporter.start();

  this.serviceManager.startAsync();
  this.serviceManager.awaitStopped();
}
 

import org.apache.helix.model.Message; //导入依赖的package包/类
static Pair<String, String> getHdfsPathAndPartitionNum(Message message) {
  // Since the partition names in helix are of the form resource$120 etc., we need to
  // strip out the resource from partition name.
  String partitionNum = message.getPartitionName().substring(
      message.getResourceName().length() + 1);
  String partitionName = TerrapinUtil.formatPartitionName(Integer.parseInt(partitionNum));
  String hdfsPath = TerrapinUtil.helixResourceToHdfsDir(message.getResourceName()) +
      "/" + partitionName;

  return new ImmutablePair(hdfsPath, partitionNum);
}
 

import org.apache.helix.model.Message; //导入依赖的package包/类
@Transition(from = "ONLINE", to = "OFFLINE")
public void onBecomeOfflineFromOnline(Message message,
                                      NotificationContext context) {
  Pair<String, String> hdfsPathAndPartition = getHdfsPathAndPartitionNum(message);
  String hdfsPath = hdfsPathAndPartition.getLeft();
  LOG.info("Closing " + hdfsPath);
  try {
    Reader r = resourcePartitionMap.removeReader(message.getResourceName(),
        hdfsPathAndPartition.getRight());
    r.close();
  } catch (Exception e) {
    LOG.warn("Could not close reader for " + hdfsPath, e);
  }
}
 

import org.apache.helix.model.Message; //导入依赖的package包/类
@Transition(from = "ONLINE", to = "OFFLINE")
public void onBecomeOfflineFromOnline(Message message, NotificationContext context) {
  try {
    LOGGER.info("BrokerResourceOnlineOfflineStateModel.onBecomeOfflineFromOnline() : " + message);
    String resourceName = message.getResourceName();
    _helixExternalViewBasedRouting.markDataResourceOffline(resourceName);
  } catch (Exception e) {
    LOGGER.error("Caught exception during ONLINE -> OFFLINE transition", e);
    Utils.rethrowException(e);
    throw new AssertionError("Should not reach this");
  }
}