Fix Header Host Lookup

Author: joaolcorreia

modules/kafka/src/main/scala/com.snowplowanalytics.snowplow.enrich.kafka/Sink.scala

@@ -123,8 +123,10 @@ object Sink {
         .as[List[Json]]
         .toOption
         .flatMap { contexts =>
-          contexts.collectFirst {
-            case ctx if ctx.hcursor.downField("schema").as[String].toOption.exists(_.contains("http_header")) =>
+          contexts
+            .iterator
+            .filter(ctx => ctx.hcursor.downField("schema").as[String].toOption.exists(_.contains("http_header")))
+            .map { ctx =>
               val dataCursor = ctx.hcursor.downField("data")
               val nameOpt = dataCursor.downField("name").as[String].toOption
               val valueOpt = dataCursor.downField("value").as[String].toOption
@@ -133,7 +135,8 @@ object Sink {
                 case (Some("Host"), Some(hostValue)) => Some(hostValue)
                 case _ => None
               }
-          }.flatten
+            }
+            .collectFirst { case Some(host) => host }
         }
     }
   }

modules/kafka/src/test/scala/com/snowplowanalytics/snowplow/enrich/kafka/SinkSpec.scala

@@ -54,6 +54,13 @@ class SinkSpec extends Specification {
               "deviceClass": "Unknown"
             }
           },
+          {
+            "schema": "iglu:org.ietf/http_header/jsonschema/1-0-0",
+            "data": {
+              "name": "X-Forwarded-For",
+              "value": "203.0.113.42"
+            }
+          },
           {
             "schema": "iglu:org.ietf/http_header/jsonschema/1-0-0",
             "data": {