code: @Override
    public double getCopyProcessingRate(long currentTime) {
      @SuppressWarnings("deprecation")
      long bytesCopied = super.getCounters().findCounter
          (Task.Counter.REDUCE_SHUFFLE_BYTES).getCounter();
      long timeSpentCopying = 0;
      long startTime = getStartTime();
      if(getPhase() == Phase.SHUFFLE) {
        if (currentTime <= startTime) {
          LOG.error("current time is " + currentTime + ", which is <= start " +
              "time " + startTime + " in " + this.getTaskID());
        }
        timeSpentCopying =  currentTime - startTime;
      } else {
        //shuffle phase is done
        long shuffleFinishTime = getShuffleFinishTime();
        if (shuffleFinishTime <= startTime) {
          LOG.error("Shuffle finish time is " + shuffleFinishTime + 
              ", which is <= start time " + startTime + 
              " in " + this.getTaskID());
          return 0;
        }
        timeSpentCopying = shuffleFinishTime - startTime;
      }
      copyProcessingRate = bytesCopied/timeSpentCopying;
      return copyProcessingRate;
    }
docs: rate