Revert "Completed refactoring to make vars in TaskMetrics class private"

This reverts commit 5525c20.

Author: Ilya Ganelin

core/src/main/scala/org/apache/spark/executor/Executor.scala

@@ -194,10 +194,10 @@ private[spark] class Executor(
         val afterSerialization = System.currentTimeMillis()
 
         for (m <- task.metrics) {
-          m.incExecutorDeserializeTime(taskStart - deserializeStartTime)
-          m.incExecutorRunTime(taskFinish - taskStart)
-          m.incJvmGCTime(gcTime - startGCTime)
-          m.incResultSerializationTime(afterSerialization - beforeSerialization)
+          m.executorDeserializeTime = taskStart - deserializeStartTime
+          m.executorRunTime = taskFinish - taskStart
+          m.jvmGCTime = gcTime - startGCTime
+          m.resultSerializationTime = afterSerialization - beforeSerialization
         }
 
         val accumUpdates = Accumulators.values
@@ -248,8 +248,8 @@ private[spark] class Executor(
           val serviceTime = System.currentTimeMillis() - taskStart
           val metrics = attemptedTask.flatMap(t => t.metrics)
           for (m <- metrics) {
-            m.incExecutorRunTime(serviceTime)
-            m.incJvmGCTime(gcTime - startGCTime)
+            m.executorRunTime = serviceTime
+            m.jvmGCTime = gcTime - startGCTime
           }
           val reason = new ExceptionFailure(t, metrics)
           execBackend.statusUpdate(taskId, TaskState.FAILED, ser.serialize(reason))

core/src/main/scala/org/apache/spark/executor/TaskMetrics.scala

@@ -39,67 +39,42 @@ class TaskMetrics extends Serializable {
   /**
    * Host's name the task runs on
    */
-  private var _hostname: String = _
-  def hostname = _hostname
-  def setHostname(value : String) = _hostname = value
-  
+  var hostname: String = _
+
   /**
    * Time taken on the executor to deserialize this task
    */
-  private var _executorDeserializeTime: Long = _
-  def executorDeserializeTime = _executorDeserializeTime
-  def incExecutorDeserializeTime(value: Long) = _executorDeserializeTime += value
-  def decExecutorDeserializeTime(value: Long) = _executorDeserializeTime -= value
-  
-  
+  var executorDeserializeTime: Long = _
+
   /**
    * Time the executor spends actually running the task (including fetching shuffle data)
    */
-  private var _executorRunTime: Long = _
-  def executorRunTime = _executorRunTime
-  def incExecutorRunTime(value: Long) = _executorRunTime += value
-  def decExecutorRunTime(value: Long) = _executorRunTime -= value
-  
+  var executorRunTime: Long = _
+
   /**
    * The number of bytes this task transmitted back to the driver as the TaskResult
    */
-  private var _resultSize: Long = _
-  def resultSize = _resultSize
-  def incResultSize(value: Long) = _resultSize += value
-  def decResultSize(value: Long) = _resultSize -= value
-
+  var resultSize: Long = _
 
   /**
    * Amount of time the JVM spent in garbage collection while executing this task
    */
-  private var _jvmGCTime: Long = _
-  def jvmGCTime = _jvmGCTime
-  def incJvmGCTime(value: Long) = _jvmGCTime += value
-  def decJvmGCTime(value: Long) = _jvmGCTime -= value
+  var jvmGCTime: Long = _
 
   /**
    * Amount of time spent serializing the task result
    */
-  private var _resultSerializationTime: Long = _
-  def resultSerializationTime = _resultSerializationTime
-  def incResultSerializationTime(value: Long) = _resultSerializationTime += value
-  def decResultSerializationTime(value: Long) = _resultSerializationTime -= value
+  var resultSerializationTime: Long = _
 
   /**
    * The number of in-memory bytes spilled by this task
    */
-  private var _memoryBytesSpilled: Long = _
-  def memoryBytesSpilled = _memoryBytesSpilled
-  def incMemoryBytesSpilled(value: Long) = _memoryBytesSpilled += value
-  def decMemoryBytesSpilled(value: Long) = _memoryBytesSpilled -= value
+  var memoryBytesSpilled: Long = _
 
   /**
    * The number of on-disk bytes spilled by this task
    */
-  private var _diskBytesSpilled: Long = _
-  def diskBytesSpilled = _diskBytesSpilled
-  def incDiskBytesSpilled(value: Long) = _diskBytesSpilled += value
-  def decDiskBytesSpilled(value: Long) = _diskBytesSpilled -= value
+  var diskBytesSpilled: Long = _
 
   /**
    * If this task reads from a HadoopRDD or from persisted data, metrics on how much data was read

core/src/main/scala/org/apache/spark/scheduler/Task.scala

@@ -47,7 +47,7 @@ private[spark] abstract class Task[T](val stageId: Int, var partitionId: Int) ex
   final def run(attemptId: Long): T = {
     context = new TaskContextImpl(stageId, partitionId, attemptId, runningLocally = false)
     TaskContextHelper.setTaskContext(context)
-    context.taskMetrics.setHostname(Utils.localHostName())
+    context.taskMetrics.hostname = Utils.localHostName()
     taskThread = Thread.currentThread()
     if (_killed) {
       kill(interruptThread = false)

core/src/main/scala/org/apache/spark/scheduler/TaskResultGetter.scala

@@ -77,7 +77,7 @@ private[spark] class TaskResultGetter(sparkEnv: SparkEnv, scheduler: TaskSchedul
               (deserializedResult, size)
           }
 
-          result.metrics.incResultSize(size)
+          result.metrics.resultSize = size
           scheduler.handleSuccessfulTask(taskSetManager, tid, result)
         } catch {
           case cnf: ClassNotFoundException =>

core/src/main/scala/org/apache/spark/util/JsonProtocol.scala

@@ -592,14 +592,14 @@ private[spark] object JsonProtocol {
       return TaskMetrics.empty
     }
     val metrics = new TaskMetrics
-    metrics.setHostname((json \ "Host Name").extract[String])
-    metrics.incExecutorDeserializeTime((json \ "Executor Deserialize Time").extract[Long])
-    metrics.incExecutorRunTime((json \ "Executor Run Time").extract[Long])
-    metrics.incResultSize((json \ "Result Size").extract[Long])
-    metrics.incJvmGCTime((json \ "JVM GC Time").extract[Long])
-    metrics.incResultSerializationTime((json \ "Result Serialization Time").extract[Long])
-    metrics.incMemoryBytesSpilled((json \ "Memory Bytes Spilled").extract[Long])
-    metrics.incDiskBytesSpilled((json \ "Disk Bytes Spilled").extract[Long])
+    metrics.hostname = (json \ "Host Name").extract[String]
+    metrics.executorDeserializeTime = (json \ "Executor Deserialize Time").extract[Long]
+    metrics.executorRunTime = (json \ "Executor Run Time").extract[Long]
+    metrics.resultSize = (json \ "Result Size").extract[Long]
+    metrics.jvmGCTime = (json \ "JVM GC Time").extract[Long]
+    metrics.resultSerializationTime = (json \ "Result Serialization Time").extract[Long]
+    metrics.memoryBytesSpilled = (json \ "Memory Bytes Spilled").extract[Long]
+    metrics.diskBytesSpilled = (json \ "Disk Bytes Spilled").extract[Long]
     metrics.setShuffleReadMetrics(
       Utils.jsonOption(json \ "Shuffle Read Metrics").map(shuffleReadMetricsFromJson))
     metrics.shuffleWriteMetrics =

core/src/test/scala/org/apache/spark/ui/jobs/JobProgressListenerSuite.scala

@@ -227,9 +227,9 @@ class JobProgressListenerSuite extends FunSuite with LocalSparkContext with Matc
       shuffleReadMetrics.incRemoteBytesRead(base + 1)
       shuffleReadMetrics.incRemoteBlocksFetched(base + 2)
       shuffleWriteMetrics.incShuffleBytesWritten(base + 3)
-      taskMetrics.incExecutorRunTime(base + 4)
-      taskMetrics.incDiskBytesSpilled(base + 5)
-      taskMetrics.incMemoryBytesSpilled(base + 6)
+      taskMetrics.executorRunTime = base + 4
+      taskMetrics.diskBytesSpilled = base + 5
+      taskMetrics.memoryBytesSpilled = base + 6
       val inputMetrics = new InputMetrics(DataReadMethod.Hadoop)
       taskMetrics.inputMetrics = Some(inputMetrics)
       inputMetrics.incBytesRead(base + 7)

core/src/test/scala/org/apache/spark/util/JsonProtocolSuite.scala

@@ -599,13 +599,13 @@ class JsonProtocolSuite extends FunSuite {
       hasHadoopInput: Boolean,
       hasOutput: Boolean) = {
     val t = new TaskMetrics
-    t.setHostname("localhost")
-    t.incExecutorDeserializeTime(a)
-    t.incExecutorRunTime(b)
-    t.incResultSize(c)
-    t.incJvmGCTime(d)
-    t.incResultSerializationTime(a + b)
-    t.incMemoryBytesSpilled(a + c)
+    t.hostname = "localhost"
+    t.executorDeserializeTime = a
+    t.executorRunTime = b
+    t.resultSize = c
+    t.jvmGCTime = d
+    t.resultSerializationTime = a + b
+    t.memoryBytesSpilled = a + c
 
     if (hasHadoopInput) {
       val inputMetrics = new InputMetrics(DataReadMethod.Hadoop)