Move logging of thread pool creation out of KinesisSink (close #129)

snowplow · Dec 6, 2021 · d30e0ee · d30e0ee
1 parent e11d867
commit d30e0ee
Show file tree

Hide file tree

Showing 2 changed files with 6 additions and 3 deletions.
diff --git a/...c/main/scala/com.snowplowanalytics.snowplow.collectors.scalastream/KinesisCollector.scala b/...c/main/scala/com.snowplowanalytics.snowplow.collectors.scalastream/KinesisCollector.scala
@@ -36,7 +36,7 @@ object KinesisCollector extends Collector {
         case kc: Kinesis => kc.asRight
         case _           => new IllegalArgumentException("Configured sink is not Kinesis").asLeft
       }
-      es         = new ScheduledThreadPoolExecutor(kc.threadPoolSize)
+      es         = buildExecutorService(kc)
       goodStream = collectorConf.streams.good
       badStream  = collectorConf.streams.bad
       bufferConf = collectorConf.streams.buffer
@@ -62,4 +62,9 @@ object KinesisCollector extends Collector {
       case Left(e)  => throw e
     }
   }
+
+  def buildExecutorService(kc: Kinesis): ScheduledThreadPoolExecutor = {
+    log.info("Creating thread pool of size " + kc.threadPoolSize)
+    new ScheduledThreadPoolExecutor(kc.threadPoolSize)
+  }
 }
diff --git a/.../main/scala/com.snowplowanalytics.snowplow.collectors.scalastream/sinks/KinesisSink.scala b/.../main/scala/com.snowplowanalytics.snowplow.collectors.scalastream/sinks/KinesisSink.scala
@@ -52,8 +52,6 @@ class KinesisSink private (
 ) extends Sink.Throttled(throttler) {
   import KinesisSink._
 
-  log.info("Creating thread pool of size " + kinesisConfig.threadPoolSize)
-
   maybeSqs match {
     case Some(sqs) =>
       log.info(