Fix concurrent modification exception in Consumer release (#1365)

By using a semaphore also for the close operation we guard against
concurrent access during release. This reverts a change made in in
#1011.

The cause of this concurrent access after the Consumer has shutdown is
unfortunately unknown and I haven't been able to reproduce it.

Fixes #1238.

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

@@ -50,6 +50,7 @@ private[consumer] object ConsumerAccess {
 
   def make(settings: ConsumerSettings): ZIO[Scope, Throwable, ConsumerAccess] =
     for {
+      access <- Semaphore.make(1)
       consumer <- ZIO.acquireRelease {
                     ZIO.attemptBlocking {
                       new KafkaConsumer[Array[Byte], Array[Byte]](
@@ -59,10 +60,9 @@ private[consumer] object ConsumerAccess {
                       )
                     }
                   } { consumer =>
-                    ZIO.blocking(ZIO.attempt(consumer.close(settings.closeTimeout))).orDie
+                    ZIO.blocking(access.withPermit(ZIO.attempt(consumer.close(settings.closeTimeout)))).orDie
                   }
-      result <- make(consumer)
-    } yield result
+    } yield new ConsumerAccess(consumer, access)
 
   def make(consumer: ByteArrayKafkaConsumer): ZIO[Scope, Throwable, ConsumerAccess] =
     for {