amendment 1: throttle acks per second

Author: istreeter

modules/common-transformer-stream/src/main/scala/com/snowplowanalytics/snowplow/rdbloader/transformer/stream/common/Completion.scala

@@ -11,7 +11,8 @@ import java.net.URI
 
 import cats.implicits._
 
-import cats.effect.Sync
+import cats.effect.implicits._
+import cats.effect.{Async, Sync}
 
 import io.circe.syntax.EncoderOps
 
@@ -56,7 +57,7 @@ object Completion {
    * @param state
    *   all metadata shredder extracted from a batch
    */
-  def seal[F[_]: Sync, C: Checkpointer[F, *]](
+  def seal[F[_]: Async, C: Checkpointer[F, *]](
     blobStorage: BlobStorage[F],
     compression: Compression,
     getTypes: Set[Data.ShreddedType] => TypesInfo,
@@ -84,8 +85,9 @@ object Completion {
                 )
       body = message.selfDescribingData(legacyMessageFormat).asJson.noSpaces
       _ <- writeFile(blobStorage, shreddingCompletePath, body)
-      _ <- Checkpointer[F, C].checkpoint(state.checkpointer)
+      fiber <- Checkpointer[F, C].checkpoint(state.checkpointer).start
       _ <- producer.send(body)
+      _ <- fiber.join
     } yield ()
 
   def writeFile[F[_]: Sync](

modules/common-transformer-stream/src/main/scala/com/snowplowanalytics/snowplow/rdbloader/transformer/stream/common/Processing.scala

@@ -104,7 +104,7 @@ object Processing {
         .through(windowing)
 
     val sink: Pipe[F, Record[Window, List[(SinkPath, Transformed.Data)], State[C]], Unit] =
-      _.through(getSink(resources, config.output, config.formats))
+      _.through(getSink(resources, config.output, config.formats)).prefetch
         .evalMap(onComplete)
 
     Shutdown

modules/transformer-pubsub/src/main/scala/com/snowplowanalytics/snowplow/rdbloader/transformer/stream/pubsub/PubsubCheckpointer.scala

@@ -7,21 +7,28 @@
  */
 package com.snowplowanalytics.snowplow.rdbloader.transformer.stream.pubsub
 
-import cats.Applicative
+import cats.effect.Async
 import cats.implicits._
 
 import com.snowplowanalytics.snowplow.rdbloader.common.cloud.Queue
 import com.snowplowanalytics.snowplow.rdbloader.transformer.stream.common.sources.Checkpointer
 
+import scala.concurrent.duration.DurationLong
+
 case class PubsubCheckpointer[F[_]](acks: List[F[Unit]])
 
 object PubsubCheckpointer {
 
+  val acksPerSecond = 1000 // TODO: Find a neat way to make this configurable
+
   def checkpointer[F[_]](message: Queue.Consumer.Message[F]): PubsubCheckpointer[F] =
     PubsubCheckpointer[F](List(message.ack))
 
-  implicit def pubsubCheckPointer[F[_]: Applicative]: Checkpointer[F, PubsubCheckpointer[F]] = new Checkpointer[F, PubsubCheckpointer[F]] {
-    override def checkpoint(c: PubsubCheckpointer[F]): F[Unit] = c.acks.sequence_
+  implicit def pubsubCheckpointer[F[_]: Async]: Checkpointer[F, PubsubCheckpointer[F]] = new Checkpointer[F, PubsubCheckpointer[F]] {
+    override def checkpoint(c: PubsubCheckpointer[F]): F[Unit] =
+      c.acks.grouped(acksPerSecond).toList.traverse_ { acks =>
+        acks.sequence_ *> Async[F].sleep(1.second)
+      }
 
     override def combine(x: PubsubCheckpointer[F], y: PubsubCheckpointer[F]): PubsubCheckpointer[F] =
       PubsubCheckpointer[F](x.acks ::: y.acks)