Make PartitionFlowSpec deterministic under the poll gates (evolution-gaming#886)

Five tests failed intermittently, 113-168 times per 200 iterations, while
passing when run in isolation. Both poll gates compare strictly (clock isAfter
<at>, PartitionFlow.scala:275 and :279) against refs seeded with the flow's
acquisition instant, read from a millisecond-truncated clock. A poll takes a few
hundred microseconds, so the opening polls finish inside that same millisecond
and skip the gated work - no timers triggered, no commit offset computed - and
the assertions that follow see no scheduled commit, or a committed offset that
never advanced. Records still fold and later polls evaluate, so the fault is in
the tests.

Running those bodies under TestControl and stepping virtual time by 1 ms before
the first poll opens the gates deterministically, at no wall-clock cost. All
five measure 0/200 afterwards, with no assertion changed.

Author: tobiajo

core/src/test/scala/com/evolutiongaming/kafka/flow/PartitionFlowSpec.scala

@@ -1,5 +1,6 @@
 package com.evolutiongaming.kafka.flow
 
+import cats.effect.testkit.TestControl
 import cats.effect.unsafe.IORuntime
 import cats.effect.{IO, Ref, Resource}
 import cats.syntax.all.*
@@ -67,14 +68,16 @@ class PartitionFlowSpec extends FunSuite {
 
     val flow = f.flow use { flow =>
       for {
+        // step past the acquisition ms: poll gates are strict `isAfter`
+        _ <- IO.sleep(1.milli)
         // When("exactly 3 messages come")
         _      <- flow(f.records("key1", 100, List("event1", "event2", "event3")))
         offset <- f.pendingOffset.get
         // Then("offset of first message is returned")
         _ <- IO { assertEquals(offset, Some(Offset.unsafe(103))) }
       } yield ()
     }
-    flow.unsafeRunSync()
+    TestControl.executeEmbed(flow).unsafeRunSync()
   }
 
   test("PartitionFlow does not allow commit until working with key is finished") {
@@ -84,6 +87,8 @@ class PartitionFlowSpec extends FunSuite {
 
     val flow = f.flow use { flow =>
       for {
+        // step past the acquisition ms: poll gates are strict `isAfter`
+        _ <- IO.sleep(1.milli)
         // When("2 messages come for the key1")
         _ <- flow(f.records("key1", 100, List("event1", "event2")))
         // And("2 messages come for the key2")
@@ -102,7 +107,7 @@ class PartitionFlowSpec extends FunSuite {
 
       } yield ()
     }
-    flow.unsafeRunSync()
+    TestControl.executeEmbed(flow).unsafeRunSync()
 
   }
 
@@ -192,6 +197,8 @@ class PartitionFlowSpec extends FunSuite {
 
     val flow = f.flow.use { flow =>
       for {
+        // step past the acquisition ms: poll gates are strict `isAfter`
+        _ <- IO.sleep(1.milli)
         // The first two events for each state are handled without errors, offset is committed
         _ <- flow(f.records("key1", 100, List("event1", "event2")) ++ f.records("key2", 102, List("event3", "event4")))
         _ <- f.pendingOffset.get.map(offset => assertEquals(offset, Some(Offset.unsafe(104))))
@@ -204,7 +211,7 @@ class PartitionFlowSpec extends FunSuite {
       } yield ()
     }
 
-    flow.unsafeRunSync()
+    TestControl.executeEmbed(flow).unsafeRunSync()
   }
 
   test("PartitionFlow filters out events but commits offsets") {
@@ -247,6 +254,8 @@ class PartitionFlowSpec extends FunSuite {
 
     val flow = f.flow.use { flow =>
       for {
+        // step past the acquisition ms: poll gates are strict `isAfter`
+        _ <- IO.sleep(1.milli)
         // Only one key is processed and persisted, the second one is not, but the latest offset is committed nonetheless
         _ <- flow(f.records(processedKey, 100, List("event1")) ++ f.records(skippedKey, 101, List("event2")))
         _ <- f.pendingOffset.get.map(offset => assertEquals(offset, Some(Offset.unsafe(102))))
@@ -265,7 +274,7 @@ class PartitionFlowSpec extends FunSuite {
       } yield ()
     }
 
-    flow.unsafeRunSync()
+    TestControl.executeEmbed(flow).unsafeRunSync()
   }
 
   test("RemapKeys derives keys correctly and updates them before applying filters and folds") {
@@ -294,6 +303,8 @@ class PartitionFlowSpec extends FunSuite {
         )
 
         for {
+          // step past the acquisition ms: poll gates are strict `isAfter`
+          _ <- IO.sleep(1.milli)
           // Ensure pre-existing data is loaded correctly from the storage
           _ <- cache.keys.map(keys => assertEquals(keys.size, 1))
           _ <- keys.get.map(keys => assertEquals(keys, Set(initialKey)))
@@ -319,7 +330,7 @@ class PartitionFlowSpec extends FunSuite {
 
     }
 
-    test.unsafeRunSync()
+    TestControl.executeEmbed(test).unsafeRunSync()
   }
 
   test(