Github user zsxwing commented on a diff in the pull request:
https://github.com/apache/spark/pull/9421#discussion_r43817494
--- 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
--- End diff --
This should be `volatile`
---
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
