Fix the flaky test

zio · Oct 29, 2023 · beb1817 · beb1817 · github-actions · Oct 29, 2023
1 parent ab3c17d
commit beb1817
Show file tree

Hide file tree

Showing 2 changed files with 5 additions and 2 deletions.
diff --git a/zio-kafka-test/src/test/scala/zio/kafka/consumer/ConsumerSpec.scala b/zio-kafka-test/src/test/scala/zio/kafka/consumer/ConsumerSpec.scala
@@ -377,7 +377,7 @@ object ConsumerSpec extends ZIOSpecDefaultSlf4j with KafkaRandom {
             subscriptions.isEmpty
           )
         }
-      } @@ flaky(3),
+      },
       test("a slow producer doesnot interrupt the stream") {
         ZIO.scoped {
           for {

diff --git a/zio-kafka/src/main/scala/zio/kafka/consumer/internal/PartitionStreamControl.scala b/zio-kafka/src/main/scala/zio/kafka/consumer/internal/PartitionStreamControl.scala
@@ -154,7 +154,10 @@ object PartitionStreamControl {
                    dataQueue.takeAll.flatMap(data => if (data.isEmpty) requestAndAwaitData else ZIO.succeed(data))
                  }.flattenTake
                    .chunksWith(_.tap(records => registerPull(queueInfo, records)))
-                   .interruptWhen(interruptionPromise)
+                   // Due to https://github.com/zio/zio/issues/8515 we cannot use Zstream.interruptWhen.
+                   .mapZIO { chunk =>
+                     interruptionPromise.await.whenZIO(interruptionPromise.isDone).as(chunk)
+                   }
     } yield new PartitionStreamControl(
       tp,
       stream,
Benchmark suite	Current: `beb1817`	Previous: `c64a16e`	Ratio
`zio.kafka.bench.ConsumerBenchmark.throughputWithCommits`	`1598.7258690799997` ms/op	`564.9141662200001` ms/op	`2.83`
`zio.kafka.bench.comparison.ZioKafkaBenchmarks.manualZioKafka`	`5350.5568240799985` ms/op	`1567.2222182400003` ms/op	`3.41`
`zio.kafka.bench.comparison.ZioKafkaBenchmarks.zioKafka`	`5509.9517172` ms/op	`1346.20013004` ms/op	`4.09`