Fetch consumerGroupMetaData less often

zio-kafka/src/main/scala/zio/kafka/consumer/internal/Runloop.scala

@@ -113,7 +113,8 @@ private[consumer] final class Runloop private (
     partitionStreams: Chunk[PartitionStreamControl],
     pendingRequests: Chunk[RunloopCommand.Request],
     ignoreRecordsForTps: Set[TopicPartition],
-    polledRecords: ConsumerRecords[Array[Byte], Array[Byte]]
+    polledRecords: ConsumerRecords[Array[Byte], Array[Byte]],
+    consumerGroupMetadata: Option[ConsumerGroupMetadata]
   ): UIO[Runloop.FulfillResult] = {
     type Record = CommittableRecord[Array[Byte], Array[Byte]]
 
@@ -128,7 +129,6 @@ private[consumer] final class Runloop private (
     if (streams.isEmpty) ZIO.succeed(fulfillResult)
     else {
       for {
-        consumerGroupMetadata <- getConsumerGroupMetadataIfAny
         _ <- ZIO.foreachParDiscard(streams) { streamControl =>
                val tp      = streamControl.tp
                val records = polledRecords.records(tp)
@@ -153,8 +153,9 @@ private[consumer] final class Runloop private (
     }
   }
 
-  private val getConsumerGroupMetadataIfAny: UIO[Option[ConsumerGroupMetadata]] =
-    if (settings.hasGroupId) consumer.runloopAccess(c => ZIO.attempt(c.groupMetadata())).fold(_ => None, Some(_))
+  private def getConsumerGroupMetadataIfAny(c: ByteArrayKafkaConsumer): UIO[Option[ConsumerGroupMetadata]] =
+    if (settings.hasGroupId)
+      ZIO.attempt(c.groupMetadata()).fold(_ => None, Option.apply)
     else ZIO.none
 
   /** @return the topic-partitions for which received records should be ignored */
@@ -204,7 +205,7 @@ private[consumer] final class Runloop private (
           settings.authErrorRetrySchedule
       )
 
-  private def handlePoll(state: State): Task[State] = {
+  private def handlePoll(state: State, initialConsumerGroupMetadata: Option[ConsumerGroupMetadata]): Task[State] = {
     for {
       partitionsToFetch  <- settings.fetchStrategy.selectPartitionsToFetch(state.assignedStreams)
       pendingCommitCount <- committer.pendingCommitCount
@@ -243,7 +244,8 @@ private[consumer] final class Runloop private (
                                   records = polledRecords,
                                   ignoreRecordsForTps = Set.empty,
                                   pendingRequests = state.pendingRequests,
-                                  assignedStreams = state.assignedStreams
+                                  assignedStreams = state.assignedStreams,
+                                  consumerGroupMetadata = initialConsumerGroupMetadata
                                 )
                               )
 
@@ -252,9 +254,10 @@ private[consumer] final class Runloop private (
                               //   some partitions were assigned, revoked or lost,
                               //   some streams have ended.
 
-                              val currentAssigned = c.assignment().asScala.toSet
-                              val endedTps        = endedStreams.map(_.tp).toSet
                               for {
+                                consumerGroupMetadata <- getConsumerGroupMetadataIfAny(c)
+                                currentAssigned = c.assignment().asScala.toSet
+                                endedTps        = endedStreams.map(_.tp).toSet
                                 ignoreRecordsForTps <- doSeekForNewPartitions(c, assignedTps)
 
                                 // The topic partitions that need a new stream are:
@@ -319,7 +322,8 @@ private[consumer] final class Runloop private (
                                 records = polledRecords,
                                 ignoreRecordsForTps = ignoreRecordsForTps,
                                 pendingRequests = updatedPendingRequests,
-                                assignedStreams = updatedAssignedStreams
+                                assignedStreams = updatedAssignedStreams,
+                                consumerGroupMetadata = consumerGroupMetadata
                               )
                           }
           } yield pollresult
@@ -328,7 +332,8 @@ private[consumer] final class Runloop private (
                          pollResult.assignedStreams,
                          pollResult.pendingRequests,
                          pollResult.ignoreRecordsForTps,
-                         pollResult.records
+                         pollResult.records,
+                         pollResult.consumerGroupMetadata
                        )
       _ <- committer.cleanupPendingCommits
       _ <- checkStreamPullInterval(pollResult.assignedStreams)
@@ -497,10 +502,11 @@ private[consumer] final class Runloop private (
           _ <- ZIO.logDebug(s"Processing ${commands.size} commands: ${commands.mkString(",")}")
           _ <- consumer.runloopAccess(committer.processQueuedCommits(_))
           streamCommands = commands.collect { case cmd: RunloopCommand.StreamCommand => cmd }
-          stateAfterCommands <- ZIO.foldLeft(streamCommands)(state)(handleCommand)
+          stateAfterCommands           <- ZIO.foldLeft(streamCommands)(state)(handleCommand)
+          initialConsumerGroupMetadata <- consumer.withConsumerZIO(c => getConsumerGroupMetadataIfAny(c))
 
           updatedStateAfterPoll <- shouldPoll(stateAfterCommands).flatMap {
-                                     case true  => handlePoll(stateAfterCommands)
+                                     case true  => handlePoll(stateAfterCommands, initialConsumerGroupMetadata)
                                      case false => ZIO.succeed(stateAfterCommands)
                                    }
           // Immediately poll again, after processing all new queued commands
@@ -564,7 +570,8 @@ object Runloop {
     records: ConsumerRecords[Array[Byte], Array[Byte]],
     ignoreRecordsForTps: Set[TopicPartition],
     pendingRequests: Chunk[RunloopCommand.Request],
-    assignedStreams: Chunk[PartitionStreamControl]
+    assignedStreams: Chunk[PartitionStreamControl],
+    consumerGroupMetadata: Option[ConsumerGroupMetadata]
   )
   private final case class RevokeResult(
     pendingRequests: Chunk[RunloopCommand.Request],