@Override
    protected Void callInternal() {
      while (!isShutdown.get() && !Thread.currentThread().isInterrupted()) {
        try {
          final AMNodeInfo amNodeInfo = pendingHeartbeatQueeu.take();
          if (amNodeInfo.hasAmFailed() || amNodeInfo.isDone()) {
            synchronized (knownAppMasters) {
              if (LOG.isDebugEnabled()) {
                LOG.debug(
                    "Removing am {} with last associated dag {} from heartbeat with taskCount={}, amFailed={}, isDone={}",
                    amNodeInfo.amNodeId, amNodeInfo.getQueryIdentifier(), amNodeInfo.getTaskCount(),
                    amNodeInfo.hasAmFailed(), amNodeInfo.isDone());
              }
              knownAppMasters.remove(amNodeInfo.getQueryIdentifier());
            }
          } else {
            // Always re-schedule the next callable - irrespective of task count,
            // in case new tasks come in later.
            long next = System.currentTimeMillis() + heartbeatInterval;
            amNodeInfo.setNextHeartbeatTime(next);
            pendingHeartbeatQueeu.add(amNodeInfo);

            // Send an actual heartbeat only if the task count is > 0
            if (amNodeInfo.getTaskCount() > 0) {
              // Add back to the queue for the next heartbeat, and schedule the actual heartbeat
              ListenableFuture<Void> future = executor.submit(new AMHeartbeatCallable(amNodeInfo));
              Futures.addCallback(future, new FutureCallback<Void>() {
                @Override
                public void onSuccess(Void result) {
                  // Nothing to do.
                }

                @Override
                public void onFailure(Throwable t) {
                  QueryIdentifier currentQueryIdentifier = amNodeInfo.getQueryIdentifier();
                  amNodeInfo.setAmFailed(true);
                  LOG.warn("Heartbeat failed to AM {}. Marking query as failed. query={}",
                    amNodeInfo.amNodeId, currentQueryIdentifier, t);
                  queryFailedHandler.queryFailed(currentQueryIdentifier);
                }
              }, MoreExecutors.directExecutor());
            }
          }
        } catch (InterruptedException e) {
          if (isShutdown.get()) {
            LOG.info("QueueLookup thread interrupted after shutdown");
          } else {
            LOG.warn("Received unexpected interrupt while waiting on heartbeat queue");
          }
        }

      }
      return null;
    }
Responsible for communicating with various AMs.
llap-server/src/java/org/apache/hadoop/hive/llap/daemon/impl/AMReporter.java