Enrich-kinesis: improve sink retry policy for when throttled by kines…

…is (close #697)
snowplow · Oct 6, 2022 · 27b15a7 · 27b15a7
1 parent 5c30da0
commit 27b15a7
Show file tree

Hide file tree

Showing 9 changed files with 226 additions and 131 deletions.
diff --git a/config/config.kinesis.extended.hocon b/config/config.kinesis.extended.hocon
@@ -89,16 +89,19 @@
       # Otherwise, the partition key will be a random UUID.
       # "partitionKey": "user_id"
 
-      # Optional. Policy to retry if writing to kinesis fails.
-      # This policy is used in 2 places:
-      # - When the PutRecords request errors
-      # - When the requests succeeds but some records couldn't get inserted
+      # Optional. Policy to retry if writing to kinesis fails with unexepected errors
       "backoffPolicy": {
         "minBackoff": 100 milliseconds
         "maxBackoff": 10 seconds
         "maxRetries": 10
       }
 
+      # Optional. Policy to retry if writing to kinesis exceeds the provisioned throughput.
+      "throttledBackoffPolicy": {
+        "minBackoff": 100 milliseconds
+        "maxBackoff": 1 second
+      }
+
       # Optional. Limits the number of events in a single PutRecords request.
       # Several requests are made in parallel
       # Maximum allowed: 500
@@ -136,16 +139,19 @@
       # Otherwise, the partition key will be a random UUID.
       # "partitionKey": "user_id"
 
-      # Optional. Policy to retry if writing to kinesis fails.
-      # This policy is used in 2 places:
-      # - When the PutRecords request errors
-      # - When the requests succeeds but some records couldn't get inserted
+      # Optional. Policy to retry if writing to kinesis fails with unexepcted errors
       "backoffPolicy": {
         "minBackoff": 100 milliseconds
         "maxBackoff": 10 seconds
         "maxRetries": 10
       }
 
+      # Optional. Policy to retry if writing to kinesis exceeds the provisioned throughput.
+      "throttledBackoffPolicy": {
+        "minBackoff": 100 milliseconds
+        "maxBackoff": 1 second
+      }
+
       # Optional. Limits the number of events in a single PutRecords request.
       # Several requests are made in parallel
       # Maximum allowed: 500
@@ -175,16 +181,19 @@
       # https://sdk.amazonaws.com/java/api/latest/software/amazon/awssdk/regions/providers/DefaultAwsRegionProviderChain.html
       "region": "eu-central-1"
 
-      # Optional. Policy to retry if writing to kinesis fails.
-      # This policy is used in 2 places:
-      # - When the PutRecords request errors
-      # - When the requests succeeds but some records couldn't get inserted
+      # Optional. Policy to retry if writing to kinesis fails with unexepcted errors
       "backoffPolicy": {
         "minBackoff": 100 milliseconds
         "maxBackoff": 10 seconds
         "maxRetries": 10
       }
 
+      # Optional. Policy to retry if writing to kinesis exceeds the provisioned throughput.
+      "throttledBackoffPolicy": {
+        "minBackoff": 100 milliseconds
+        "maxBackoff": 1 second
+      }
+
       # Optional. Limits the number of events in a single PutRecords request.
       # Several requests are made in parallel
       # Maximum allowed: 500

diff --git a/modules/common-fs2/src/main/scala/com/snowplowanalytics/snowplow/enrich/common/fs2/Run.scala b/modules/common-fs2/src/main/scala/com/snowplowanalytics/snowplow/enrich/common/fs2/Run.scala
@@ -17,7 +17,6 @@ import cats.implicits._
 
 import fs2.Stream
 
-import scala.concurrent.duration._
 import scala.concurrent.ExecutionContext
 
 import cats.effect.{Blocker, Clock, Concurrent, ConcurrentEffect, ContextShift, ExitCode, Resource, Sync, Timer}
@@ -26,13 +25,12 @@ import org.typelevel.log4cats.Logger
 import org.typelevel.log4cats.slf4j.Slf4jLogger
 
 import retry.syntax.all._
-import retry.RetryPolicies._
 
 import com.snowplowanalytics.snowplow.badrows.Processor
 
-import com.snowplowanalytics.snowplow.enrich.common.fs2.config.io.{Input, Monitoring, Output, RetryCheckpointing}
+import com.snowplowanalytics.snowplow.enrich.common.fs2.config.io.{BackoffPolicy, Input, Monitoring, Output, RetryCheckpointing}
 import com.snowplowanalytics.snowplow.enrich.common.fs2.config.{CliConfig, ParsedConfigs}
-import com.snowplowanalytics.snowplow.enrich.common.fs2.io.{FileSink, Source}
+import com.snowplowanalytics.snowplow.enrich.common.fs2.io.{FileSink, Retries, Source}
 import com.snowplowanalytics.snowplow.enrich.common.fs2.io.Clients.Client
 
 import org.http4s.client.{Client => Http4sClient}
@@ -109,9 +107,7 @@ object Run {
                                 val checkpointing = input match {
                                   case retrySettings: RetryCheckpointing =>
                                     withRetries(
-                                      retrySettings.checkpointBackoff.minBackoff,
-                                      retrySettings.checkpointBackoff.maxBackoff,
-                                      retrySettings.checkpointBackoff.maxRetries,
+                                      retrySettings.checkpointBackoff,
                                       "Checkpointing failed",
                                       checkpoint
                                     )
@@ -186,13 +182,11 @@ object Run {
     }
 
   private def withRetries[F[_]: Sync: Timer, A, B](
-    minBackOff: FiniteDuration,
-    maxBackOff: FiniteDuration,
-    maxRetries: Int,
+    config: BackoffPolicy,
     errorMessage: String,
     f: A => F[B]
   ): A => F[B] = { a =>
-    val retryPolicy = capDelay[F](maxBackOff, fullJitter[F](minBackOff)).join(limitRetries(maxRetries))
+    val retryPolicy = Retries.fullJitter[F](config)
 
     f(a)
       .retryingOnAllErrors(

diff --git a/...rc/main/scala/com/snowplowanalytics/snowplow/enrich/common/fs2/config/ParsedConfigs.scala b/...rc/main/scala/com/snowplowanalytics/snowplow/enrich/common/fs2/config/ParsedConfigs.scala
@@ -112,7 +112,7 @@ object ParsedConfigs {
             if (invalidAttributes.nonEmpty) NonEmptyList(invalidAttributes.head, invalidAttributes.tail.toList).invalid
             else output.valid
           }
-      case OutputConfig.Kinesis(_, _, Some(key), _, _, _, _) if !enrichedFieldsMap.contains(key) =>
+      case OutputConfig.Kinesis(_, _, Some(key), _, _, _, _, _) if !enrichedFieldsMap.contains(key) =>
         NonEmptyList.one(s"Partition key $key not valid").invalid
       case _ =>
         output.valid
@@ -126,7 +126,7 @@ object ParsedConfigs {
             attributes.contains(s)
         }
         attributesFromFields(fields)
-      case OutputConfig.Kinesis(_, _, Some(key), _, _, _, _) =>
+      case OutputConfig.Kinesis(_, _, Some(key), _, _, _, _, _) =>
         val fields = ParsedConfigs.enrichedFieldsMap.filter {
           case (s, _) =>
             s == key

diff --git a/...ommon-fs2/src/main/scala/com/snowplowanalytics/snowplow/enrich/common/fs2/config/io.scala b/...ommon-fs2/src/main/scala/com/snowplowanalytics/snowplow/enrich/common/fs2/config/io.scala
@@ -63,7 +63,7 @@ object io {
   case class BackoffPolicy(
     minBackoff: FiniteDuration,
     maxBackoff: FiniteDuration,
-    maxRetries: Int
+    maxRetries: Option[Int]
   )
   object BackoffPolicy {
     implicit def backoffPolicyDecoder: Decoder[BackoffPolicy] =
@@ -269,6 +269,7 @@ object io {
       region: Option[String],
       partitionKey: Option[String],
       backoffPolicy: BackoffPolicy,
+      throttledBackoffPolicy: BackoffPolicy,
       recordLimit: Int,
       byteLimit: Int,
       customEndpoint: Option[URI]

diff --git a/...mmon-fs2/src/main/scala/com/snowplowanalytics/snowplow/enrich/common/fs2/io/Retries.scala b/...mmon-fs2/src/main/scala/com/snowplowanalytics/snowplow/enrich/common/fs2/io/Retries.scala
@@ -0,0 +1,34 @@
+/*
+ * Copyright (c) 2020-2021 Snowplow Analytics Ltd. All rights reserved.
+ *
+ * This program is licensed to you under the Apache License Version 2.0,
+ * and you may not use this file except in compliance with the Apache License Version 2.0.
+ * You may obtain a copy of the Apache License Version 2.0 at http://www.apache.org/licenses/LICENSE-2.0.
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the Apache License Version 2.0 is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the Apache License Version 2.0 for the specific language governing permissions and limitations there under.
+ */
+package com.snowplowanalytics.snowplow.enrich.common.fs2.io
+
+import retry.{RetryPolicies, RetryPolicy}
+
+import cats.Applicative
+
+import com.snowplowanalytics.snowplow.enrich.common.fs2.config.io.BackoffPolicy
+
+object Retries {
+
+  def fullJitter[F[_]: Applicative](config: BackoffPolicy): RetryPolicy[F] =
+    capBackoffAndRetries(config, RetryPolicies.fullJitter[F](config.minBackoff))
+
+  def fibonacci[F[_]: Applicative](config: BackoffPolicy): RetryPolicy[F] =
+    capBackoffAndRetries(config, RetryPolicies.fibonacciBackoff[F](config.minBackoff))
+
+  private def capBackoffAndRetries[F[_]: Applicative](config: BackoffPolicy, policy: RetryPolicy[F]): RetryPolicy[F] = {
+    val capped = RetryPolicies.capDelay[F](config.maxBackoff, policy)
+    config.maxRetries.fold(capped)(max => capped.join(RetryPolicies.limitRetries(max)))
+  }
+
+}
diff --git a/...c/test/scala/com/snowplowanalytics/snowplow/enrich/common/fs2/config/ConfigFileSpec.scala b/...c/test/scala/com/snowplowanalytics/snowplow/enrich/common/fs2/config/ConfigFileSpec.scala
@@ -103,7 +103,7 @@ class ConfigFileSpec extends Specification with CatsIO {
           io.Input.Kinesis.InitPosition.TrimHorizon,
           io.Input.Kinesis.Retrieval.Polling(10000),
           3,
-          io.BackoffPolicy(100.milli, 10.second, 10),
+          io.BackoffPolicy(100.milli, 10.second, Some(10)),
           None,
           None,
           None
@@ -113,7 +113,8 @@ class ConfigFileSpec extends Specification with CatsIO {
             "enriched",
             Some("eu-central-1"),
             None,
-            io.BackoffPolicy(100.millis, 10.seconds, 10),
+            io.BackoffPolicy(100.millis, 10.seconds, Some(10)),
+            io.BackoffPolicy(100.millis, 1.second, None),
             500,
             5242880,
             None
@@ -123,7 +124,8 @@ class ConfigFileSpec extends Specification with CatsIO {
               "pii",
               Some("eu-central-1"),
               None,
-              io.BackoffPolicy(100.millis, 10.seconds, 10),
+              io.BackoffPolicy(100.millis, 10.seconds, Some(10)),
+              io.BackoffPolicy(100.millis, 1.second, None),
               500,
               5242880,
               None
@@ -133,7 +135,8 @@ class ConfigFileSpec extends Specification with CatsIO {
             "bad",
             Some("eu-central-1"),
             None,
-            io.BackoffPolicy(100.millis, 10.seconds, 10),
+            io.BackoffPolicy(100.millis, 10.seconds, Some(10)),
+            io.BackoffPolicy(100.millis, 1.second, None),
             500,
             5242880,
             None
@@ -208,7 +211,7 @@ class ConfigFileSpec extends Specification with CatsIO {
             true
           ),
           "raw",
-          io.BackoffPolicy(100.millis, 10.seconds, 10)
+          io.BackoffPolicy(100.millis, 10.seconds, Some(10))
         ),
         io.Outputs(
           io.Output.RabbitMQ(
@@ -227,7 +230,7 @@ class ConfigFileSpec extends Specification with CatsIO {
             ),
             "enriched",
             "enriched",
-            io.BackoffPolicy(100.millis, 10.seconds, 10)
+            io.BackoffPolicy(100.millis, 10.seconds, Some(10))
           ),
           None,
           io.Output.RabbitMQ(
@@ -246,7 +249,7 @@ class ConfigFileSpec extends Specification with CatsIO {
             ),
             "bad-1",
             "bad-1",
-            io.BackoffPolicy(100.millis, 10.seconds, 10)
+            io.BackoffPolicy(100.millis, 10.seconds, Some(10))
           )
         ),
         io.Concurrency(256, 3),

diff --git a/modules/kinesis/src/main/resources/application.conf b/modules/kinesis/src/main/resources/application.conf
@@ -25,6 +25,10 @@
         "maxBackoff": 10 seconds
         "maxRetries": 10
       }
+      "throttledBackoffPolicy": {
+        "minBackoff": 100 milliseconds
+        "maxBackoff": 1 second
+      }
       "recordLimit": 500
       "byteLimit": 5242880
     }
@@ -38,6 +42,10 @@
         "maxBackoff": 10 seconds
         "maxRetries": 10
       }
+      "throttledBackoffPolicy": {
+        "minBackoff": 100 milliseconds
+        "maxBackoff": 1 second
+      }
       "recordLimit": 500
       "byteLimit": 5242880
     }
@@ -49,6 +57,10 @@
         "maxBackoff": 10 seconds
         "maxRetries": 10
       }
+      "throttledBackoffPolicy": {
+        "minBackoff": 100 milliseconds
+        "maxBackoff": 1 second
+      }
       "recordLimit": 500
       "byteLimit": 5242880
     }