[GitHub] [openwhisk] chetanmeh commented on a change in pull request #4887: Add RestartSource to restart failing Kafka consumer for user-events service

[GitBox <gi...@apache.org>]

chetanmeh commented on a change in pull request #4887:
URL: https://github.com/apache/openwhisk/pull/4887#discussion_r412176055



##########
File path: core/monitoring/user-events/src/main/scala/org/apache/openwhisk/core/monitoring/metrics/EventConsumer.scala
##########
@@ -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) { () =>

Review comment:
       May be add a logger which we can later check how many restart happened




----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
users@infra.apache.org