Distribute records sequentially

It is running on the runloops single thread executor anyway!
zio · Dec 4, 2024 · e4bbd3d · e4bbd3d
1 parent 29ff1c2
commit e4bbd3d
Showing 1 changed file with 1 addition and 3 deletions.
diff --git a/zio-kafka/src/main/scala/zio/kafka/consumer/internal/Runloop.scala b/zio-kafka/src/main/scala/zio/kafka/consumer/internal/Runloop.scala
@@ -130,7 +130,7 @@ private[consumer] final class Runloop private (
       for {
         consumerGroupMetadata <- getConsumerGroupMetadataIfAny
         _ <- ZIO
-               .foreachParDiscard(streams) { streamControl =>
+               .foreachDiscard(streams) { streamControl =>
                  val tp      = streamControl.tp
                  val records = polledRecords.records(tp)
                  if (records.isEmpty) {
@@ -150,8 +150,6 @@ private[consumer] final class Runloop private (
                    streamControl.offerRecords(builder.result())
                  }
                }
-              .fork
-              .onExecutor(topLevelExecutor)
       } yield fulfillResult
     }
   }