Remove isStopped and use PrivateMethod to access it

Author: zsxwing

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

@@ -72,8 +72,10 @@ class RecurringTimer(clock: Clock, period: Long, callback: (Long) => Unit, name:
 
   /**
    * Stop the timer, and return the last time the callback was made.
-   * interruptTimer = true will interrupt the callback
+   * - interruptTimer = true will interrupt the callback
    * if it is in progress (not guaranteed to give correct time in this case).
+   * - interruptTimer = false guarantees that there will be at least one callback after `stop` has
+   * been called.
    */
   def stop(interruptTimer: Boolean): Long = synchronized {
     if (!stopped) {
@@ -108,11 +110,6 @@ class RecurringTimer(clock: Clock, period: Long, callback: (Long) => Unit, name:
       case e: InterruptedException =>
     }
   }
-
-  /*
-   * Return whether `stop` is called.
-   */
-  def isStopped: Boolean = stopped
 }
 
 private[streaming]

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

@@ -20,12 +20,13 @@ package org.apache.spark.streaming.util
 import scala.collection.mutable
 import scala.concurrent.duration._
 
+import org.scalatest.PrivateMethodTester
 import org.scalatest.concurrent.Eventually._
 
 import org.apache.spark.SparkFunSuite
 import org.apache.spark.util.ManualClock
 
-class RecurringTimerSuite extends SparkFunSuite {
+class RecurringTimerSuite extends SparkFunSuite with PrivateMethodTester {
 
   test("basic") {
     val clock = new ManualClock()
@@ -60,21 +61,22 @@ class RecurringTimerSuite extends SparkFunSuite {
     }
     @volatile var lastTime = -1L
     // Now RecurringTimer is waiting for the next interval
-    val t = new Thread {
+    val thread = new Thread {
       override def run(): Unit = {
         lastTime = timer.stop(interruptTimer = false)
       }
     }
-    t.start()
+    thread.start()
+    val stopped = PrivateMethod[RecurringTimer]('stopped)
     // Make sure the `stopped` field has been changed
     eventually(timeout(10.seconds), interval(10.millis)) {
-      assert(timer.isStopped)
+      assert(timer.invokePrivate(stopped()) === true)
     }
     clock.advance(200)
     // When RecurringTimer is awake from clock.waitTillTime, it will call `callback` once.
     // Then it will find `stopped` is true and exit the loop, but it should call `callback` again
     // before exiting its internal thread.
-    t.join()
+    thread.join()
     assert(results === Seq(0L, 100L, 200L))
     assert(lastTime === 200L)
   }