Fetch consumerGroupMetaData less often

zio-kafka-test/src/test/scala/zio/kafka/consumer/SubscriptionsSpec.scala

@@ -128,14 +128,13 @@ object SubscriptionsSpec extends ZIOSpecDefaultSlf4j with KafkaRandom {
         firstMessagesRef <- Ref.make(("", ""))
         finalizersRef    <- Ref.make(Chunk.empty[String])
 
-        consumer           <- KafkaTestUtils.makeConsumer(client, Some(group))
-        consumer0IsRunning <- Promise.make[Nothing, Unit]
+        consumer <- KafkaTestUtils.makeConsumer(client, Some(group))
+
         c1Fib <- consumer
                    .plainStream(Subscription.topics(topic1), Serde.string, Serde.string)
                    // Here we delay each message to be sure that `consumer1` will fail while `consumer0` is still running
                    .mapZIO { r =>
-                     consumer0IsRunning.succeed(()) *> // Notify that `consumer0` is running
-                       firstMessagesRef.updateSome { case ("", v) => ("First consumer0 message", v) } *>
+                     firstMessagesRef.updateSome { case ("", v) => ("First consumer0 message", v) } *>
                        ZIO
                          .logDebug(s"Consumed ${counter.getAndIncrement()} records")
                          .delay(10.millis)
@@ -147,7 +146,7 @@ object SubscriptionsSpec extends ZIOSpecDefaultSlf4j with KafkaRandom {
                    .zipLeft(finalizersRef.update(_ :+ "consumer0 finalized"))
                    .fork
 
-        _ <- consumer0IsRunning.await // Wait to be sure that `consumer0` is running
+        _ <- ZIO.sleep(100.millis) // Wait to be sure that `consumer0` is running
 
         c2Fib <- consumer
                    .plainStream(

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

@@ -154,7 +154,8 @@ 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(_))
+    if (settings.hasGroupId)
+      consumer.rebalanceListenerAccess(c => ZIO.attempt(c.groupMetadata())).fold(_ => None, Some(_))
     else ZIO.none
 
   /** @return the topic-partitions for which received records should be ignored */