Reduce the net change in this patch by reversing some unnecessary syntax changes along the way

Author: srowen

core/src/main/scala/org/apache/spark/ExecutorAllocationManager.scala

@@ -227,7 +227,7 @@ private[spark] class ExecutorAllocationManager(
    * This is factored out into its own method for testing.
    */
   private def schedule(): Unit = synchronized {
-    val now = clock.getTimeMillis()
+    val now = clock.getTimeMillis
 
     addOrCancelExecutorRequests(now)
 
@@ -410,7 +410,7 @@ private[spark] class ExecutorAllocationManager(
     if (addTime == NOT_SET) {
       logDebug(s"Starting timer to add executors because pending tasks " +
         s"are building up (to expire in $schedulerBacklogTimeout seconds)")
-      addTime = clock.getTimeMillis() + schedulerBacklogTimeout * 1000
+      addTime = clock.getTimeMillis + schedulerBacklogTimeout * 1000
     }
   }
 
@@ -434,7 +434,7 @@ private[spark] class ExecutorAllocationManager(
       if (!removeTimes.contains(executorId) && !executorsPendingToRemove.contains(executorId)) {
         logDebug(s"Starting idle timer for $executorId because there are no more tasks " +
           s"scheduled to run on the executor (to expire in $executorIdleTimeout seconds)")
-        removeTimes(executorId) = clock.getTimeMillis() + executorIdleTimeout * 1000
+        removeTimes(executorId) = clock.getTimeMillis + executorIdleTimeout * 1000
       }
     } else {
       logWarning(s"Attempted to mark unknown executor $executorId idle")

core/src/test/scala/org/apache/spark/ExecutorAllocationManagerSuite.scala

@@ -330,7 +330,7 @@ class ExecutorAllocationManagerSuite extends FunSuite with LocalSparkContext {
     assert(addTime(manager) === NOT_SET)
     onSchedulerBacklogged(manager)
     val firstAddTime = addTime(manager)
-    assert(firstAddTime === clock.getTimeMillis() + schedulerBacklogTimeout * 1000)
+    assert(firstAddTime === clock.getTimeMillis + schedulerBacklogTimeout * 1000)
     clock.advance(100L)
     onSchedulerBacklogged(manager)
     assert(addTime(manager) === firstAddTime) // timer is already started
@@ -344,7 +344,7 @@ class ExecutorAllocationManagerSuite extends FunSuite with LocalSparkContext {
     assert(addTime(manager) === NOT_SET)
     onSchedulerBacklogged(manager)
     val secondAddTime = addTime(manager)
-    assert(secondAddTime === clock.getTimeMillis() + schedulerBacklogTimeout * 1000)
+    assert(secondAddTime === clock.getTimeMillis + schedulerBacklogTimeout * 1000)
     clock.advance(100L)
     onSchedulerBacklogged(manager)
     assert(addTime(manager) === secondAddTime) // timer is already started
@@ -366,7 +366,7 @@ class ExecutorAllocationManagerSuite extends FunSuite with LocalSparkContext {
     assert(removeTimes(manager).size === 1)
     assert(removeTimes(manager).contains("1"))
     val firstRemoveTime = removeTimes(manager)("1")
-    assert(firstRemoveTime === clock.getTimeMillis() + executorIdleTimeout * 1000)
+    assert(firstRemoveTime === clock.getTimeMillis + executorIdleTimeout * 1000)
     clock.advance(100L)
     onExecutorIdle(manager, "1")
     assert(removeTimes(manager)("1") === firstRemoveTime) // timer is already started
@@ -376,11 +376,11 @@ class ExecutorAllocationManagerSuite extends FunSuite with LocalSparkContext {
     clock.advance(300L)
     onExecutorIdle(manager, "2")
     assert(removeTimes(manager)("2") !== firstRemoveTime) // different executor
-    assert(removeTimes(manager)("2") === clock.getTimeMillis() + executorIdleTimeout * 1000)
+    assert(removeTimes(manager)("2") === clock.getTimeMillis + executorIdleTimeout * 1000)
     clock.advance(400L)
     onExecutorIdle(manager, "3")
     assert(removeTimes(manager)("3") !== firstRemoveTime)
-    assert(removeTimes(manager)("3") === clock.getTimeMillis() + executorIdleTimeout * 1000)
+    assert(removeTimes(manager)("3") === clock.getTimeMillis + executorIdleTimeout * 1000)
     assert(removeTimes(manager).size === 3)
     assert(removeTimes(manager).contains("2"))
     assert(removeTimes(manager).contains("3"))
@@ -393,7 +393,7 @@ class ExecutorAllocationManagerSuite extends FunSuite with LocalSparkContext {
     assert(removeTimes(manager).size === 3)
     assert(removeTimes(manager).contains("1"))
     val secondRemoveTime = removeTimes(manager)("1")
-    assert(secondRemoveTime === clock.getTimeMillis() + executorIdleTimeout * 1000)
+    assert(secondRemoveTime === clock.getTimeMillis + executorIdleTimeout * 1000)
     assert(removeTimes(manager)("1") === secondRemoveTime) // timer is already started
     assert(removeTimes(manager)("1") !== firstRemoveTime)
     assert(firstRemoveTime !== secondRemoveTime)

streaming/src/main/scala/org/apache/spark/streaming/receiver/ReceivedBlockHandler.scala

@@ -111,7 +111,7 @@ private[streaming] class WriteAheadLogBasedBlockHandler(
     conf: SparkConf,
     hadoopConf: Configuration,
     checkpointDir: String,
-    clock: Clock = new SystemClock()
+    clock: Clock = new SystemClock
   ) extends ReceivedBlockHandler with Logging {
 
   private val blockStoreTimeout = conf.getInt(

streaming/src/main/scala/org/apache/spark/streaming/util/WriteAheadLogManager.scala

@@ -51,7 +51,7 @@ private[streaming] class WriteAheadLogManager(
     rollingIntervalSecs: Int = 60,
     maxFailures: Int = 3,
     callerName: String = "",
-    clock: Clock = new SystemClock()
+    clock: Clock = new SystemClock
   ) extends Logging {
 
   private val pastLogs = new ArrayBuffer[LogInfo]

streaming/src/test/scala/org/apache/spark/streaming/ReceivedBlockHandlerSuite.scala

@@ -53,7 +53,7 @@ class ReceivedBlockHandlerSuite extends FunSuite with BeforeAndAfter with Matche
   val mapOutputTracker = new MapOutputTrackerMaster(conf)
   val shuffleManager = new HashShuffleManager(conf)
   val serializer = new KryoSerializer(conf)
-  val manualClock = new ManualClock()
+  val manualClock = new ManualClock
   val blockManagerSize = 10000000
 
   var actorSystem: ActorSystem = null

streaming/src/test/scala/org/apache/spark/streaming/ReceivedBlockTrackerSuite.scala

@@ -94,7 +94,7 @@ class ReceivedBlockTrackerSuite
   }
 
   test("block addition, block to batch allocation and cleanup with write ahead log") {
-    val manualClock = new ManualClock()
+    val manualClock = new ManualClock
     // Set the time increment level to twice the rotation interval so that every increment creates
     // a new log file
 
@@ -211,7 +211,7 @@ class ReceivedBlockTrackerSuite
    */
   def createTracker(
       setCheckpointDir: Boolean = true,
-      clock: Clock = new SystemClock()): ReceivedBlockTracker = {
+      clock: Clock = new SystemClock): ReceivedBlockTracker = {
     val cpDirOption = if (setCheckpointDir) Some(checkpointDirectory.toString) else None
     val tracker = new ReceivedBlockTracker(conf, hadoopConf, Seq(streamId), clock, cpDirOption)
     allReceivedBlockTrackers += tracker

streaming/src/test/scala/org/apache/spark/streaming/util/WriteAheadLogSuite.scala

@@ -191,7 +191,7 @@ class WriteAheadLogSuite extends FunSuite with BeforeAndAfter {
 
   private def logCleanUpTest(waitForCompletion: Boolean): Unit = {
     // Write data with manager, recover with new manager and verify
-    val manualClock = new ManualClock()
+    val manualClock = new ManualClock
     val dataToWrite = generateRandomData()
     manager = writeDataUsingManager(testDir, dataToWrite, manualClock, stopManager = false)
     val logFiles = getLogFilesInDirectory(testDir)
@@ -210,17 +210,17 @@ class WriteAheadLogSuite extends FunSuite with BeforeAndAfter {
 
   test("WriteAheadLogManager - handling file errors while reading rotating logs") {
     // Generate a set of log files
-    val fakeClock = new ManualClock()
+    val manualClock = new ManualClock
     val dataToWrite1 = generateRandomData()
-    writeDataUsingManager(testDir, dataToWrite1, fakeClock)
+    writeDataUsingManager(testDir, dataToWrite1, manualClock)
     val logFiles1 = getLogFilesInDirectory(testDir)
     assert(logFiles1.size > 1)
 
 
     // Recover old files and generate a second set of log files
     val dataToWrite2 = generateRandomData()
-    fakeClock.advance(100000)
-    writeDataUsingManager(testDir, dataToWrite2, fakeClock)
+    manualClock.advance(100000)
+    writeDataUsingManager(testDir, dataToWrite2, manualClock)
     val logFiles2 = getLogFilesInDirectory(testDir)
     assert(logFiles2.size > logFiles1.size)
 
@@ -276,15 +276,15 @@ object WriteAheadLogSuite {
   def writeDataUsingManager(
       logDirectory: String,
       data: Seq[String],
-      fakeClock: ManualClock = new ManualClock(),
+      manualClock: ManualClock = new ManualClock,
       stopManager: Boolean = true
     ): WriteAheadLogManager = {
-    if (fakeClock.getTimeMillis() < 100000) fakeClock.setTime(10000)
+    if (manualClock.getTimeMillis() < 100000) manualClock.setTime(10000)
     val manager = new WriteAheadLogManager(logDirectory, hadoopConf,
-      rollingIntervalSecs = 1, callerName = "WriteAheadLogSuite", clock = fakeClock)
+      rollingIntervalSecs = 1, callerName = "WriteAheadLogSuite", clock = manualClock)
     // Ensure that 500 does not get sorted after 2000, so put a high base value.
     data.foreach { item =>
-      fakeClock.advance(500)
+      manualClock.advance(500)
       manager.writeToLog(item)
     }
     if (stopManager) manager.stop()