Update to keep postfix time operations

Author: holdenk

core/src/main/scala/org/apache/spark/deploy/FaultToleranceTest.scala

@@ -26,6 +26,7 @@ import scala.collection.mutable.ListBuffer
 import scala.concurrent.{Future, Promise}
 import scala.concurrent.ExecutionContext.Implicits.global
 import scala.concurrent.duration._
+import scala.language.postfixOps
 import scala.sys.process._
 
 import org.json4s._
@@ -112,7 +113,7 @@ private object FaultToleranceTest extends App with Logging {
     assertValidClusterState()
 
     killLeader()
-    delay(30.seconds)
+    delay(30 seconds)
     assertValidClusterState()
     createClient()
     assertValidClusterState()
@@ -126,12 +127,12 @@ private object FaultToleranceTest extends App with Logging {
 
     killLeader()
     addMasters(1)
-    delay(30.seconds)
+    delay(30 seconds)
     assertValidClusterState()
 
     killLeader()
     addMasters(1)
-    delay(30.seconds)
+    delay(30 seconds)
     assertValidClusterState()
   }
 
@@ -156,7 +157,7 @@ private object FaultToleranceTest extends App with Logging {
     killLeader()
     workers.foreach(_.kill())
     workers.clear()
-    delay(30.seconds)
+    delay(30 seconds)
     addWorkers(2)
     assertValidClusterState()
   }
@@ -174,7 +175,7 @@ private object FaultToleranceTest extends App with Logging {
 
     (1 to 3).foreach { _ =>
       killLeader()
-      delay(30.seconds)
+      delay(30 seconds)
       assertValidClusterState()
       assertTrue(getLeader == masters.head)
       addMasters(1)
@@ -264,7 +265,7 @@ private object FaultToleranceTest extends App with Logging {
     }
 
     // Avoid waiting indefinitely (e.g., we could register but get no executors).
-    assertTrue(ThreadUtils.awaitResult(f, 120.seconds))
+    assertTrue(ThreadUtils.awaitResult(f, 120 seconds))
   }
 
   /**
@@ -317,7 +318,7 @@ private object FaultToleranceTest extends App with Logging {
     }
 
     try {
-      assertTrue(ThreadUtils.awaitResult(f, 120.seconds))
+      assertTrue(ThreadUtils.awaitResult(f, 120 seconds))
     } catch {
       case e: TimeoutException =>
         logError("Master states: " + masters.map(_.state))
@@ -421,7 +422,7 @@ private object SparkDocker {
     }
 
     dockerCmd.run(ProcessLogger(findIpAndLog _))
-    val ip = ThreadUtils.awaitResult(ipPromise.future, 30.seconds)
+    val ip = ThreadUtils.awaitResult(ipPromise.future, 30 seconds)
     val dockerId = Docker.getLastProcessId
     (ip, dockerId, outFile)
   }

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

@@ -27,6 +27,7 @@ import scala.collection.Map
 import scala.collection.mutable.{HashMap, HashSet, Stack}
 import scala.concurrent.duration._
 import scala.language.existentials
+import scala.language.postfixOps
 import scala.util.control.NonFatal
 
 import org.apache.commons.lang3.SerializationUtils
@@ -232,7 +233,7 @@ class DAGScheduler(
       blockManagerId: BlockManagerId): Boolean = {
     listenerBus.post(SparkListenerExecutorMetricsUpdate(execId, accumUpdates))
     blockManagerMaster.driverEndpoint.askWithRetry[Boolean](
-      BlockManagerHeartbeat(blockManagerId), new RpcTimeout(600.seconds, "BlockManagerHeartbeat"))
+      BlockManagerHeartbeat(blockManagerId), new RpcTimeout(600 seconds, "BlockManagerHeartbeat"))
   }
 
   /**

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

@@ -21,6 +21,7 @@ import java.util.concurrent.{Executors, TimeUnit}
 
 import scala.collection.JavaConverters._
 import scala.concurrent.duration._
+import scala.language.postfixOps
 import scala.util.{Random, Try}
 
 import com.esotericsoftware.kryo.Kryo
@@ -261,10 +262,10 @@ class SparkConfSuite extends SparkFunSuite with LocalSparkContext with ResetSyst
     assert(RpcUtils.retryWaitMs(conf) === 2L)
 
     conf.set("spark.akka.askTimeout", "3")
-    assert(RpcUtils.askRpcTimeout(conf).duration === (3.seconds))
+    assert(RpcUtils.askRpcTimeout(conf).duration === (3 seconds))
 
     conf.set("spark.akka.lookupTimeout", "4")
-    assert(RpcUtils.lookupRpcTimeout(conf).duration === (4.seconds))
+    assert(RpcUtils.lookupRpcTimeout(conf).duration === (4 seconds))
   }
 
   test("SPARK-13727") {

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

@@ -19,6 +19,7 @@ package org.apache.spark
 
 import scala.concurrent.duration._
 import scala.language.implicitConversions
+import scala.language.postfixOps
 
 import org.scalatest.Matchers
 import org.scalatest.concurrent.Eventually._
@@ -30,25 +31,25 @@ class StatusTrackerSuite extends SparkFunSuite with Matchers with LocalSparkCont
   test("basic status API usage") {
     sc = new SparkContext("local", "test", new SparkConf(false))
     val jobFuture = sc.parallelize(1 to 10000, 2).map(identity).groupBy(identity).collectAsync()
-    val jobId: Int = eventually(timeout(10.seconds)) {
+    val jobId: Int = eventually(timeout(10 seconds)) {
       val jobIds = jobFuture.jobIds
       jobIds.size should be(1)
       jobIds.head
     }
-    val jobInfo = eventually(timeout(10.seconds)) {
+    val jobInfo = eventually(timeout(10 seconds)) {
       sc.statusTracker.getJobInfo(jobId).get
     }
     jobInfo.status() should not be FAILED
     val stageIds = jobInfo.stageIds()
     stageIds.size should be(2)
 
-    val firstStageInfo = eventually(timeout(10.seconds)) {
+    val firstStageInfo = eventually(timeout(10 seconds)) {
       sc.statusTracker.getStageInfo(stageIds(0)).get
     }
     firstStageInfo.stageId() should be(stageIds(0))
     firstStageInfo.currentAttemptId() should be(0)
     firstStageInfo.numTasks() should be(2)
-    eventually(timeout(10.seconds)) {
+    eventually(timeout(10 seconds)) {
       val updatedFirstStageInfo = sc.statusTracker.getStageInfo(stageIds(0)).get
       updatedFirstStageInfo.numCompletedTasks() should be(2)
       updatedFirstStageInfo.numActiveTasks() should be(0)
@@ -60,27 +61,27 @@ class StatusTrackerSuite extends SparkFunSuite with Matchers with LocalSparkCont
     sc = new SparkContext("local", "test", new SparkConf(false))
     // Passing `null` should return jobs that were not run in a job group:
     val defaultJobGroupFuture = sc.parallelize(1 to 1000).countAsync()
-    val defaultJobGroupJobId = eventually(timeout(10.seconds)) {
+    val defaultJobGroupJobId = eventually(timeout(10 seconds)) {
       defaultJobGroupFuture.jobIds.head
     }
-    eventually(timeout(10.seconds)) {
+    eventually(timeout(10 seconds)) {
       sc.statusTracker.getJobIdsForGroup(null).toSet should be (Set(defaultJobGroupJobId))
     }
     // Test jobs submitted in job groups:
     sc.setJobGroup("my-job-group", "description")
     sc.statusTracker.getJobIdsForGroup("my-job-group") should be (Seq.empty)
     val firstJobFuture = sc.parallelize(1 to 1000).countAsync()
-    val firstJobId = eventually(timeout(10.seconds)) {
+    val firstJobId = eventually(timeout(10 seconds)) {
       firstJobFuture.jobIds.head
     }
-    eventually(timeout(10.seconds)) {
+    eventually(timeout(10 seconds)) {
       sc.statusTracker.getJobIdsForGroup("my-job-group") should be (Seq(firstJobId))
     }
     val secondJobFuture = sc.parallelize(1 to 1000).countAsync()
-    val secondJobId = eventually(timeout(10.seconds)) {
+    val secondJobId = eventually(timeout(10 seconds)) {
       secondJobFuture.jobIds.head
     }
-    eventually(timeout(10.seconds)) {
+    eventually(timeout(10 seconds)) {
       sc.statusTracker.getJobIdsForGroup("my-job-group").toSet should be (
         Set(firstJobId, secondJobId))
     }
@@ -91,10 +92,10 @@ class StatusTrackerSuite extends SparkFunSuite with Matchers with LocalSparkCont
     sc.setJobGroup("my-job-group2", "description")
     sc.statusTracker.getJobIdsForGroup("my-job-group2") shouldBe empty
     val firstJobFuture = sc.parallelize(1 to 1000, 1).takeAsync(1)
-    val firstJobId = eventually(timeout(10.seconds)) {
+    val firstJobId = eventually(timeout(10 seconds)) {
       firstJobFuture.jobIds.head
     }
-    eventually(timeout(10.seconds)) {
+    eventually(timeout(10 seconds)) {
       sc.statusTracker.getJobIdsForGroup("my-job-group2") should be (Seq(firstJobId))
     }
   }
@@ -104,10 +105,10 @@ class StatusTrackerSuite extends SparkFunSuite with Matchers with LocalSparkCont
     sc.setJobGroup("my-job-group2", "description")
     sc.statusTracker.getJobIdsForGroup("my-job-group2") shouldBe empty
     val firstJobFuture = sc.parallelize(1 to 1000, 2).takeAsync(999)
-    val firstJobId = eventually(timeout(10.seconds)) {
+    val firstJobId = eventually(timeout(10 seconds)) {
       firstJobFuture.jobIds.head
     }
-    eventually(timeout(10.seconds)) {
+    eventually(timeout(10 seconds)) {
       sc.statusTracker.getJobIdsForGroup("my-job-group2") should have size 2
     }
   }

core/src/test/scala/org/apache/spark/deploy/history/FsHistoryProviderSuite.scala

@@ -25,6 +25,7 @@ import java.util.concurrent.TimeUnit
 import java.util.zip.{ZipInputStream, ZipOutputStream}
 
 import scala.concurrent.duration._
+import scala.language.postfixOps
 
 import com.google.common.io.{ByteStreams, Files}
 import org.apache.hadoop.hdfs.DistributedFileSystem
@@ -367,7 +368,7 @@ class FsHistoryProviderSuite extends SparkFunSuite with BeforeAndAfter with Matc
       provider.inSafeMode = false
       clock.setTime(10000)
 
-      eventually(timeout(1.second), interval(10.millis)) {
+      eventually(timeout(1 second), interval(10 millis)) {
         provider.getConfig().keys should not contain ("HDFS State")
       }
     } finally {
@@ -385,7 +386,7 @@ class FsHistoryProviderSuite extends SparkFunSuite with BeforeAndAfter with Matc
       provider.inSafeMode = false
       clock.setTime(10000)
 
-      eventually(timeout(1.second), interval(10.millis)) {
+      eventually(timeout(1 second), interval(10 millis)) {
         verify(errorHandler).uncaughtException(any(), any())
       }
     } finally {

core/src/test/scala/org/apache/spark/deploy/history/HistoryServerSuite.scala

@@ -23,6 +23,7 @@ import java.util.zip.ZipInputStream
 import javax.servlet.http.{HttpServletRequest, HttpServletResponse}
 
 import scala.concurrent.duration._
+import scala.language.postfixOps
 
 import com.codahale.metrics.Counter
 import com.google.common.io.{ByteStreams, Files}
@@ -348,8 +349,8 @@ class HistoryServerSuite extends SparkFunSuite with BeforeAndAfter with Matchers
     // start initial job
     val d = sc.parallelize(1 to 10)
     d.count()
-    val stdInterval = interval(100.milliseconds)
-    val appId = eventually(timeout(20.seconds), stdInterval) {
+    val stdInterval = interval(100 milliseconds)
+    val appId = eventually(timeout(20 seconds), stdInterval) {
       val json = getContentAndCode("applications", port)._2.get
       val apps = parse(json).asInstanceOf[JArray].arr
       apps should have size 1
@@ -429,7 +430,7 @@ class HistoryServerSuite extends SparkFunSuite with BeforeAndAfter with Matchers
     d2.count()
     dumpLogDir("After second job")
 
-    val stdTimeout = timeout(10.seconds)
+    val stdTimeout = timeout(10 seconds)
     logDebug("waiting for UI to update")
     eventually(stdTimeout, stdInterval) {
       assert(2 === getNumJobs(""),

core/src/test/scala/org/apache/spark/deploy/master/MasterSuite.scala

@@ -23,6 +23,7 @@ import java.util.concurrent.ConcurrentLinkedQueue
 import scala.collection.JavaConverters._
 import scala.concurrent.duration._
 import scala.io.Source
+import scala.language.postfixOps
 
 import org.json4s._
 import org.json4s.jackson.JsonMethods._
@@ -139,7 +140,7 @@ class MasterSuite extends SparkFunSuite
     val localCluster = new LocalSparkCluster(2, 2, 512, conf)
     localCluster.start()
     try {
-      eventually(timeout(5.seconds), interval(100.milliseconds)) {
+      eventually(timeout(5 seconds), interval(100 milliseconds)) {
         val json = Source.fromURL(s"http://localhost:${localCluster.masterWebUIPort}/json")
           .getLines().mkString("\n")
         val JArray(workers) = (parse(json) \ "workers")

core/src/test/scala/org/apache/spark/launcher/LauncherBackendSuite.scala

@@ -20,6 +20,7 @@ package org.apache.spark.launcher
 import java.util.concurrent.TimeUnit
 
 import scala.concurrent.duration._
+import scala.language.postfixOps
 
 import org.scalatest.Matchers
 import org.scalatest.concurrent.Eventually._
@@ -52,13 +53,13 @@ class LauncherBackendSuite extends SparkFunSuite with Matchers {
       .startApplication()
 
     try {
-      eventually(timeout(30.seconds), interval(100.millis)) {
+      eventually(timeout(30 seconds), interval(100 millis)) {
         handle.getAppId() should not be (null)
       }
 
       handle.stop()
 
-      eventually(timeout(30.seconds), interval(100.millis)) {
+      eventually(timeout(30 seconds), interval(100 millis)) {
         handle.getState() should be (SparkAppHandle.State.KILLED)
       }
     } finally {