[greyhound] parallel consumer - add logs (#36082)

[greyhound] parllel consumer - add logs GitOrigin-RevId: 2d3421c1d110e20b68222e195cd82227b772dc93
wix · Sep 23, 2023 · e973c59 · e973c59
1 parent e882a4b
commit e973c59
Show file tree

Hide file tree

Showing 2 changed files with 12 additions and 3 deletions.
diff --git a/core/src/main/scala/com/wixpress/dst/greyhound/core/consumer/OffsetsInitializer.scala b/core/src/main/scala/com/wixpress/dst/greyhound/core/consumer/OffsetsInitializer.scala
@@ -1,7 +1,7 @@
 package com.wixpress.dst.greyhound.core.consumer
 
 import java.time.Clock
-import com.wixpress.dst.greyhound.core.consumer.ConsumerMetric.{CommittedMissingOffsets, CommittedMissingOffsetsFailed, SkippedGapsOnInitialization}
+import com.wixpress.dst.greyhound.core.consumer.ConsumerMetric.{CommittedMissingOffsets, CommittedMissingOffsetsFailed, FoundGapsOnInitialization, SkippedGapsOnInitialization}
 import com.wixpress.dst.greyhound.core.{ClientId, Group, Offset, OffsetAndMetadata, TopicPartition}
 import com.wixpress.dst.greyhound.core.metrics.{GreyhoundMetric, GreyhoundMetrics}
 import zio.{URIO, ZIO}
@@ -82,8 +82,11 @@ class OffsetsInitializer(
     val toPause                             = seekTo.collect { case (k, SeekTo.Pause) => k }
     val seekToEndOffsets                    = fetchEndOffsets(seekToEndPartitions, timeout).mapValues(OffsetAndMetadata.apply)
     val gapsSmallestOffsets                 = OffsetsAndGaps.gapsSmallestOffsets(currentCommittedOffsets)
-    val seekToGapsOffsets                   = if (parallelConsumer) gapsSmallestOffsets else Map.empty
-    val toOffsets                           = seekToOffsets ++ seekToEndOffsets ++ seekToGapsOffsets
+
+    if (gapsSmallestOffsets.nonEmpty) reporter(FoundGapsOnInitialization(clientId, group, gapsSmallestOffsets))
+
+    val seekToGapsOffsets = if (parallelConsumer) gapsSmallestOffsets else Map.empty
+    val toOffsets         = seekToOffsets ++ seekToEndOffsets ++ seekToGapsOffsets
 
     if (!parallelConsumer && gapsSmallestOffsets.nonEmpty) reportSkippedGaps(currentCommittedOffsets)
     PartitionActions(offsetSeeks = toOffsets, partitionsToPause = toPause.toSet)

diff --git a/core/src/main/scala/com/wixpress/dst/greyhound/core/consumer/ReportingConsumer.scala b/core/src/main/scala/com/wixpress/dst/greyhound/core/consumer/ReportingConsumer.scala
@@ -421,4 +421,10 @@ object ConsumerMetric {
     currentCommittedOffsetsAndGaps: Map[TopicPartition, OffsetAndGaps]
   ) extends ConsumerMetric
 
+  case class FoundGapsOnInitialization(
+    clientId: ClientId,
+    group: Group,
+    gapsSmallestOffsets: Map[TopicPartition, OffsetAndMetadata]
+  ) extends ConsumerMetric
+
 }