Merge pull request #24 from ps-dev/adapt1-1533

Adapt1 1533 | Use offset-streams to check lag in internal topic

Author: abhivermaaa

ingestors/kafka/src/main/scala/hydra/kafka/algebras/ConsumerGroupsAlgebra.scala

@@ -33,6 +33,8 @@ trait ConsumerGroupsAlgebra[F[_]] {
 
   def getAllConsumers: F[List[ConsumerTopics]]
 
+  def getOffsetsForInternalConsumerGroup: F[List[PartitionOffset]]
+
   def getAllConsumersByTopic: F[List[TopicConsumers]]
 
   def startConsumer: F[Unit]
@@ -96,6 +98,7 @@ final case class TestConsumerGroupsAlgebra(consumerGroupMap: Map[TopicConsumerKe
 
   override def getUniquePerNodeConsumerGroup: String = "uniquePerNodeConsumerGroup"
 
+  override def getOffsetsForInternalConsumerGroup: IO[List[PartitionOffset]] = ???
 }
 
 object TestConsumerGroupsAlgebra {
@@ -148,11 +151,41 @@ object ConsumerGroupsAlgebra {
 
     for {
       consumerGroupsStorageFacade <- Ref[F].of(ConsumerGroupsStorageFacade.empty)
+      consumerGroupsOffsetFacade <- Ref[F].of(ConsumerGroupsOffsetFacade.empty)
     } yield new ConsumerGroupsAlgebra[F] {
 
       override def getConsumersForTopic(topicName: String): F[TopicConsumers] =
         consumerGroupsStorageFacade.get.flatMap(a => addStateToTopicConsumers(a.getConsumersForTopicName(topicName)))
 
+
+      override def getOffsetsForInternalConsumerGroup: F[List[PartitionOffset]] = {
+
+        for {
+          groupOffsetsFromOffsetStream <- consumerGroupsOffsetFacade.get.map(_.getAllPartitionOffset())
+
+          // TODO: To be optimized
+          largestOffsets <- kAA.getLatestOffsets(dvsConsumersTopic.value)
+            .map(_.map(k => PartitionOffset
+            (
+              k._1.partition,
+              groupOffsetsFromOffsetStream.getOrElse(k._1.partition, 0),
+              k._2.value,
+              -1
+            )).toList)
+
+          offsetsWithLag = largestOffsets
+            .map(
+              k => PartitionOffset
+              (
+                k.partition,
+                k.groupOffset,
+                k.largestOffset,
+                k.largestOffset - k.groupOffset
+              )
+            )
+        }yield offsetsWithLag
+      }
+
       private def addStateToTopicConsumers(topicConsumers: TopicConsumers): F[TopicConsumers] = {
         val detailedF: F[List[Consumer]] = topicConsumers.consumers.traverse { consumer =>
           val fState = getConsumerActiveState(consumer.consumerGroupName)
@@ -179,6 +212,7 @@ object ConsumerGroupsAlgebra {
             ConsumerGroupsOffsetConsumer.start(kafkaClientAlgebra, kAA, sra, uniquePerNodeConsumerGroup, consumerOffsetsOffsetsTopicConfig,
               kafkaInternalTopic, dvsConsumersTopic, bootstrapServers, commonConsumerGroup, kafkaClientSecurityConfig)
           }
+          _ <- Concurrent[F].start(consumeOffsetStreamIntoCache(offsetStream, consumerGroupsOffsetFacade))
         } yield ()
       }
 
@@ -233,6 +267,17 @@ object ConsumerGroupsAlgebra {
       .makeRetryableWithNotification(Infinite, "ConsumerGroupsAlgebra")
       .compile.drain
   }
+
+  private def consumeOffsetStreamIntoCache[F[_] : ContextShift : ConcurrentEffect : Timer : Logger](
+                                                                                                          offsetStream: fs2.Stream[F, Either[Throwable, (Partition, Offset)]],
+                                                                                                          consumerGroupsOffsetFacade: Ref[F, ConsumerGroupsOffsetFacade]
+                                                                                                        )(implicit notificationsService: InternalNotificationSender[F]): F[Unit] = {
+
+      offsetStream.evalTap {
+        case Right((partition, offset))  => consumerGroupsOffsetFacade.update(_.addOffset(partition, offset))
+        case _ => Logger[F].error("Error in consumeOffsetStreamIntoCache")
+    }.compile.drain
+  }
 }
 
 private case class ConsumerGroupsStorageFacade(consumerMap: Map[TopicConsumerKey, TopicConsumerValue]) {
@@ -268,14 +313,15 @@ private object ConsumerGroupsStorageFacade {
 
 private case class ConsumerGroupsOffsetFacade(offsetMap: Map[Partition, Offset]) {
 
-  def addOffset(key: Partition, value: Offset): ConsumerGroupsOffsetFacade = {
-    val res = this.copy(this.offsetMap + (key -> value))
-    println(this.offsetMap)
-    res
-  }
+  def addOffset(key: Partition, value: Offset): ConsumerGroupsOffsetFacade =
+    this.copy(this.offsetMap + (key -> value))
+
+  def getAllPartitionOffset(): Map[Partition, Offset] =
+    this.offsetMap
 
   def removeOffset(key: Partition): ConsumerGroupsOffsetFacade =
-      this.copy(this.offsetMap - key)
+    this.copy(this.offsetMap - key)
+
   }
 
   private object ConsumerGroupsOffsetFacade {

ingestors/kafka/src/main/scala/hydra/kafka/endpoints/ConsumerGroupsEndpoint.scala

@@ -34,6 +34,18 @@ class ConsumerGroupsEndpoint[F[_]: Futurable](consumerGroupsAlgebra: ConsumerGro
                   addHttpMetric("", StatusCodes.InternalServerError, "/v2/consumer-groups", startTime, method.value, error = Some(exception.getMessage))
                   complete(StatusCodes.InternalServerError, exception.getMessage)
               }
+            } ~ pathPrefix("hydra-internal-topic") {
+              val startTime = Instant.now
+              pathEndOrSingleSlash {
+                onComplete(Futurable[F].unsafeToFuture(consumerGroupsAlgebra.getOffsetsForInternalConsumerGroup)) {
+                  case Success(detailedConsumer) =>
+                    addHttpMetric("hydra-internal-topic", StatusCodes.OK, "/v2/consumer-groups/hydra-internal-topic", startTime, method.value)
+                    complete(StatusCodes.OK, detailedConsumer)
+                  case Failure(exception) =>
+                    addHttpMetric("hydra-internal-topic", StatusCodes.InternalServerError, "/v2/consumer-groups/hydra-internal-topic", startTime, method.value, error = Some(exception.getMessage))
+                    complete(StatusCodes.InternalServerError, exception.getMessage)
+                }
+              }
             } ~ pathPrefix(Segment) { consumerGroupName =>
               val startTime = Instant.now
               pathEndOrSingleSlash {