code: private void failTasksWithMaxMemory(long memoryToRelease) {
      
      List<TaskAttemptID> allTasks = new ArrayList<TaskAttemptID>();
      allTasks.addAll(processTreeInfoMap.keySet());
      // Sort the tasks descendingly according to RSS memory usage 
      Collections.sort(allTasks, new Comparator<TaskAttemptID>() {
        @Override
        public int compare(TaskAttemptID tid1, TaskAttemptID tid2) {
          return  processTreeInfoMap.get(tid2).getMemoryUsed() >
                  processTreeInfoMap.get(tid1).getMemoryUsed() ?
                  1 : -1;
        }});
      
      long memoryReleased = 0;
      // Fail the tasks one by one until the memory requirement is met
      while (memoryReleased < memoryToRelease && !allTasks.isEmpty()) {
        TaskAttemptID tid = allTasks.remove(0);
        if (!isKillable(tid)) {
          continue;
        }
        long memoryUsed = processTreeInfoMap.get(tid).getMemoryUsed();
        if (memoryUsed == 0) {
          break; // Skip tasks without process tree information currently
        }
        tasksToKill.add(tid);
        memoryReleased += memoryUsed;
      }
      if (tasksToKill.isEmpty()) {
        LOG.error("The total memory usage is over CGroup limits. "
            + "But found no alive task to kill for freeing memory.");
      } else if (memoryReleased < memoryToRelease) {
        LOG.error("The total memory usage is over CGroup limits. "
            + "But uanble to find enough tasks to kill for freeing memory.");
      }
      killTasks();
    }
docs: Starting from the tasks use the highest amount of memory,
  fail the tasks until the memory released meets the requirement
  @param memoryToRelease the mix memory to get released