@Override
  public void newClusterMetrics(Map<String, LlapMetricsCollector.LlapMetrics> newMetrics) {
    long sumAverageTime = 0;
    long sumEmptyExecutors = 0;
    long maxAverageTime = 0;
    long maxAverageTimeEmptyExecutors = 0;
    long maxAverageTimeMaxExecutors = 0;
    long workerNum = 0;
    int blacklistedNodes = 0;
    String maxAverageTimeIdentity = null;
    for (String workerIdentity : newMetrics.keySet()) {
      Map<String, Long> metrics = newMetrics.get(workerIdentity).getMetrics();
      long requestHandled = metrics.get(LlapDaemonExecutorInfo.ExecutorTotalRequestsHandled.name());
      long averageTime = metrics.get(LlapDaemonExecutorInfo.AverageResponseTime.name());
      long emptyExecutor =
          metrics.get(LlapDaemonExecutorInfo.ExecutorNumExecutorsAvailableAverage.name());
      long maxExecutors = metrics.get(LlapDaemonExecutorInfo.ExecutorNumExecutors.name());

      LOG.debug("Checking node {} with data: " +
                    "requestHandled={}, " +
                    "averageTime={}, " +
                    "emptyExecutors={}, " +
                    "maxExecutors={}",
          workerIdentity, requestHandled, averageTime, emptyExecutor, maxExecutors);

      if (maxExecutors == 0) {
        blacklistedNodes++;
        if (blacklistedNodes >= this.maxBlacklistedNodes) {
          LOG.info("Already enough blacklisted nodes {}. Skipping.", blacklistedNodes);
          return;
        } else {
          // We do not interested in the data for the blacklisted nodes
          continue;
        }
      }

      if (requestHandled > this.minServedTasksNumber) {
        workerNum++;
        sumAverageTime += averageTime;
        if (averageTime > maxAverageTime) {
          maxAverageTime = averageTime;
          maxAverageTimeEmptyExecutors = emptyExecutor;
          maxAverageTimeMaxExecutors = maxExecutors;
          maxAverageTimeIdentity = workerIdentity;
        }
        sumEmptyExecutors += emptyExecutor;
      }
    }

    // If we do not have enough data then return.
    if (workerNum < 2) {
      return;
    }

    LOG.debug("Found slowest node {} with data: " +
                  "sumAverageTime={}, " +
                  "sumEmptyExecutors={}, " +
                  "maxAverageTime={}, " +
                  "maxAverageTimeEmptyExecutors={}, " +
                  "maxAverageTimeMaxExecutors={}, " +
                  "workerNum={}, " +
                  "maxAverageTimeIdentity={}, " +
                  "blacklistedNodes={}",
        sumAverageTime, sumEmptyExecutors, maxAverageTime, maxAverageTimeEmptyExecutors,
        maxAverageTimeMaxExecutors, workerNum, maxAverageTimeIdentity, blacklistedNodes);
    // Check if the slowest node is at least timeThreshold times slower than the average
    double averageTimeWithoutSlowest = (double)(sumAverageTime - maxAverageTime) / (workerNum - 1);
    if (averageTimeWithoutSlowest * this.timeThreshold < maxAverageTime) {
      // We have a candidate, let's see if we have enough empty executors.
      long emptyExecutorsWithoutSlowest = sumEmptyExecutors - maxAverageTimeEmptyExecutors;
      if (emptyExecutorsWithoutSlowest > maxAverageTimeMaxExecutors * this.emptyExecutorsThreshold) {
        // Seems like a good candidate, let's try to blacklist
        try {
          LOG.debug("Trying to blacklist node: " + maxAverageTimeIdentity);
          setCapacity(maxAverageTimeIdentity, 0, 0);
        } catch (Throwable t) {
          LOG.debug("Can not blacklist node: " + maxAverageTimeIdentity, t);
        }
      }
    }
  }
Implementation of MetricsListener which blacklists slow nodes based on the statistics.
llap-tez/src/java/org/apache/hadoop/hive/llap/tezplugins/metrics/BlacklistingLlapMetricsListener.java