setting inputMetrics in HadoopRDD#compute()

[Ted Yu]

Hi,
Starting at line 203:
      try {
        /* bytesRead may not exactly equal the bytes read by a task: split
boundaries aren't
         * always at record boundaries, so tasks may need to read into
other splits to complete
         * a record. */
        inputMetrics.bytesRead = split.inputSplit.value.getLength()
      } catch {
        case e: java.io.IOException =>
          logWarning("Unable to get input size to set InputMetrics for
task", e)
      }
      context.taskMetrics.inputMetrics = Some(inputMetrics)

If there is IOException, context.taskMetrics.inputMetrics is set by
wrapping inputMetrics - as if there wasn't any error.

I wonder if the above code should distinguish the error condition.

Cheers