Github user pwendell commented on a diff in the pull request:
https://github.com/apache/spark/pull/247#discussion_r11229360
--- Diff:
streaming/src/main/scala/org/apache/spark/streaming/util/RecurringTimer.scala
---
@@ -17,44 +17,80 @@
package org.apache.spark.streaming.util
+import org.apache.spark.Logging
+
private[streaming]
-class RecurringTimer(val clock: Clock, val period: Long, val callback:
(Long) => Unit) {
+class RecurringTimer(clock: Clock, period: Long, callback: (Long) => Unit,
name: String)
+ extends Logging {
- private val thread = new Thread("RecurringTimer") {
+ private val thread = new Thread("RecurringTimer - " + name) {
+ setDaemon(true)
override def run() { loop }
}
-
- private var nextTime = 0L
+ private var prevTime = -1L
+ private var nextTime = -1L
+ private var stopped = false
+
+ /**
+ * Get the earliest time when this timer can be started. The time must
be a
+ * multiple of this timer's period and more than current time.
+ */
def getStartTime(): Long = {
(math.floor(clock.currentTime.toDouble / period) + 1).toLong * period
}
+ /**
+ * Get the earliest time when this timer can be restarted, based on the
earlier start time.
--- End diff --
I found this really confusing to follow. Could this say "based on when it
started" rather than "based on the earlier start time"?
Also, can you give an example in the doc here. In particular, when is this
different than the value returned by `getStartTime`? I'm a bit confused on that
because it seems like this would always return the same result (but I'm sure
there is a difference).
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at [email protected] or file a JIRA ticket
with INFRA.
---
