Temporarily enable logging of pubsub message IDs

snowplow-incubator · Nov 8, 2024 · 042f02a · 042f02a
1 parent d883f4e
commit 042f02a
Show file tree

Hide file tree

Showing 3 changed files with 17 additions and 1 deletion.
diff --git a/modules/gcp/src/main/resources/application.conf b/modules/gcp/src/main/resources/application.conf
@@ -17,6 +17,7 @@
     "maxMessagesPerPull": 1000
     "debounceRequests": "100 millis"
     "prefetch": 4
+    "logMessageIds": false
   }
   "output": {
     "bad": ${snowplow.defaults.sinks.pubsub}

diff --git a/modules/gcp/src/main/scala/common-streams-extensions/v2/PubsubSourceConfigV2.scala b/modules/gcp/src/main/scala/common-streams-extensions/v2/PubsubSourceConfigV2.scala
@@ -25,13 +25,24 @@ case class PubsubSourceConfigV2(
   gcpUserAgent: GcpUserAgent,
   maxMessagesPerPull: Int,
   debounceRequests: FiniteDuration,
-  prefetch: Int
+  prefetch: Int,
+  logMessageIds: PubsubSourceConfigV2.CustomBoolean
 )
 
 object PubsubSourceConfigV2 {
 
   case class Subscription(projectId: String, subscriptionId: String)
 
+  case class CustomBoolean(value: Boolean) extends AnyVal
+
+  object CustomBoolean {
+    implicit def decoder: Decoder[CustomBoolean] =
+      Decoder.decodeBoolean
+        .or(Decoder.decodeString.emapTry(s => scala.util.Try(s.toBoolean)))
+        .map(CustomBoolean(_))
+
+  }
+
   object Subscription {
     implicit def show: Show[Subscription] = Show[Subscription] { s =>
       ProjectSubscriptionName.of(s.projectId, s.subscriptionId).toString

diff --git a/modules/gcp/src/main/scala/common-streams-extensions/v2/PubsubSourceV2.scala b/modules/gcp/src/main/scala/common-streams-extensions/v2/PubsubSourceV2.scala
@@ -91,6 +91,10 @@ object PubsubSourceV2 {
         val ackIds = records.map(_.getAckId)
         Sync[F].uncancelable { _ =>
           for {
+            _ <- if (config.logMessageIds.value) Sync[F].delay {
+                   println(records.map(_.getMessage.getMessageId).mkString("Pubsub message IDs: ", ",", ""))
+                 }
+                 else Sync[F].unit
             timeReceived <- Sync[F].realTimeInstant
             _ <- Utils.modAck[F](config.subscription, stub, ackIds, config.durationPerAckExtension)
             token <- Unique[F].unique