Add RestartSource to restart failing Kafka consumer for user-events service (#4887)

* Add RestartSource for user-events service

* Add missing configuration

Author: upgle

core/monitoring/user-events/src/main/resources/application.conf

@@ -15,6 +15,12 @@
 # limitations under the License.
 #
 
+akka.kafka.committer {
+
+  max-batch = 20
+
+}
+
 akka.kafka.consumer {
   # Properties defined by org.apache.kafka.clients.consumer.ConsumerConfig
   # can be defined in this configuration section.

core/monitoring/user-events/src/main/resources/reference.conf

@@ -31,5 +31,21 @@ whisk {
       # rename/relabel prometheus metrics tags
       # "namespace" = "ow_namespae"
     }
+
+    retry {
+      # minimum (initial) duration until the Kafka consumer is started again if it is terminated
+      min-backoff = 3 secs
+
+      # the exponential back-off is capped to this duration
+      max-backoff = 30 secs
+
+      # after calculation of the exponential back-off an additional
+      # random delay based on this factor is added, e.g. `0.2` adds up to `20%` delay
+      random-factor = 0.2
+
+      # the amount of restarts is capped to this amount within a time frame of minBackoff
+      max-restarts = 10
+    }
+
   }
 }

core/monitoring/user-events/src/main/scala/org/apache/openwhisk/core/monitoring/metrics/EventConsumer.scala

@@ -18,14 +18,14 @@
 package org.apache.openwhisk.core.monitoring.metrics
 
 import java.lang.management.ManagementFactory
+import java.util.concurrent.atomic.AtomicReference
 
 import akka.Done
 import akka.actor.ActorSystem
 import akka.kafka.scaladsl.{Committer, Consumer}
-import akka.kafka.scaladsl.Consumer.DrainingControl
 import akka.kafka.{CommitterSettings, ConsumerSettings, Subscriptions}
 import akka.stream.ActorMaterializer
-import akka.stream.scaladsl.Keep
+import akka.stream.scaladsl.{RestartSource, Sink}
 import javax.management.ObjectName
 import org.apache.kafka.clients.consumer.ConsumerConfig
 import kamon.Kamon
@@ -77,25 +77,34 @@ case class EventConsumer(settings: ConsumerSettings[String, String],
 
   def shutdown(): Future[Done] = {
     lagRecorder.cancel()
-    control.drainAndShutdown()(system.dispatcher)
+    control.get().drainAndShutdown(result)(system.dispatcher)
   }
 
-  def isRunning: Boolean = !control.isShutdown.isCompleted
-
-  override def metrics(): Future[Map[MetricName, common.Metric]] = control.metrics
-
-  private val committerSettings = CommitterSettings(system).withMaxBatch(20)
-
-  //TODO Use RestartSource
-  private val control: DrainingControl[Done] = Consumer
-    .committableSource(updatedSettings, Subscriptions.topics(userEventTopic))
-    .map { msg =>
-      processEvent(msg.record.value())
-      msg.committableOffset
+  def isRunning: Boolean = !control.get().isShutdown.isCompleted
+
+  override def metrics(): Future[Map[MetricName, common.Metric]] = control.get().metrics
+
+  private val committerSettings = CommitterSettings(system)
+  private val control = new AtomicReference[Consumer.Control](Consumer.NoopControl)
+
+  private val result = RestartSource
+    .onFailuresWithBackoff(
+      minBackoff = metricConfig.retry.minBackoff,
+      maxBackoff = metricConfig.retry.maxBackoff,
+      randomFactor = metricConfig.retry.randomFactor,
+      maxRestarts = metricConfig.retry.maxRestarts) { () =>
+      Consumer
+        .committableSource(updatedSettings, Subscriptions.topics(userEventTopic))
+        // this is to access to the Consumer.Control
+        // instances of the latest Kafka Consumer source
+        .mapMaterializedValue(c => control.set(c))
+        .map { msg =>
+          processEvent(msg.record.value())
+          msg.committableOffset
+        }
+        .via(Committer.flow(committerSettings))
     }
-    .toMat(Committer.sink(committerSettings))(Keep.both)
-    .mapMaterializedValue(DrainingControl.apply)
-    .run()
+    .runWith(Sink.ignore)
 
   private val lagRecorder =
     system.scheduler.schedule(10.seconds, 10.seconds)(lagGauge.update(consumerLag))

core/monitoring/user-events/src/main/scala/org/apache/openwhisk/core/monitoring/metrics/OpenWhiskEvents.scala

@@ -29,14 +29,18 @@ import org.apache.kafka.common.serialization.StringDeserializer
 import pureconfig._
 import pureconfig.generic.auto._
 
+import scala.concurrent.duration.FiniteDuration
 import scala.concurrent.{ExecutionContext, Future}
 
 object OpenWhiskEvents extends SLF4JLogging {
 
   case class MetricConfig(port: Int,
                           enableKamon: Boolean,
                           ignoredNamespaces: Set[String],
-                          renameTags: Map[String, String])
+                          renameTags: Map[String, String],
+                          retry: RetryConfig)
+
+  case class RetryConfig(minBackoff: FiniteDuration, maxBackoff: FiniteDuration, randomFactor: Double, maxRestarts: Int)
 
   def start(config: Config)(implicit system: ActorSystem,
                             materializer: ActorMaterializer): Future[Http.ServerBinding] = {

core/monitoring/user-events/src/test/resources/application.conf

@@ -29,4 +29,20 @@ user-events {
   rename-tags {
     #namespace = "ow_namespace"
   }
+
+  retry {
+    # minimum (initial) duration until the Kafka consumer is started again if it is terminated
+    min-backoff = 3 secs
+
+    # the exponential back-off is capped to this duration
+    max-backoff = 30 secs
+
+    # after calculation of the exponential back-off an additional
+    # random delay based on this factor is added, e.g. `0.2` adds up to `20%` delay
+    random-factor = 0.2
+
+    # the amount of restarts is capped to this amount within a time frame of minBackoff
+    max-restarts = 10
+  }
+
 }

core/monitoring/user-events/src/test/scala/org/apache/openwhisk/core/monitoring/metrics/PrometheusRecorderTests.scala

@@ -99,6 +99,12 @@ class PrometheusRecorderTests extends KafkaSpecBase with BeforeAndAfterEach with
             |    rename-tags {
             |      namespace = "ow_namespace"
             |    }
+            |    retry {
+            |      min-backoff = 3 secs
+            |      max-backoff = 30 secs
+            |      random-factor = 0.2
+            |      max-restarts = 10
+            |    }
             |  }
             | }
          """.stripMargin)