}
@Override
public HelixTaskResult handleMessage() throws InterruptedException {
String type = _message.getMsgType();
HelixTaskResult result = new HelixTaskResult();
if (!type.equals(MessageType.SCHEDULER_MSG.toString())) {
throw new HelixException("Unexpected msg type for message " + _message.getMsgId()
+ " type:" + _message.getMsgType());
}
// Parse timeout value
int timeOut = -1;
if (_message.getRecord().getSimpleFields().containsKey("TIMEOUT")) {
try {
timeOut = Integer.parseInt(_message.getRecord().getSimpleFields().get("TIMEOUT"));
} catch (Exception e) {
}
}
// Parse the message template
ZNRecord record = new ZNRecord("templateMessage");
record.getSimpleFields().putAll(_message.getRecord().getMapField("MessageTemplate"));
Message messageTemplate = new Message(record);
// Parse the criteria
StringReader sr = new StringReader(_message.getRecord().getSimpleField("Criteria"));
ObjectMapper mapper = new ObjectMapper();
Criteria recipientCriteria;
try {
recipientCriteria = mapper.readValue(sr, Criteria.class);
} catch (Exception e) {
_logger.error("", e);
result.setException(e);
result.setSuccess(false);
return result;
}
_logger.info("Scheduler sending message, criteria:" + recipientCriteria);
boolean waitAll = false;
if (_message.getRecord().getSimpleField(DefaultSchedulerMessageHandlerFactory.WAIT_ALL) != null) {
try {
waitAll =
Boolean.parseBoolean(_message.getRecord().getSimpleField(
DefaultSchedulerMessageHandlerFactory.WAIT_ALL));
} catch (Exception e) {
_logger.warn("", e);
}
}
boolean hasSchedulerTaskQueue =
_message.getRecord().getSimpleFields().containsKey(SCHEDULER_TASK_QUEUE);
// If the target is PARTICIPANT, use the ScheduledTaskQueue
if (InstanceType.PARTICIPANT == recipientCriteria.getRecipientInstanceType()
&& hasSchedulerTaskQueue) {
handleMessageUsingScheduledTaskQueue(recipientCriteria, messageTemplate,
_message.getMsgId());
result.setSuccess(true);
result.getTaskResultMap().put(SCHEDULER_MSG_ID, _message.getMsgId());
result.getTaskResultMap().put("ControllerResult",
"msg " + _message.getMsgId() + " from " + _message.getMsgSrc() + " processed");
return result;
}
_logger.info("Scheduler sending message to Controller");
int nMsgsSent = 0;
SchedulerAsyncCallback callback = new SchedulerAsyncCallback(_message, _manager);
if (waitAll) {
nMsgsSent =
_manager.getMessagingService().sendAndWait(recipientCriteria, messageTemplate,
callback, timeOut);
} else {
nMsgsSent =
_manager.getMessagingService().send(recipientCriteria, messageTemplate, callback,
timeOut);
}
HelixDataAccessor accessor = _manager.getHelixDataAccessor();
Builder keyBuilder = accessor.keyBuilder();
// Record the number of messages sent into status updates
Map<String, String> sendSummary = new HashMap<String, String>();
sendSummary.put("MessageCount", "" + nMsgsSent);
ZNRecord statusUpdate =
accessor.getProperty(
keyBuilder.controllerTaskStatus(MessageType.SCHEDULER_MSG.toString(),
_message.getMsgId())).getRecord();
statusUpdate.getMapFields().put("SentMessageCount", sendSummary);
accessor.setProperty(keyBuilder.controllerTaskStatus(MessageType.SCHEDULER_MSG.toString(),
_message.getMsgId()), new StatusUpdate(statusUpdate));
result.getTaskResultMap().put("ControllerResult",
"msg " + _message.getMsgId() + " from " + _message.getMsgSrc() + " processed");
result.getTaskResultMap().put(SCHEDULER_MSG_ID, _message.getMsgId());
result.setSuccess(true);
return result;
}