@Override
  public void registerRunningTaskAttempt(final ContainerId containerId, final TaskSpec taskSpec,
                                         Map<String, LocalResource> additionalResources,
                                         Credentials credentials,
                                         boolean credentialsChanged,
                                         int priority)  {
    super.registerRunningTaskAttempt(containerId, taskSpec, additionalResources, credentials,
        credentialsChanged, priority);
    int dagId = taskSpec.getDAGID().getId();
    if (currentQueryIdentifierProto == null || (dagId != currentQueryIdentifierProto.getDagIndex())) {
      String hiveQueryId = extractQueryIdFromContext();
      try {
        hiveQueryId = (hiveQueryId == null) ? extractQueryId(taskSpec) : hiveQueryId;
      } catch (IOException e) {
        throw new RuntimeException("Failed to extract query id from task spec: " + taskSpec, e);
      }
      Preconditions.checkNotNull(hiveQueryId, "Unexpected null query id");

      resetCurrentDag(dagId, hiveQueryId);
    }

    ContainerInfo containerInfo = getContainerInfo(containerId);
    String host;
    int port;
    if (containerInfo != null) {
      synchronized (containerInfo) {
        host = containerInfo.host;
        port = containerInfo.port;
      }
    } else {
      // TODO Handle this properly
      throw new RuntimeException("ContainerInfo not found for container: " + containerId +
          ", while trying to launch task: " + taskSpec.getTaskAttemptID());
    }

    LlapNodeId nodeId = LlapNodeId.getInstance(host, port);
    registerKnownNode(nodeId);
    entityTracker.registerTaskAttempt(containerId, taskSpec.getTaskAttemptID(), host, port);
    nodesForQuery.add(nodeId);

    sourceStateTracker.registerTaskForStateUpdates(host, port, taskSpec.getInputs());
    FragmentRuntimeInfo fragmentRuntimeInfo;
    try {
      fragmentRuntimeInfo = sourceStateTracker.getFragmentRuntimeInfo(
          taskSpec.getVertexName(),
          taskSpec.getTaskAttemptID().getTaskID().getId(), priority);
    } catch (Exception e) {
      LOG.error(
          "Error while trying to get runtimeFragmentInfo for fragmentId={}, containerId={}, currentQI={}, currentQueryId={}",
          taskSpec.getTaskAttemptID(), containerId, currentQueryIdentifierProto,
          currentHiveQueryId, e);
      if (e instanceof RuntimeException) {
        throw (RuntimeException) e;
      } else {
        throw new RuntimeException(e);
      }
    }
    SubmitWorkRequestProto requestProto;

    try {
      requestProto = constructSubmitWorkRequest(containerId, taskSpec, fragmentRuntimeInfo, currentHiveQueryId);
    } catch (IOException e) {
      throw new RuntimeException("Failed to construct request", e);
    }

    // Have to register this up front right now. Otherwise, it's possible for the task to start
    // sending out status/DONE/KILLED/FAILED messages before TAImpl knows how to handle them.
    getContext().taskStartedRemotely(taskSpec.getTaskAttemptID(), containerId);
    communicator.sendSubmitWork(requestProto, host, port,
        new LlapProtocolClientProxy.ExecuteRequestCallback<SubmitWorkResponseProto>() {
          @Override
          public void setResponse(SubmitWorkResponseProto response) {
            if (response.hasSubmissionState()) {
              LlapDaemonProtocolProtos.SubmissionStateProto ss = response.getSubmissionState();
              if (ss.equals(LlapDaemonProtocolProtos.SubmissionStateProto.REJECTED)) {
                LOG.info(
                    "Unable to run task: " + taskSpec.getTaskAttemptID() + " on containerId: " +
                        containerId + ", Service Busy");
                getContext().taskKilled(taskSpec.getTaskAttemptID(),
                    TaskAttemptEndReason.EXECUTOR_BUSY, "Service Busy");
                return;
              }
            } else {
              // TODO: Provide support for reporting errors
              // This should never happen as server always returns a valid status on success
              throw new RuntimeException("SubmissionState in response is expected!");
            }
            if (response.hasUniqueNodeId()) {
              entityTracker.registerTaskSubmittedToNode(
                  taskSpec.getTaskAttemptID(), response.getUniqueNodeId());
            }
            LOG.info("Successfully launched task: " + taskSpec.getTaskAttemptID());
            scheduler.notifyStarted(taskSpec.getTaskAttemptID());
          }

          @Override
          public void indicateError(Throwable t) {
             Throwable originalError = t;
            if (t instanceof ServiceException) {
              ServiceException se = (ServiceException) t;
              t = se.getCause();
            }

            if (t instanceof RemoteException) {
              // All others from the remote service cause the task to FAIL.
              LOG.info(
                  "Failed to run task: " + taskSpec.getTaskAttemptID() + " on containerId: " +
                      containerId, t);
              processSendError(originalError);
              getContext()
                  .taskFailed(taskSpec.getTaskAttemptID(), TaskFailureType.NON_FATAL, TaskAttemptEndReason.OTHER,
                      t.toString());
            } else {
              // Exception from the RPC layer - communication failure, consider as KILLED / service down.
              if (t instanceof IOException) {
                LOG.info(
                    "Unable to run task: " + taskSpec.getTaskAttemptID() + " on containerId: " +
                        containerId + ", Communication Error");
                processSendError(originalError);
                getContext().taskKilled(taskSpec.getTaskAttemptID(),
                    TaskAttemptEndReason.COMMUNICATION_ERROR, "Communication Error");
              } else {
                // Anything else is a FAIL.
                LOG.info(
                    "Failed to run task: " + taskSpec.getTaskAttemptID() + " on containerId: " +
                        containerId, t);
                processSendError(originalError);
                getContext()
                    .taskFailed(taskSpec.getTaskAttemptID(), TaskFailureType.NON_FATAL, TaskAttemptEndReason.OTHER,
                        t.getMessage());
              }
            }
          }
        });
  }
@param node
@param callback
@return if it was possible to attempt the registration. Sometimes it's
not possible because is a dag is not running
llap-tez/src/java/org/apache/hadoop/hive/llap/tezplugins/LlapTaskCommunicator.java