Use ZIO's timed, and make runloopMetricsSchedule configurable

Also: simplify runloop construction.
zio · Jan 14, 2024 · 90ff904 · 90ff904
1 parent f137a70
commit 90ff904
Show file tree

Hide file tree

Showing 3 changed files with 140 additions and 164 deletions.
diff --git a/zio-kafka/src/main/scala/zio/kafka/consumer/ConsumerSettings.scala b/zio-kafka/src/main/scala/zio/kafka/consumer/ConsumerSettings.scala
@@ -32,7 +32,8 @@ final case class ConsumerSettings(
   rebalanceSafeCommits: Boolean = false,
   maxRebalanceDuration: Option[Duration] = None,
   fetchStrategy: FetchStrategy = QueueSizeBasedFetchStrategy(),
-  metricLabels: Set[MetricLabel] = Set.empty
+  metricLabels: Set[MetricLabel] = Set.empty,
+  runloopMetricsSchedule: Schedule[Any, Unit, Long] = Schedule.fixed(500.millis)
 ) {
 
   /**
@@ -295,6 +296,14 @@ final case class ConsumerSettings(
   def withMetricsLabels(metricLabels: Set[MetricLabel]): ConsumerSettings =
     copy(metricLabels = metricLabels)
 
+  /**
+   * @param runloopMetricsSchedule
+   *   The schedule at which the runloop metrics are measured. Example runloop metrics are queue sizes and number of
+   *   outstanding commits. The default is to measure every 500ms.
+   */
+  def withRunloopMetricsSchedule(runloopMetricsSchedule: Schedule[Any, Unit, Long]): ConsumerSettings =
+    copy(runloopMetricsSchedule = runloopMetricsSchedule)
+
 }
 
 object ConsumerSettings {