Fix app from exiting on elasticsearch error responses (close #233)

Author: istreeter

core/src/main/scala/com.snowplowanalytics.stream/loader/clients/BulkSender.scala

@@ -27,14 +27,13 @@ import org.slf4j.Logger
 // Scala
 import scala.concurrent.Future
 import scala.concurrent.duration._
-import scala.util.Random
 
 // cats
 import cats.effect.{ContextShift, IO}
 import scala.concurrent.ExecutionContext
 import cats.{Applicative, Id}
 
-import retry.{PolicyDecision, RetryDetails, RetryPolicy}
+import retry.{RetryDetails, RetryPolicies, RetryPolicy}
 
 // Snowplow
 import com.snowplowanalytics.snowplow.scalatracker.Tracker
@@ -81,28 +80,25 @@ object BulkSender {
   def delayPolicy[M[_]: Applicative](
     maxAttempts: Int,
     maxConnectionWaitTimeMs: Long
-  ): RetryPolicy[M] =
-    RetryPolicy.lift { status =>
-      if (status.retriesSoFar >= maxAttempts) PolicyDecision.GiveUp
-      else {
-        val maxDelay                  = 2.milliseconds * Math.pow(2, status.retriesSoFar.toDouble).toLong
-        val randomDelayNanos          = (maxDelay.toNanos * Random.nextDouble()).toLong
-        val maxConnectionWaitTimeNano = maxConnectionWaitTimeMs * 1000
-        val delayNanos =
-          if (maxConnectionWaitTimeMs >= randomDelayNanos) maxConnectionWaitTimeNano
-          else randomDelayNanos
-        PolicyDecision.DelayAndRetry(new FiniteDuration(delayNanos, TimeUnit.NANOSECONDS))
-      }
-    }
+  ): RetryPolicy[M] = {
+    val basePolicy =
+      RetryPolicies
+        .fullJitter(20.milliseconds)
+        .join(RetryPolicies.limitRetries(maxAttempts))
+    RetryPolicies.limitRetriesByCumulativeDelay(
+      FiniteDuration(maxConnectionWaitTimeMs, TimeUnit.MILLISECONDS),
+      basePolicy
+    )
+  }
 
   def onError(log: Logger, tracker: Option[Tracker[Id]], connectionAttemptStartTime: Long)(
     error: Throwable,
     details: RetryDetails
   ): IO[Unit] = {
-    val duration = (error, details) match {
-      case (error, RetryDetails.GivingUp(_, totalDelay)) =>
+    val duration = details match {
+      case RetryDetails.GivingUp(_, totalDelay) =>
         IO(log.error("Storage threw an unexpected exception. Giving up ", error)).as(totalDelay)
-      case (error, RetryDetails.WillDelayAndRetry(nextDelay, retriesSoFar, cumulativeDelay)) =>
+      case RetryDetails.WillDelayAndRetry(nextDelay, retriesSoFar, cumulativeDelay) =>
         IO(
           log.error(
             s"Storage threw an unexpected exception, after $retriesSoFar retries. Next attempt in $nextDelay ",
@@ -127,12 +123,6 @@ object BulkSender {
     }
   }
 
-  /** Predicate about whether or not we should retry sending stuff to ES */
-  def exPredicate: Throwable => Boolean = {
-    case _: Exception => true
-    case _            => false
-  }
-
   def futureToTask[T](f: => Future[T]): IO[T] =
     IO.fromFuture(IO.delay(f))
 

elasticsearch/src/main/scala/com/snowplowanalytics/stream/loader/clients/ElasticsearchBulkSender.scala

@@ -20,6 +20,7 @@ import org.slf4j.LoggerFactory
 
 // Scala
 import scala.concurrent.ExecutionContext.Implicits.global
+import scala.concurrent.Future
 import scala.util.{Failure => SFailure, Success => SSuccess}
 
 import org.elasticsearch.client.RestClient
@@ -43,7 +44,7 @@ import cats.syntax.validated._
 
 import retry.implicits._
 import retry.{RetryDetails, RetryPolicy}
-import retry.CatsEffect._
+import retry._
 
 import com.snowplowanalytics.snowplow.scalatracker.Tracker
 
@@ -116,9 +117,11 @@ class ElasticsearchBulkSender(
 
   override def send(records: List[EmitterJsonInput]): List[EmitterJsonInput] = {
     val connectionAttemptStartTime = System.currentTimeMillis()
-    implicit def onErrorHandler: (Throwable, RetryDetails) => IO[Unit] =
+    val onErrorHandler: (Throwable, RetryDetails) => IO[Unit] =
       BulkSender.onError(log, tracker, connectionAttemptStartTime)
-    implicit def retryPolicy: RetryPolicy[IO] =
+    def onFailureHandler[A](res: Response[A], rd: RetryDetails): IO[Unit] =
+      onErrorHandler(res.error.asException, rd)
+    val retryPolicy: RetryPolicy[IO] =
       BulkSender.delayPolicy[IO](maxAttempts, maxConnectionWaitTimeMs)
 
     // oldFailures - failed at the transformation step
@@ -132,7 +135,12 @@ class ElasticsearchBulkSender(
     val newFailures: List[EmitterJsonInput] = if (actions.nonEmpty) {
       BulkSender
         .futureToTask(client.execute(bulk(actions)))
-        .retryingOnSomeErrors(BulkSender.exPredicate)
+        .retryingOnFailuresAndAllErrors(
+          r => r.isSuccess,
+          retryPolicy,
+          onFailureHandler,
+          onErrorHandler
+        )
         .map(extractResult(records))
         .attempt
         .unsafeRunSync() match {
@@ -168,12 +176,14 @@ class ElasticsearchBulkSender(
   def extractResult(
     records: List[EmitterJsonInput]
   )(response: Response[BulkResponse]): List[EmitterJsonInput] =
-    response.result.items
-      .zip(records)
-      .flatMap { case (bulkResponseItem, record) =>
-        handleResponse(bulkResponseItem.error.map(_.reason), record)
-      }
-      .toList
+    response.fold(records) { result =>
+      result.items
+        .zip(records)
+        .flatMap { case (bulkResponseItem, record) =>
+          handleResponse(bulkResponseItem.error.map(_.reason), record)
+        }
+        .toList
+    }
 
   def composeObject(jsonRecord: JsonRecord): ElasticsearchObject = {
     val index = jsonRecord.shard match {
@@ -186,18 +196,20 @@ class ElasticsearchBulkSender(
 
   /** Logs the cluster health */
   override def logHealth(): Unit =
-    client.execute(clusterHealth).onComplete {
-      case SSuccess(health) =>
-        health match {
-          case response =>
-            response.result.status match {
-              case "green"  => log.info("Cluster health is green")
-              case "yellow" => log.warn("Cluster health is yellow")
-              case "red"    => log.error("Cluster health is red")
-            }
-        }
-      case SFailure(e) => log.error("Couldn't retrieve cluster health", e)
-    }
+    client
+      .execute(clusterHealth)
+      .flatMap { health =>
+        health.fold(failure => Future.failed(failure.error.asException), Future.successful(_))
+      }
+      .onComplete {
+        case SSuccess(result) =>
+          result.status match {
+            case "green"  => log.info("Cluster health is green")
+            case "yellow" => log.warn("Cluster health is yellow")
+            case "red"    => log.error("Cluster health is red")
+          }
+        case SFailure(e) => log.error("Couldn't retrieve cluster health", e)
+      }
 
   /**
    * Handle the response given for a bulk request, by producing a failure if we failed to insert

project/Dependencies.scala

@@ -25,7 +25,7 @@ object Dependencies {
     val netty            = "4.1.67.Final" // Override provided version to fix security vulnerability
     val jackson          = "2.12.7"
     // Scala
-    val catsRetry        = "0.3.2"
+    val catsRetry        = "2.1.1"
     val circe            = "0.14.1"
     val decline          = "2.1.0"
     val snowplowTracker  = "1.0.0"
@@ -51,7 +51,7 @@ object Dependencies {
     val netty            = "io.netty"                         %  "netty-all"                    % V.netty
     val jacksonCbor      = "com.fasterxml.jackson.dataformat" % "jackson-dataformat-cbor"       % V.jackson // Override provided version to fix security vulnerability
     // Scala
-    val catsRetry        = "com.github.cb372"                 %% "cats-retry-cats-effect"       % V.catsRetry
+    val catsRetry        = "com.github.cb372"                 %% "cats-retry"                   % V.catsRetry
     val circeOptics      = "io.circe"                         %% "circe-optics"                 % V.circe
     val decline          = "com.monovore"                     %% "decline"                      % V.decline
     val jacksonScala     = "com.fasterxml.jackson.module"     %% "jackson-module-scala"         % V.jackson // Compatible version required for elastic4s