-
Notifications
You must be signed in to change notification settings - Fork 31
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
- Loading branch information
Showing
12 changed files
with
465 additions
and
92 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,14 @@ | ||
| collector { | ||
| interface = "0.0.0.0" | ||
| port = ${PORT} | ||
|
|
||
| streams { | ||
| good = ${TOPIC_GOOD} | ||
| bad = ${TOPIC_BAD} | ||
|
|
||
| sink { | ||
| brokers = ${BROKER} | ||
| maxBytes = ${MAX_BYTES} | ||
| } | ||
| } | ||
| } |
119 changes: 119 additions & 0 deletions
119
.../it/scala/com/snowplowanalytics/snowplow/collectors/scalastream/it/kafka/Containers.scala
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,119 @@ | ||
| package com.snowplowanalytics.snowplow.collectors.scalastream.it.kafka | ||
|
|
||
| import cats.effect._ | ||
| import com.dimafeng.testcontainers.{FixedHostPortGenericContainer, GenericContainer} | ||
| import com.snowplowanalytics.snowplow.collectors.scalastream.BuildInfo | ||
| import com.snowplowanalytics.snowplow.collectors.scalastream.it.CollectorContainer | ||
| import com.snowplowanalytics.snowplow.collectors.scalastream.it.utils._ | ||
| import org.testcontainers.containers.wait.strategy.Wait | ||
| import org.testcontainers.containers.{BindMode, Network, GenericContainer => JGenericContainer} | ||
|
|
||
| object Containers { | ||
|
|
||
| val zookeeperContainerName = "zookeeper" | ||
| val zookeeperPort = 2181 | ||
| val brokerContainerName = "broker" | ||
| val brokerExternalPort = 9092 | ||
| val brokerInternalPort = 29092 | ||
|
|
||
| def createContainers( | ||
| goodTopic: String, | ||
| badTopic: String, | ||
| maxBytes: Int | ||
| ): Resource[IO, CollectorContainer] = | ||
| for { | ||
| network <- network() | ||
| _ <- zookeeper(network) | ||
| _ <- kafka(network) | ||
| c <- collectorKafka(network, goodTopic, badTopic, maxBytes) | ||
| } yield c | ||
|
|
||
| private def network(): Resource[IO, Network] = | ||
| Resource.make(IO(Network.newNetwork()))(n => IO(n.close())) | ||
|
|
||
| private def kafka( | ||
| network: Network | ||
| ): Resource[IO, JGenericContainer[_]] = | ||
| Resource.make( | ||
| IO { | ||
| val container = FixedHostPortGenericContainer( | ||
| imageName = "confluentinc/cp-kafka:7.0.1", | ||
| env = Map( | ||
| "KAFKA_BROKER_ID" -> "1", | ||
| "KAFKA_ZOOKEEPER_CONNECT" -> s"$zookeeperContainerName:$zookeeperPort", | ||
| "KAFKA_LISTENER_SECURITY_PROTOCOL_MAP" -> "PLAINTEXT:PLAINTEXT,PLAINTEXT_INTERNAL:PLAINTEXT", | ||
| "KAFKA_ADVERTISED_LISTENERS" -> s"PLAINTEXT://localhost:$brokerExternalPort,PLAINTEXT_INTERNAL://$brokerContainerName:$brokerInternalPort", | ||
| "KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR" -> "1", | ||
| "KAFKA_TRANSACTION_STATE_LOG_MIN_ISR" -> "1", | ||
| "KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR" -> "1" | ||
| ), | ||
| exposedPorts = List(brokerExternalPort, brokerInternalPort), | ||
| exposedHostPort = brokerExternalPort, | ||
| exposedContainerPort = brokerExternalPort | ||
| ) | ||
| container.container.withNetwork(network) | ||
| container.container.withNetworkAliases(brokerContainerName) | ||
| container.start() | ||
| container.container | ||
| } | ||
| )(e => IO(e.stop())) | ||
|
|
||
| private def zookeeper( | ||
| network: Network, | ||
| ): Resource[IO, JGenericContainer[_]] = | ||
| Resource.make( | ||
| IO { | ||
| val container = GenericContainer( | ||
| dockerImage = "confluentinc/cp-zookeeper:7.0.1", | ||
| env = Map( | ||
| "ZOOKEEPER_CLIENT_PORT" -> zookeeperPort.toString, | ||
| "ZOOKEEPER_TICK_TIME" -> "2000" | ||
| ), | ||
| exposedPorts = List(zookeeperPort) | ||
| ) | ||
| container.container.withNetwork(network) | ||
| container.container.withNetworkAliases(zookeeperContainerName) | ||
| container.start() | ||
| container.container | ||
| } | ||
| )(e => IO(e.stop())) | ||
|
|
||
| def collectorKafka( | ||
| network: Network, | ||
| goodTopic: String, | ||
| badTopic: String, | ||
| maxBytes: Int | ||
| ): Resource[IO, CollectorContainer] = { | ||
| Resource.make( | ||
| IO { | ||
| val collectorPort = 8080 | ||
| val container = GenericContainer( | ||
| dockerImage = BuildInfo.dockerAlias, | ||
| env = Map( | ||
| "PORT" -> collectorPort.toString, | ||
| "BROKER" -> s"$brokerContainerName:$brokerInternalPort", | ||
| "TOPIC_GOOD" -> goodTopic, | ||
| "TOPIC_BAD" -> badTopic, | ||
| "MAX_BYTES" -> maxBytes.toString | ||
| ), | ||
| exposedPorts = Seq(collectorPort), | ||
| fileSystemBind = Seq( | ||
| GenericContainer.FileSystemBind( | ||
| "kafka/src/it/resources/collector.hocon", | ||
| "/snowplow/config/collector.hocon", | ||
| BindMode.READ_ONLY | ||
| ) | ||
| ), | ||
| command = Seq( | ||
| "--config", | ||
| "/snowplow/config/collector.hocon" | ||
| ), | ||
| waitStrategy = Wait.forLogMessage(s".*Service bound to address.*", 1) | ||
| ) | ||
| container.container.withNetwork(network) | ||
| val c = startContainerWithLogs(container.container, "collector") | ||
| CollectorContainer(c, c.getHost, c.getMappedPort(collectorPort)) | ||
| } | ||
| )(c => IO(c.container.stop())) | ||
| } | ||
| } |
55 changes: 55 additions & 0 deletions
55
...a/com/snowplowanalytics/snowplow/collectors/scalastream/it/kafka/KafkaCollectorSpec.scala
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,55 @@ | ||
| package com.snowplowanalytics.snowplow.collectors.scalastream.it.kafka | ||
|
|
||
| import scala.concurrent.duration._ | ||
|
|
||
| import cats.effect.IO | ||
| import cats.effect.testing.specs2.CatsEffect | ||
|
|
||
| import com.snowplowanalytics.snowplow.collectors.scalastream.it.EventGenerator | ||
| import com.snowplowanalytics.snowplow.collectors.scalastream.it.utils._ | ||
|
|
||
| import org.specs2.mutable.Specification | ||
|
|
||
| class KafkaCollectorSpec extends Specification with CatsEffect { | ||
|
|
||
| override protected val Timeout = 5.minutes | ||
|
|
||
| val maxBytes = 10000 | ||
|
|
||
| "emit the correct number of collector payloads and bad rows" in { | ||
| val testName = "count" | ||
| val nbGood = 1000 | ||
| val nbBad = 10 | ||
| val goodTopic = "test-raw" | ||
| val badTopic = "test-bad" | ||
|
|
||
| Containers.createContainers( | ||
| goodTopic = goodTopic, | ||
| badTopic = badTopic, | ||
| maxBytes = maxBytes | ||
| ).use { collector => | ||
| for { | ||
| _ <- log(testName, "Sending data") | ||
| _ <- EventGenerator.sendEvents( | ||
| collector.host, | ||
| collector.port, | ||
| nbGood, | ||
| nbBad, | ||
| maxBytes | ||
| ) | ||
| _ <- log(testName, "Data sent. Waiting for collector to work") | ||
| _ <- IO.sleep(30.second) | ||
| _ <- log(testName, "Consuming collector's output") | ||
| collectorOutput <- KafkaUtils.readOutput( | ||
| brokerAddr = s"localhost:${Containers.brokerExternalPort}", | ||
| goodTopic = goodTopic, | ||
| badTopic = badTopic | ||
| ) | ||
| } yield { | ||
| collectorOutput.good.size must beEqualTo(nbGood) | ||
| collectorOutput.bad.size must beEqualTo(nbBad) | ||
| } | ||
| } | ||
| } | ||
|
|
||
| } |
46 changes: 46 additions & 0 deletions
46
.../it/scala/com/snowplowanalytics/snowplow/collectors/scalastream/it/kafka/KafkaUtils.scala
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,46 @@ | ||
| package com.snowplowanalytics.snowplow.collectors.scalastream.it.kafka | ||
|
|
||
| import cats.effect._ | ||
| import org.apache.kafka.clients.consumer._ | ||
| import java.util.Properties | ||
| import java.time.Duration | ||
| import scala.collection.JavaConverters._ | ||
| import com.snowplowanalytics.snowplow.collectors.scalastream.it.utils._ | ||
| import com.snowplowanalytics.snowplow.collectors.scalastream.it.CollectorOutput | ||
|
|
||
| object KafkaUtils { | ||
|
|
||
| def readOutput( | ||
| brokerAddr: String, | ||
| goodTopic: String, | ||
| badTopic: String | ||
| ): IO[CollectorOutput] = { | ||
| createConsumer(brokerAddr).use { kafkaConsumer => | ||
| IO { | ||
| kafkaConsumer.subscribe(List(goodTopic, badTopic).asJava) | ||
| val records = kafkaConsumer.poll(Duration.ofSeconds(20)) | ||
| val extract = (r: ConsumerRecords[String, Array[Byte]], topicName: String) => | ||
| r.records(topicName).asScala.toList.map(_.value()) | ||
| val goodCount = extract(records, goodTopic).map(parseCollectorPayload) | ||
| val badCount = extract(records, badTopic).map(parseBadRow) | ||
| CollectorOutput(goodCount, badCount) | ||
| } | ||
| } | ||
| } | ||
|
|
||
| private def createConsumer(brokerAddr: String): Resource[IO, KafkaConsumer[String, Array[Byte]]] = { | ||
| val acquire = IO { | ||
| val props = new Properties() | ||
| props.setProperty("bootstrap.servers", brokerAddr) | ||
| props.setProperty("group.id", "it-collector") | ||
| props.setProperty("auto.offset.reset", "earliest") | ||
| props.setProperty("max.poll.records", "2000") | ||
| props.setProperty("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer") | ||
| props.setProperty("value.deserializer", "org.apache.kafka.common.serialization.ByteArrayDeserializer") | ||
| new KafkaConsumer[String, Array[Byte]](props) | ||
| } | ||
| val release = (p: KafkaConsumer[String, Array[Byte]]) => IO(p.close()) | ||
| Resource.make(acquire)(release) | ||
| } | ||
|
|
||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.