address feedback

snowplow-incubator · Sep 16, 2024 · 2e31a12 · 2e31a12
1 parent 76680f6
commit 2e31a12
Show file tree

Hide file tree

Showing 6 changed files with 12 additions and 9 deletions.
diff --git a/modules/core/src/main/resources/reference.conf b/modules/core/src/main/resources/reference.conf
@@ -45,7 +45,7 @@
   "exitOnMissingIgluSchema": true
 
   "http": {
-    "maxConnectionsPerServer": 4
+    "client": ${snowplow.defaults.http.client}
   }
 
   "monitoring": {

diff --git a/modules/core/src/main/scala/com.snowplowanalytics.snowplow.bigquery/Config.scala b/modules/core/src/main/scala/com.snowplowanalytics.snowplow.bigquery/Config.scala
@@ -33,7 +33,7 @@ case class Config[+Source, +Sink](
   skipSchemas: List[SchemaCriterion],
   legacyColumns: List[SchemaCriterion],
   exitOnMissingIgluSchema: Boolean,
-  http: HttpClient.Config
+  http: Config.Http
 )
 
 object Config {
@@ -92,6 +92,8 @@ object Config {
     tooManyColumns: TooManyColumnsRetries
   )
 
+  case class Http(client: HttpClient.Config)
+
   implicit def decoder[Source: Decoder, Sink: Decoder]: Decoder[Config[Source, Sink]] = {
     implicit val configuration = Configuration.default.withDiscriminator("type")
     implicit val sinkWithMaxSize = for {
@@ -115,6 +117,7 @@ object Config {
     implicit val alterTableRetries  = deriveConfiguredDecoder[AlterTableWaitRetries]
     implicit val tooManyColsRetries = deriveConfiguredDecoder[TooManyColumnsRetries]
     implicit val retriesDecoder     = deriveConfiguredDecoder[Retries]
+    implicit val httpDecoder        = deriveConfiguredDecoder[Http]
 
     // TODO add bigquery docs
     implicit val licenseDecoder =

diff --git a/modules/core/src/main/scala/com.snowplowanalytics.snowplow.bigquery/Environment.scala b/modules/core/src/main/scala/com.snowplowanalytics.snowplow.bigquery/Environment.scala
@@ -53,7 +53,7 @@ object Environment {
       sourceReporter = sourceAndAck.isHealthy(config.main.monitoring.healthProbe.unhealthyLatency).map(_.showIfUnhealthy)
       appHealth <- Resource.eval(AppHealth.init[F, Alert, RuntimeService](List(sourceReporter)))
       resolver <- mkResolver[F](config.iglu)
-      httpClient <- HttpClient.resource[F](HttpClient.Config(config.main.http.maxConnectionsPerServer))
+      httpClient <- HttpClient.resource[F](config.main.http.client)
       _ <- HealthProbe.resource(config.main.monitoring.healthProbe.port, appHealth)
       _ <- Webhook.resource(config.main.monitoring.webhook, appInfo, httpClient, appHealth)
       badSink <-

diff --git a/modules/kafka/src/test/scala/com/snowplowanalytics/snowplow/bigquery/KafkaConfigSpec.scala b/modules/kafka/src/test/scala/com/snowplowanalytics/snowplow/bigquery/KafkaConfigSpec.scala
@@ -127,7 +127,7 @@ object KafkaConfigSpec {
     skipSchemas             = List.empty,
     legacyColumns           = List.empty,
     exitOnMissingIgluSchema = true,
-    http                    = HttpClient.Config(4)
+    http                    = Config.Http(HttpClient.Config(4))
   )
 
   private val extendedConfig = Config[KafkaSourceConfig, KafkaSinkConfig](
@@ -221,6 +221,6 @@ object KafkaConfigSpec {
       SchemaCriterion.parse("iglu:com.acme/legacy/jsonschema/2-*-*").get
     ),
     exitOnMissingIgluSchema = true,
-    http                    = HttpClient.Config(4)
+    http                    = Config.Http(HttpClient.Config(4))
   )
 }
diff --git a/...es/kinesis/src/test/scala/com/snowplowanalytics/snowplow/bigquery/KinesisConfigSpec.scala b/...es/kinesis/src/test/scala/com/snowplowanalytics/snowplow/bigquery/KinesisConfigSpec.scala
@@ -123,7 +123,7 @@ object KinesisConfigSpec {
     skipSchemas             = List.empty,
     legacyColumns           = List.empty,
     exitOnMissingIgluSchema = true,
-    http                    = HttpClient.Config(4)
+    http                    = Config.Http(HttpClient.Config(4))
   )
 
   // workerIdentifer coming from "HOSTNAME" env variable set in BuildSettings
@@ -213,6 +213,6 @@ object KinesisConfigSpec {
       SchemaCriterion.parse("iglu:com.acme/legacy/jsonschema/2-*-*").get
     ),
     exitOnMissingIgluSchema = true,
-    http                    = HttpClient.Config(4)
+    http                    = Config.Http(HttpClient.Config(4))
   )
 }
diff --git a/modules/pubsub/src/test/scala/com/snowplowanalytics/snowplow/bigquery/PubsubConfigSpec.scala b/modules/pubsub/src/test/scala/com/snowplowanalytics/snowplow/bigquery/PubsubConfigSpec.scala
@@ -122,7 +122,7 @@ object PubsubConfigSpec {
     skipSchemas             = List.empty,
     legacyColumns           = List.empty,
     exitOnMissingIgluSchema = true,
-    http                    = HttpClient.Config(4)
+    http                    = Config.Http(HttpClient.Config(4))
   )
 
   private val extendedConfig = Config[PubsubSourceConfig, PubsubSinkConfig](
@@ -209,6 +209,6 @@ object PubsubConfigSpec {
       SchemaCriterion.parse("iglu:com.acme/legacy/jsonschema/2-*-*").get
     ),
     exitOnMissingIgluSchema = true,
-    http                    = HttpClient.Config(4)
+    http                    = Config.Http(HttpClient.Config(4))
   )
 }