Streams core: Eager processing of inner streams (#151)

Common streams is built around the concept of a stream of streams:
`Stream[F, Stream[F, A]]`. The nesting of `Stream` is needed so the
Source has a way to force the app to immediately checkpoint. This is
needed e.g. at the end of a kinesis shard, or when kafka rebalances.

We already allowed eager processing of inner streams under some
circumstances, i.e. the app can start processing the next inner stream
while the preceding inner stream is still finalizing. But we did not
have eager processing under all circumstances.

This PR allows the eager processing feature to work in more scenarios.
It will have most impact on the kinesis source near a shard end, and in
particular an app that works with timed windows, e.g. Lake Loader.

Author: istreeter

modules/streams-core/src/main/scala/com.snowplowanalytics.snowplow/streams/internal/LowLevelSource.scala

@@ -38,9 +38,11 @@ private[streams] trait LowLevelSource[F[_], C] {
   /**
    * Provides a stream of stream of low level events
    *
-   * The inner streams are processed one at a time, with clean separation before starting the next
-   * inner stream. This is required e.g. for Kafka, where the end of a stream represents client
-   * rebalancing.
+   * The inner streams are processed sequentially, though processing may be interleaved during
+   * finalization — the next inner stream can begin while the previous is still finalizing.
+   *
+   * A new inner stream is emitted when the source requires a hard boundary, e.g. for Kafka when a
+   * client rebalance occurs, or for Kinesis when a shard ends.
    *
    * A new [[EventProcessor]] will be invoked for each inner stream
    *
@@ -107,35 +109,35 @@ private[streams] object LowLevelSource {
   def toSourceAndAck[F[_]: Async, C](source: LowLevelSource[F, C]): F[SourceAndAck[F]] =
     for {
       stateRef <- Ref[F].of[InternalState](InternalState.Disconnected)
-    } yield sourceAndAckImpl(source, stateRef)
+      acksRef <- Ref[F].of(Map.empty[Unique.Token, C])
+    } yield sourceAndAckImpl(source, stateRef, acksRef)
 
   private def sourceAndAckImpl[F[_]: Async, C](
     source: LowLevelSource[F, C],
-    stateRef: Ref[F, InternalState]
+    stateRef: Ref[F, InternalState],
+    acksRef: Ref[F, Map[Unique.Token, C]]
   ): SourceAndAck[F] = new SourceAndAck[F] {
     def stream(config: EventProcessingConfig[F], processor: EventProcessor[F]): Stream[F, Nothing] = {
-      val str = for {
+      val sinks = EagerWindows.pipes { control: EagerWindows.Control[F] =>
+        CleanCancellation(messageSink(processor, acksRef, source.checkpointer, control, source.debounceCheckpoints))
+      }
+
+      val tokenedSources = for {
         s2 <- source.stream
-        acksRef <- Stream.bracket(Ref[F].of(Map.empty[Unique.Token, C]))(nackUnhandled(source.checkpointer, _))
         now <- Stream.eval(Sync[F].realTime)
         _ <- Stream.bracket(stateRef.set(InternalState.AwaitingUpstream(now)))(_ => stateRef.set(InternalState.Disconnected))
-      } yield {
-        val tokenedSources = s2
-          .through(monitorLatency(config, stateRef))
-          .through(tokened(acksRef))
-          .through(windowed(config.windowing))
-
-        val sinks = EagerWindows.pipes { control: EagerWindows.Control[F] =>
-          CleanCancellation(messageSink(processor, acksRef, source.checkpointer, control, source.debounceCheckpoints))
+      } yield s2
+        .through(monitorLatency(config, stateRef))
+        .through(tokened(acksRef))
+        .through(windowed(config.windowing))
+
+      tokenedSources.flatten
+        .zip(sinks)
+        .map { case (tokenedSource, sink) => sink(tokenedSource) }
+        .parJoin(eagerness(config.windowing)) // so we start processing the next window while the previous window is still finishing up.
+        .onFinalize {
+          nackUnhandled(source.checkpointer, acksRef)
         }
-
-        tokenedSources
-          .zip(sinks)
-          .map { case (tokenedSource, sink) => sink(tokenedSource) }
-          .parJoin(eagerness(config.windowing)) // so we start processing the next window while the previous window is still finishing up.
-      }
-
-      str.flatten
     }
 
     def isHealthy(maxAllowedProcessingLatency: FiniteDuration): F[SourceAndAck.HealthStatus] =

modules/streams-core/src/test/scala/com.snowplowanalytics.snowplow/streams/internal/LowLevelSourceSpec.scala

@@ -317,10 +317,10 @@ class LowLevelSourceSpec extends Specification with CatsEffect {
         Action.ProcessorStartedWindow("1970-01-01T00:01:00Z"),
         Action.ProcessorReceivedEvents("1970-01-01T00:01:00Z", List("25", "26", "27", "28", "29", "30", "31", "32")),
         Action.ProcessorReceivedEvents("1970-01-01T00:01:20Z", List("33", "34", "35", "36", "37", "38", "39", "40")),
-        Action.ProcessorReachedEndOfWindow("1970-01-01T00:01:40Z"),
-        Action.Checkpointed(List("25", "26", "27", "28", "29", "30", "31", "32", "33", "34", "35", "36", "37", "38", "39", "40")),
         Action.ProcessorStartedWindow("1970-01-01T00:01:40Z"),
         Action.ProcessorReceivedEvents("1970-01-01T00:01:40Z", List("41", "42", "43", "44", "45", "46", "47", "48")),
+        Action.ProcessorReachedEndOfWindow("1970-01-01T00:01:40Z"),
+        Action.Checkpointed(List("25", "26", "27", "28", "29", "30", "31", "32", "33", "34", "35", "36", "37", "38", "39", "40")),
         Action.ProcessorReceivedEvents("1970-01-01T00:02:00Z", List("49", "50", "51", "52", "53", "54", "55", "56")),
         Action.ProcessorReceivedEvents("1970-01-01T00:02:20Z", List("57", "58", "59", "60", "61", "62", "63", "64")),
         Action.ProcessorReachedEndOfWindow("1970-01-01T00:02:25Z"),
@@ -355,10 +355,10 @@ class LowLevelSourceSpec extends Specification with CatsEffect {
         Action.ProcessorStartedWindow("1970-01-01T00:02:40Z"),
         Action.ProcessorReceivedEvents("1970-01-01T00:02:40Z", List("65", "66", "67", "68", "69", "70", "71", "72")),
         Action.ProcessorReceivedEvents("1970-01-01T00:03:00Z", List("73", "74", "75", "76", "77", "78", "79", "80")),
-        Action.ProcessorReachedEndOfWindow("1970-01-01T00:03:20Z"),
-        Action.Checkpointed(List("65", "66", "67", "68", "69", "70", "71", "72", "73", "74", "75", "76", "77", "78", "79", "80")),
         Action.ProcessorStartedWindow("1970-01-01T00:03:20Z"),
         Action.ProcessorReceivedEvents("1970-01-01T00:03:20Z", List("81", "82", "83", "84", "85", "86", "87", "88")),
+        Action.ProcessorReachedEndOfWindow("1970-01-01T00:03:20Z"),
+        Action.Checkpointed(List("65", "66", "67", "68", "69", "70", "71", "72", "73", "74", "75", "76", "77", "78", "79", "80")),
         Action.ProcessorReceivedEvents("1970-01-01T00:03:40Z", List("89", "90", "91", "92", "93", "94", "95", "96")),
         Action.ProcessorReachedEndOfWindow("1970-01-01T00:03:55Z"),
         Action.Checkpointed(List("81", "82", "83", "84", "85", "86", "87", "88", "89", "90", "91", "92", "93", "94", "95", "96"))
@@ -624,10 +624,10 @@ class LowLevelSourceSpec extends Specification with CatsEffect {
         Action.ProcessorStartedWindow("1970-01-01T00:01:00Z"),
         Action.ProcessorReceivedEvents("1970-01-01T00:01:00Z", List("25", "26", "27", "28", "29", "30", "31", "32")),
         Action.ProcessorReceivedEvents("1970-01-01T00:01:20Z", List("33", "34", "35", "36", "37", "38", "39", "40")),
-        Action.ProcessorReachedEndOfWindow("1970-01-01T00:01:40Z"),
-        Action.Checkpointed(List("25", "26", "27", "28", "29", "30", "31", "32", "33", "34", "35", "36", "37", "38", "39", "40")),
         Action.ProcessorStartedWindow("1970-01-01T00:01:40Z"),
         Action.ProcessorReceivedEvents("1970-01-01T00:01:40Z", List("41", "42", "43", "44", "45", "46", "47", "48")),
+        Action.ProcessorReachedEndOfWindow("1970-01-01T00:01:40Z"),
+        Action.Checkpointed(List("25", "26", "27", "28", "29", "30", "31", "32", "33", "34", "35", "36", "37", "38", "39", "40")),
         Action.ProcessorReceivedEvents("1970-01-01T00:02:00Z", List("49", "50", "51", "52", "53", "54", "55", "56")),
         Action.ProcessorReceivedEvents("1970-01-01T00:02:20Z", List("57", "58", "59", "60", "61", "62", "63", "64")),
         Action.ProcessorReachedEndOfWindow("1970-01-01T00:02:25Z"),
@@ -662,10 +662,10 @@ class LowLevelSourceSpec extends Specification with CatsEffect {
         Action.ProcessorStartedWindow("1970-01-01T00:02:40Z"),
         Action.ProcessorReceivedEvents("1970-01-01T00:02:40Z", List("65", "66", "67", "68", "69", "70", "71", "72")),
         Action.ProcessorReceivedEvents("1970-01-01T00:03:00Z", List("73", "74", "75", "76", "77", "78", "79", "80")),
-        Action.ProcessorReachedEndOfWindow("1970-01-01T00:03:20Z"),
-        Action.Checkpointed(List("65", "66", "67", "68", "69", "70", "71", "72", "73", "74", "75", "76", "77", "78", "79", "80")),
         Action.ProcessorStartedWindow("1970-01-01T00:03:20Z"),
         Action.ProcessorReceivedEvents("1970-01-01T00:03:20Z", List("81", "82", "83", "84", "85", "86", "87", "88")),
+        Action.ProcessorReachedEndOfWindow("1970-01-01T00:03:20Z"),
+        Action.Checkpointed(List("65", "66", "67", "68", "69", "70", "71", "72", "73", "74", "75", "76", "77", "78", "79", "80")),
         Action.ProcessorReceivedEvents("1970-01-01T00:03:40Z", List("89", "90", "91", "92", "93", "94", "95", "96")),
         Action.ProcessorReachedEndOfWindow("1970-01-01T00:03:55Z"),
         Action.Checkpointed(List("81", "82", "83", "84", "85", "86", "87", "88", "89", "90", "91", "92", "93", "94", "95", "96"))
@@ -1058,6 +1058,7 @@ object LowLevelSourceSpec {
       val start = IO.realTimeInstant.flatMap(t => ref.update(_ :+ Action.ProcessorStartedWindow(t.toString)))
       val end = for {
         t <- IO.realTimeInstant
+        _ <- IO.sleep(1.nanos) // Forces deterministic order of Actions
         _ <- ref.update(_ :+ Action.ProcessorReachedEndOfWindow(t.toString))
         _ <- IO.sleep(config.timeToFinalizeWindow)
         tokens <- checkpoints.get