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

snowplow · Jan 11, 2022 · 3702ad0 · 3702ad0
1 parent 5c6f5cf
commit 3702ad0
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
@@ -34,7 +34,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
@@ -56,4 +56,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
@@ -51,8 +51,6 @@ class KinesisSink private (
 ) extends Sink {
   import KinesisSink._
 
-  log.info("Creating thread pool of size " + kinesisConfig.threadPoolSize)
-
   maybeSqs match {
     case Some(sqs) =>
       log.info(