private synchronized ResponseWrapper heartbeat(Collection<TezEvent> eventsArg) throws IOException,
        TezException {

      if (eventsArg != null) {
        eventsToSend.addAll(eventsArg);
      }

      TezEvent updateEvent = null;
      List<TezEvent> events = new ArrayList<TezEvent>();
      eventsToSend.drainTo(events);

      if (!task.isTaskDone() && !task.wasErrorReported()) {
        boolean sendCounters = false;
        /**
         * Increasing the heartbeat interval can delay the delivery of events. Sending just updated
         * records would save CPU in DAG AM, but certain counters are updated very frequently. Until
         * real time decisions are made based on these counters, it can be sent once per second.
         */
        // Not completely accurate, since OOB heartbeats could go out.
        if ((nonOobHeartbeatCounter.get() - prevCounterSendHeartbeatNum) * pollInterval >= sendCounterInterval) {
          sendCounters = true;
          prevCounterSendHeartbeatNum = nonOobHeartbeatCounter.get();
        }
        updateEvent = new TezEvent(getStatusUpdateEvent(sendCounters, false), updateEventMetadata);
        events.add(updateEvent);
      }

      long requestId = requestCounter.incrementAndGet();
      int fromEventId = task.getNextFromEventId();
      int fromPreRoutedEventId = task.getNextPreRoutedEventId();
      int maxEvents = Math.min(maxEventsToGet, task.getMaxEventsToHandle());
      TezHeartbeatRequest request = new TezHeartbeatRequest(requestId, events, fromPreRoutedEventId,
          containerIdStr, task.getTaskAttemptID(), fromEventId, maxEvents, 0);
      LOG.debug("Sending heartbeat to AM, request={}", request);

      maybeLogCounters();

      TezHeartbeatResponse response = umbilical.heartbeat(request);
      LOG.debug("Received heartbeat response from AM, response={}", response);

      if (response.shouldDie()) {
        LOG.info("Received should die response from AM: {}", task.getTaskAttemptID());
        askedToDie.set(true);
        return new ResponseWrapper(true, 1);
      }
      if (response.getLastRequestId() != requestId) {
        throw new TezException("AM and Task out of sync" + ", responseReqId="
            + response.getLastRequestId() + ", expectedReqId=" + requestId);
      }

      // The same umbilical is used by multiple tasks. Problematic in the case where multiple tasks
      // are running using the same umbilical.
      int numEventsReceived = 0;
      if (task.isTaskDone() || task.wasErrorReported()) {
        if (response.getEvents() != null && !response.getEvents().isEmpty()) {
          LOG.warn("Current task already complete, Ignoring all event in"
              + " heartbeat response, eventCount=" + response.getEvents().size());
        }
      } else {
        task.setNextFromEventId(response.getNextFromEventId());
        task.setNextPreRoutedEventId(response.getNextPreRoutedEventId());
        List<TezEvent> taskEvents = null;
        if (response.getEvents() != null && !response.getEvents().isEmpty()) {
          if (LOG.isInfoEnabled()) {
            LOG.info("Routing events from heartbeat response to task" + ", currentTaskAttemptId="
                + task.getTaskAttemptID() + ", eventCount=" + response.getEvents().size()
                + " fromEventId=" + fromEventId
                + " nextFromEventId=" + response.getNextFromEventId());
          }
          // This should ideally happen in a separate thread
          numEventsReceived = response.getEvents().size();
          taskEvents = response.getEvents();
        }
        if (initialEvent != null) {
          // We currently only give the initial event to the task on the first heartbeat. Given
          // that the split is ready, it seems pointless to wait, but that's how Tez works.
          List<TezEvent> oldEvents = taskEvents;
          taskEvents = new ArrayList<>(1 + (taskEvents == null ? 0 : taskEvents.size()));
          taskEvents.add(initialEvent);
          initialEvent = null;
          if (oldEvents != null) {
            taskEvents.addAll(oldEvents);
          }
        }
        if (taskEvents != null) {
          task.handleEvents(taskEvents);
        }
      }
      return new ResponseWrapper(false, numEventsReceived);
    }
@param eventsArg
@return
@throws IOException
          indicates an RPC communication failure.
@throws TezException
          indicates an exception somewhere in the AM.
llap-server/src/java/org/apache/hadoop/hive/llap/daemon/impl/LlapTaskReporter.java