Use Semaphore (#1142)

Since Zio 2.0.20 Semaphore is fair so we can use it again. This reverts 2cc9efc.

build.sbt

@@ -100,8 +100,7 @@ lazy val zioKafka =
     .settings(
       libraryDependencies ++= Seq(
         kafkaClients,
-        scalaCollectionCompat,
-        "dev.zio" %% "zio-concurrent" % zioVersion.value
+        scalaCollectionCompat
       )
     )
 

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

@@ -8,18 +8,17 @@ import zio.kafka.consumer.ConsumerSettings
 import zio.kafka.consumer.internal.ConsumerAccess.ByteArrayKafkaConsumer
 
 import scala.jdk.CollectionConverters._
-import zio.concurrent.ReentrantLock
 
 private[consumer] final class ConsumerAccess(
   private[consumer] val consumer: ByteArrayKafkaConsumer,
-  access: ReentrantLock
+  access: Semaphore
 ) {
 
   def withConsumer[A](f: ByteArrayKafkaConsumer => A): Task[A] =
     withConsumerZIO[Any, A](c => ZIO.attempt(f(c)))
 
   def withConsumerZIO[R, A](f: ByteArrayKafkaConsumer => RIO[R, A]): RIO[R, A] =
-    access.lock.zipRight(withConsumerNoPermit(f)).ensuring(access.unlock)
+    access.withPermit(withConsumerNoPermit(f))
 
   private def withConsumerNoPermit[R, A](
     f: ByteArrayKafkaConsumer => RIO[R, A]
@@ -36,7 +35,7 @@ private[consumer] final class ConsumerAccess(
    * Use this method only from Runloop.
    */
   private[internal] def runloopAccess[R, E, A](f: ByteArrayKafkaConsumer => ZIO[R, E, A]): ZIO[R, E, A] =
-    access.lock.zipRight(f(consumer)).ensuring(access.unlock)
+    access.withPermit(f(consumer))
 
   /**
    * Use this method ONLY from the rebalance listener.
@@ -67,6 +66,6 @@ private[consumer] object ConsumerAccess {
 
   def make(consumer: ByteArrayKafkaConsumer): ZIO[Scope, Throwable, ConsumerAccess] =
     for {
-      access <- ReentrantLock.make(fairness = true)
+      access <- Semaphore.make(1)
     } yield new ConsumerAccess(consumer, access)
 }