public HelixTaskResult handleMessage()

in helix-core/src/main/java/org/apache/helix/manager/zk/DefaultSchedulerMessageHandlerFactory.java [250:342]
79 lines of code
10 McCabe index (conditional complexity)

    public HelixTaskResult handleMessage() throws InterruptedException {
      String type = _message.getMsgType();
      HelixTaskResult result = new HelixTaskResult();
      if (!type.equals(MessageType.SCHEDULER_MSG.name())) {
        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"));
      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.name(),
                  _message.getMsgId())).getRecord();

      statusUpdate.getMapFields().put("SentMessageCount", sendSummary);

      accessor.setProperty(keyBuilder.controllerTaskStatus(MessageType.SCHEDULER_MSG.name(),
          _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;
    }