Transformer kinesis: version 4.0.2 Throws java.lang.InterruptedExcept…

…ion: sleep interrupted (close #938)
snowplow · Jun 16, 2022 · b683757 · b683757
1 parent b7bbed0
commit b683757
Showing 1 changed file with 2 additions and 1 deletion.
diff --git a/...a/com/snowplowanalytics/snowplow/rdbloader/transformer/kinesis/sinks/generic/Record.scala b/...a/com/snowplowanalytics/snowplow/rdbloader/transformer/kinesis/sinks/generic/Record.scala
@@ -84,7 +84,8 @@ case object EndWindow extends Record[Nothing, Nothing, Nothing]
               // Same window, drop Left(w)
               go(Some(w), tail, emptyWindow)
             case Some(_) if emptyWindow =>
-                Pull.pure(())
+              // New window, but nothing to emit
+              go(Some(w), tail, true)
             case Some(_) =>
               Pull.output1[F, Record[W, A, S]](Record.EndWindow) *>
                 go(Some(w), tail, true)