code: @Override
     public double getReduceProcessingRate(long currentTime) {
       Phase phase = getPhase();
       
       if (phase != Phase.REDUCE) {
         return 0;
       }
       
       @SuppressWarnings("deprecation")
      long bytesProcessed = super.getCounters().findCounter
           (Task.Counter.REDUCE_INPUT_BYTES).getCounter();
       long timeSpentInReduce = 0;
       long sortFinishTime = getSortFinishTime();
       if (sortFinishTime >= currentTime) {
         LOG.error("Sort finish time is " + sortFinishTime + 
             " which is >= current time " + currentTime + 
             " in " + this.getTaskID());
         return 0;
       }
       timeSpentInReduce = currentTime - sortFinishTime;
       
       reduceProcessingRate = bytesProcessed/timeSpentInReduce;   
       return reduceProcessingRate;
     }
docs: as the processing rate