Kinesis source: make idleTimeBetweenReadsInMillis configurable for polling

Author: benjben

modules/it/src/test/scala/com/snowplowanalytics/snowplow/sources/kinesis/Utils.scala

@@ -88,7 +88,7 @@ object Utils {
     streamName,
     UUID.randomUUID.toString,
     KinesisSourceConfig.InitialPosition.TrimHorizon,
-    KinesisSourceConfig.Retrieval.Polling(1),
+    KinesisSourceConfig.Retrieval.Polling(1, 1500.millis),
     Some(endpoint),
     Some(endpoint),
     Some(endpoint),

modules/kinesis/src/main/resources/reference.conf

@@ -8,6 +8,7 @@ snowplow.defaults: {
       retrievalMode: {
         type: "Polling"
         maxRecords: 1000
+        idleTimeBetweenReads: "1500 millis"
       }
       leaseDuration: "10 seconds"
       maxLeasesToStealAtOneTimeFactor: 2.0

modules/kinesis/src/main/scala/com/snowplowanalytics/snowplow/streams/kinesis/KinesisSourceConfig.scala

@@ -70,7 +70,7 @@ object KinesisSourceConfig {
   sealed trait Retrieval
 
   object Retrieval {
-    case class Polling(maxRecords: Int) extends Retrieval
+    case class Polling(maxRecords: Int, idleTimeBetweenReads: FiniteDuration) extends Retrieval
     case object FanOut extends Retrieval
 
     private[KinesisSourceConfig] def decoder(implicit c: Configuration) = deriveConfiguredDecoder[Retrieval]

modules/kinesis/src/main/scala/com/snowplowanalytics/snowplow/streams/kinesis/source/KCLScheduler.scala

@@ -79,8 +79,10 @@ private[source] object KCLScheduler {
             kinesisConfig.retrievalMode match {
               case KinesisSourceConfig.Retrieval.FanOut =>
                 new FanOutConfig(kinesisClient).streamName(kinesisConfig.streamName).applicationName(kinesisConfig.appName)
-              case KinesisSourceConfig.Retrieval.Polling(maxRecords) =>
-                val c = new PollingConfig(kinesisConfig.streamName, kinesisClient).maxRecords(maxRecords)
+              case KinesisSourceConfig.Retrieval.Polling(maxRecords, idleTimeBetweenReads) =>
+                val c = new PollingConfig(kinesisConfig.streamName, kinesisClient)
+                  .maxRecords(maxRecords)
+                  .idleTimeBetweenReadsInMillis(idleTimeBetweenReads.toMillis)
                 c.recordsFetcherFactory.maxPendingProcessRecordsInput(1)
                 c
             }

modules/kinesis/src/test/scala/com/snowplowanalytics/snowplow/streams/kinesis/KinesisSourceConfigSpec.scala

@@ -34,7 +34,8 @@ class KinesisSourceConfigSpec extends Specification {
       "workerIdentifier": "my-identifier",
       "retrievalMode": {
         "type": "Polling",
-        "maxRecords": 42
+        "maxRecords": 42,
+        "idleTimeBetweenReads": "1500 millis"
       },
       "initialPosition": {
         "type": "TrimHorizon"
@@ -56,7 +57,7 @@ class KinesisSourceConfigSpec extends Specification {
         c.streamName must beEqualTo("my-stream"),
         c.workerIdentifier must beEqualTo("my-identifier"),
         c.initialPosition must beEqualTo(KinesisSourceConfig.InitialPosition.TrimHorizon),
-        c.retrievalMode must beEqualTo(KinesisSourceConfig.Retrieval.Polling(42)),
+        c.retrievalMode must beEqualTo(KinesisSourceConfig.Retrieval.Polling(42, 1500.millis)),
         c.leaseDuration must beEqualTo(20.seconds),
         c.maxLeasesToStealAtOneTimeFactor must beEqualTo(BigDecimal(0.42)),
         c.checkpointThrottledBackoffPolicy must beEqualTo(BackoffPolicy(minBackoff = 100.millis, maxBackoff = 1.second)),
@@ -73,7 +74,8 @@ class KinesisSourceConfigSpec extends Specification {
       "workerIdentifier": "my-identifier",
       "retrievalMode": {
         "type": "POLLING",
-        "maxRecords": 42
+        "maxRecords": 42,
+        "idleTimeBetweenReads": "1500 millis"
       },
       "initialPosition": {
         "type": "TRIM_HORIZON"
@@ -95,7 +97,7 @@ class KinesisSourceConfigSpec extends Specification {
         c.streamName must beEqualTo("my-stream"),
         c.workerIdentifier must beEqualTo("my-identifier"),
         c.initialPosition must beEqualTo(KinesisSourceConfig.InitialPosition.TrimHorizon),
-        c.retrievalMode must beEqualTo(KinesisSourceConfig.Retrieval.Polling(42)),
+        c.retrievalMode must beEqualTo(KinesisSourceConfig.Retrieval.Polling(42, 1500.millis)),
         c.leaseDuration must beEqualTo(20.seconds),
         c.maxLeasesToStealAtOneTimeFactor must beEqualTo(BigDecimal(0.42)),
         c.checkpointThrottledBackoffPolicy must beEqualTo(BackoffPolicy(minBackoff = 100.millis, maxBackoff = 1.second)),
@@ -122,7 +124,7 @@ class KinesisSourceConfigSpec extends Specification {
       streamName                       = "my-stream",
       workerIdentifier                 = System.getenv("HOSTNAME"),
       initialPosition                  = KinesisSourceConfig.InitialPosition.Latest,
-      retrievalMode                    = KinesisSourceConfig.Retrieval.Polling(1000),
+      retrievalMode                    = KinesisSourceConfig.Retrieval.Polling(1000, 1500.millis),
       customEndpoint                   = None,
       dynamodbCustomEndpoint           = None,
       cloudwatchCustomEndpoint         = None,