temp pubsub

snowplow-incubator · Sep 30, 2022 · cf106a7 · cf106a7
1 parent c560a5c
commit cf106a7
Show file tree

Hide file tree

Showing 5 changed files with 2 additions and 11 deletions.
diff --git a/config/config.pubsub.reference.hocon b/config/config.pubsub.reference.hocon
@@ -60,9 +60,6 @@
       # Max size of the batch in bytes before emitting
       # Default is 5MB
       "maxBatchBytes": 5000000
-      # The number of threads used internally by library to process the callback after message delivery
-      # Default is 1
-      "numCallbackExecutors": 1
 
       # Only used when "type" is "Noop" or missing. How often to log number of bad rows discarded.
       "reportPeriod": 10 seconds

diff --git a/modules/loader/src/main/resources/application.conf b/modules/loader/src/main/resources/application.conf
@@ -26,7 +26,6 @@
         "delayThreshold": 200 milliseconds
         "maxBatchSize": 500
         "maxBatchBytes": 5000000
-        "numCallbackExecutors": 1
       }
     }
 

diff --git a/...s/loader/src/main/scala/com/snowplowanalytics/snowplow/postgres/config/LoaderConfig.scala b/...s/loader/src/main/scala/com/snowplowanalytics/snowplow/postgres/config/LoaderConfig.scala
@@ -248,8 +248,7 @@ object LoaderConfig {
                       topicId: String,
                       delayThreshold: FiniteDuration,
                       maxBatchSize: Long,
-                      maxBatchBytes: Long,
-                      numCallbackExecutors: Int) extends StreamSink
+                      maxBatchBytes: Long) extends StreamSink
 
     implicit def sinkConfigDecoder: Decoder[StreamSink] =
       deriveConfiguredDecoder[StreamSink]

diff --git a/...loader/src/main/scala/com/snowplowanalytics/snowplow/postgres/env/pubsub/PubSubSink.scala b/...loader/src/main/scala/com/snowplowanalytics/snowplow/postgres/env/pubsub/PubSubSink.scala
@@ -50,9 +50,6 @@ object PubSubSink {
       // Set the delay threshold to use for batching. After this amount of time has elapsed (counting
       // from the first element added), the elements will be wrapped up in a batch and sent.
       delayThreshold = config.delayThreshold,
-      // The number of threads used internally by permutive library to process the callback after message delivery.
-      // The callback does very little "work" so it is best to use minimum number of threads.
-      callbackExecutors = config.numCallbackExecutors,
       onFailedTerminate = err => Async[F].delay(logger.error(err)("PubSub sink termination error"))
     )
 

diff --git a/modules/loader/src/test/scala/com.snowplowanalytics.snowplow.postgres/config/CliSpec.scala b/modules/loader/src/test/scala/com.snowplowanalytics.snowplow.postgres/config/CliSpec.scala
@@ -176,8 +176,7 @@ class CliSpec extends Specification {
             "my-topic",
             200.milliseconds,
             500L,
-            5000000L,
-            1
+            5000000L
           ),
         ),
         Purpose.Enriched,