Use standard timeout in ScalaTest eventually blocks.

Author: JoshRosen

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

@@ -373,19 +373,19 @@ class CheckpointSuite extends TestSuiteBase {
           clock.addToTime(batchDuration.milliseconds)
           if (i != 3) {
             // Since we want to shut down while the 3rd batch is processing
-            eventually(timeout(batchDuration * 5)) {
+            eventually(eventuallyTimeout) {
               assert(batchCounter.getNumCompletedBatches === i)
             }
           }
         }
         clock.addToTime(batchDuration.milliseconds)
-        eventually(timeout(batchDuration * 5)) {
+        eventually(eventuallyTimeout) {
           // Wait until all files have been recorded and all batches have started
           assert(recordedFiles(ssc) === Seq(1, 2, 3) && batchCounter.getNumStartedBatches === 3)
         }
         // Wait for a checkpoint to be written
         val fs = new Path(checkpointDir).getFileSystem(ssc.sc.hadoopConfiguration)
-        eventually(timeout(batchDuration * 5)) {
+        eventually(eventuallyTimeout) {
           assert(Checkpoint.getCheckpointFiles(checkpointDir, fs).size === 5)
         }
         ssc.stop()
@@ -424,7 +424,7 @@ class CheckpointSuite extends TestSuiteBase {
         for ((i, index) <- Seq(7, 8, 9).zipWithIndex) {
           writeFile(i, clock)
           clock.addToTime(batchDuration.milliseconds)
-          eventually(timeout(batchDuration * 5)) {
+          eventually(eventuallyTimeout) {
             assert(batchCounter.getNumCompletedBatches === index + 1)
           }
         }

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

@@ -263,7 +263,7 @@ class InputStreamsSuite extends TestSuiteBase with BeforeAndAfter {
           assert(file.setLastModified(clock.currentTime()))
           assert(file.lastModified === clock.currentTime)
           logInfo("Created file " + file)
-          eventually(timeout(batchDuration * 5)) {
+          eventually(eventuallyTimeout) {
             assert(batchCounter.getNumCompletedBatches === i)
           }
         }

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

@@ -25,7 +25,9 @@ import scala.language.implicitConversions
 import scala.reflect.ClassTag
 
 import org.scalatest.{BeforeAndAfter, FunSuite}
-import org.scalatest.time.{Span, Milliseconds => ScalaTestMilliseconds}
+import org.scalatest.time.{Span, Seconds => ScalaTestSeconds}
+import org.scalatest.concurrent.Eventually.timeout
+import org.scalatest.concurrent.PatienceConfiguration
 
 import org.apache.spark.streaming.dstream.{DStream, InputDStream, ForEachDStream}
 import org.apache.spark.streaming.scheduler.{StreamingListenerBatchStarted, StreamingListenerBatchCompleted, StreamingListener}
@@ -179,6 +181,9 @@ trait TestSuiteBase extends FunSuite with BeforeAndAfter with Logging {
     .setMaster(master)
     .setAppName(framework)
 
+  // Timeout for use in ScalaTest `eventually` blocks
+  val eventuallyTimeout: PatienceConfiguration.Timeout = timeout(Span(10, ScalaTestSeconds))
+
   // Default before function for any streaming test suite. Override this
   // if you want to add your stuff to "before" (i.e., don't call before { } )
   def beforeFunction() {
@@ -200,14 +205,6 @@ trait TestSuiteBase extends FunSuite with BeforeAndAfter with Logging {
   before(beforeFunction)
   after(afterFunction)
 
-  /**
-   * Implicit conversion which allows streaming Durations to be used with ScalaTest methods,
-   * such as `eventually`.
-   */
-  implicit def streamingDurationToScalatestSpan(duration: Duration): Span = {
-    Span(duration.milliseconds, ScalaTestMilliseconds)
-  }
-
   /**
    * Run a block of code with the given StreamingContext and automatically
    * stop the context when the block completes or when an exception is thrown.