private SelectHostResult selectHost(TaskInfo request, Map<String, List<NodeInfo>> availableHostMap) {
    // short-circuit when no-active instances exist
    if (availableHostMap.isEmpty()) {
      return SELECT_HOST_RESULT_DELAYED_RESOURCES;
    }
    String[] requestedHosts = request.requestedHosts;
    // The requestedHostsDebugStr is merely used by debug-logging calls,
    //   and it is properly assigned in the following isDebugEnabled block.
    String requestedHostsDebugStr = null;
    if (LOG.isDebugEnabled()) {
      requestedHostsDebugStr = Arrays.toString(requestedHosts);
      LOG.debug("selectingHost for task={} on hosts={}", request.task,
          requestedHostsDebugStr);
    }
    long schedulerAttemptTime = clock.getTime();
    readLock.lock(); // Read-lock. Not updating any stats at the moment.
    try {
      boolean shouldDelayForLocality = request.shouldDelayForLocality(schedulerAttemptTime);
      LOG.debug("ShouldDelayForLocality={} for task={} on hosts={}", shouldDelayForLocality,
          request.task, requestedHostsDebugStr);
      if (isRequestedHostPresent(request)) {
        int prefHostCount = -1;
        boolean requestedHostsWillBecomeAvailable = false;
        for (String host : requestedHosts) {
          prefHostCount++;

          // Check if the host is removed from the registry after availableHostMap is created.
          Set<LlapServiceInstance> activeInstancesByHost = activeInstances.getByHost(host);
          if (activeInstancesByHost == null || activeInstancesByHost.isEmpty()) {
            continue;
          }

          // Pick the first host always. Weak attempt at cache affinity.
          if (availableHostMap.containsKey(host)) {
            List<NodeInfo> instances = availableHostMap.getOrDefault(host, new ArrayList<>());
            // Host is there and there are available resources!
            if (!instances.isEmpty()) {
              NodeInfo nodeInfo = instances.iterator().next();
              // Successfully scheduled.
              LOG.info("Assigning {} when looking for {}." + " local=true FirstRequestedHost={}, #prefLocations={}",
                  nodeInfo.toShortString(), host, (prefHostCount == 0), requestedHosts.length);
              return new SelectHostResult(nodeInfo);
            } else {
              // The node cannot accept a task at the moment. (there is host -- but no resources)
              if (shouldDelayForLocality) {
                // Perform some checks on whether the node will become available or not.
                if (request.shouldForceLocality()) {
                  requestedHostsWillBecomeAvailable = true;
                } else {
                  for (LlapServiceInstance inst : activeInstancesByHost) {
                    NodeInfo nodeInfo = instanceToNodeMap.get(inst.getWorkerIdentity());
                    if (nodeInfo == null) {
                      LOG.warn("Null NodeInfo when attempting to get host {}", host);
                      // Leave requestedHostWillBecomeAvailable as is. If some other host is found - delay,
                      // else ends up allocating to a random host immediately.
                      continue;
                    }
                    if (nodeInfo.getEnableTime() > request.getLocalityDelayTimeout()
                            && nodeInfo.isDisabled() && nodeInfo.hadCommFailure()) {
                      LOG.debug("Host={} will not become available within requested timeout", nodeInfo);
                      // This node will likely be activated after the task timeout expires.
                    } else {
                      // Worth waiting for the timeout.
                      requestedHostsWillBecomeAvailable = true;
                    }
                  }
                }
              }
            }
          }
        }
        // Check if forcing the location is required.
        if (shouldDelayForLocality) {
          if (requestedHostsWillBecomeAvailable) {
            if (LOG.isDebugEnabled()) {
              LOG.debug("Delaying local allocation for [" + request.task +
                  "] when trying to allocate on [" + requestedHostsDebugStr + "]" +
                  ". ScheduleAttemptTime=" + schedulerAttemptTime + ", taskDelayTimeout=" +
                  request.getLocalityDelayTimeout());
            }
            return SELECT_HOST_RESULT_DELAYED_LOCALITY;
          } else {
            if (LOG.isDebugEnabled()) {
              LOG.debug("Skipping local allocation for [" + request.task +
                  "] when trying to allocate on [" + requestedHostsDebugStr +
                  "] since none of these hosts are part of the known list");
            }
          }
        }
      }
      /* fall through - miss in locality or no locality-requested */
      List<NodeInfo> activeNodesWithFreeSlots = new ArrayList<>();
      availableHostMap.values().forEach(activeNodesWithFreeSlots::addAll);

      // no locality-requested, randomly pick a node containing free slots
      if (requestedHosts == null || requestedHosts.length == 0) {
        if (LOG.isDebugEnabled()) {
          LOG.debug("No-locality requested. Selecting a random host for task={}", request.task);
        }
        return randomSelection(activeNodesWithFreeSlots);
      }

      // miss in locality request, try picking consistent location with fallback to random selection
      final String firstRequestedHost = requestedHosts[0];
      // requested host died or unknown host requested, fallback to random selection.
      // TODO: At this point we don't know the slot number of the requested host, so can't rollover to next available
      if (!availableHostMap.containsKey(firstRequestedHost)) {
        if (LOG.isDebugEnabled()) {
          LOG.debug("Requested node [{}] in consistent order does not exist. Falling back to random selection for " +
            "request {}", firstRequestedHost, request);
        }
        return randomSelection(activeNodesWithFreeSlots);
      }

      // requested host is still alive but cannot accept task, pick the next available host in consistent order
      if (!activeNodesWithFreeSlots.isEmpty()) {
        NodeInfo nextSlot = null;
        boolean found = false;
        for (Entry<String, List<NodeInfo>> entry : availableHostMap.entrySet()) {
          if (found && !entry.getValue().isEmpty()) {
            nextSlot = entry.getValue().iterator().next();
            break;
          }
          if (entry.getKey().equals(firstRequestedHost)) found = true;
        }
        // rollover
        if (nextSlot == null) nextSlot = activeNodesWithFreeSlots.stream().findFirst().get();
        LOG.info("Assigning {} in consistent order when looking for first requested host, from #hosts={},"
                + " requestedHosts={}", nextSlot.toShortString(), availableHostMap.size(),
            ((requestedHosts == null || requestedHosts.length == 0) ? "null" : requestedHostsDebugStr));
        return new SelectHostResult(nextSlot);
      }
      // When all nodes are busy, reset locality delay
      if (request.localityDelayTimeout > 0 && isRequestedHostPresent(request)) {
        request.resetLocalityDelayInfo();
      }
      return SELECT_HOST_RESULT_DELAYED_RESOURCES;
    } finally {
      readLock.unlock();
    }
  }
@param request the list of preferred hosts. null implies any host
@return
llap-tez/src/java/org/apache/hadoop/hive/llap/tezplugins/LlapTaskSchedulerService.java