Github user brkyvz commented on a diff in the pull request:
https://github.com/apache/spark/pull/9421#discussion_r43967600
--- Diff:
extras/kinesis-asl/src/main/scala/org/apache/spark/streaming/kinesis/KinesisCheckpointState.scala
---
@@ -16,39 +16,77 @@
*/
package org.apache.spark.streaming.kinesis
+import java.util.concurrent._
+
+import scala.util.control.NonFatal
+
+import
com.amazonaws.services.kinesis.clientlibrary.interfaces.IRecordProcessorCheckpointer
+import com.amazonaws.services.kinesis.clientlibrary.types.ShutdownReason
+
import org.apache.spark.Logging
import org.apache.spark.streaming.Duration
-import org.apache.spark.util.{Clock, ManualClock, SystemClock}
/**
- * This is a helper class for managing checkpoint clocks.
+ * This is a helper class for managing Kinesis checkpointing.
*
- * @param checkpointInterval
- * @param currentClock. Default to current SystemClock if none is passed
in (mocking purposes)
+ * @param receiver The receiver that keeps track of which sequence numbers
we can checkpoint
+ * @param checkpointInterval How frequently we will checkpoint to DynamoDB
+ * @param workerId Worker Id of KCL worker for logging purposes
+ * @param shardId The shard this worker was consuming data from
*/
-private[kinesis] class KinesisCheckpointState(
+private[kinesis] class KinesisCheckpointState[T](
+ receiver: KinesisReceiver[T],
checkpointInterval: Duration,
- currentClock: Clock = new SystemClock())
- extends Logging {
+ workerId: String,
+ shardId: String) extends Logging {
- /* Initialize the checkpoint clock using the given currentClock +
checkpointInterval millis */
- val checkpointClock = new ManualClock()
- checkpointClock.setTime(currentClock.getTimeMillis() +
checkpointInterval.milliseconds)
+ private var _checkpointer: Option[IRecordProcessorCheckpointer] = None
- /**
- * Check if it's time to checkpoint based on the current time and the
derived time
- * for the next checkpoint
- *
- * @return true if it's time to checkpoint
- */
- def shouldCheckpoint(): Boolean = {
- new SystemClock().getTimeMillis() > checkpointClock.getTimeMillis()
+ private val checkpointerThread = startCheckpointerThread()
+
+ /** Update the checkpointer instance to the most recent one. */
+ def setCheckpointer(checkpointer: IRecordProcessorCheckpointer): Unit = {
+ _checkpointer = Option(checkpointer)
+ }
+
+ /** Perform the checkpoint */
+ private def checkpoint(checkpointer:
Option[IRecordProcessorCheckpointer]): Unit = {
+ // if this method throws an exception, then the scheduled task will
not run again
+ try {
+ checkpointer.foreach { cp =>
--- End diff --
It should be okay
---
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 infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
