code: public synchronized Task obtainNewReduceTask(TaskTrackerStatus tts,
                                                 int clusterSize,
                                                 int numUniqueHosts
                                                ) throws IOException {
      if (status.getRunState() != JobStatus.RUNNING) {
        LOG.info("Cannot create task split for " + profile.getJobID());
        return null;
      }
  
      // Ensure we have sufficient map outputs ready to shuffle before
      // scheduling reduces
      if (!scheduleReduces()) {
        return null;
      }
  
      int  target = findNewReduceTask(tts, clusterSize, numUniqueHosts);
      if (target == -1) {
        return null;
      }
  
      Task result = reduces[target].getTaskToRun(tts.getTrackerName());
      if (result != null) {
        addRunningTaskToTIP(reduces[target], result.getTaskID(), tts, true);
      }
  
      return result;
    }
docs: Return a ReduceTask, if appropriate, to run on the given tasktracker.
  We don't have cache-sensitivity for reduce tasks, as they
  work on temporary MapRed files.