Test refactor

modules/kafka/src/main/scala/com/snowplowanalytics/snowplow/sinks/kafka/KafkaSink.scala

@@ -28,7 +28,6 @@ object KafkaSink {
     for {
       producer <- KafkaProducer[F].resource(producerSettings)
     } yield fromFs2Producer(config, producer)
-
   }
 
   private def fromFs2Producer[F[_]: Monad](config: KafkaSinkConfig, producer: KafkaProducer[F, String, Array[Byte]]): Sink[F] =

modules/kinesis-it/src/test/scala/com/snowplowanalytics/snowplow/sources/kinesis/KinesisSinkSpec.scala

@@ -0,0 +1,63 @@
+/*
+ * Copyright (c) 2023-present Snowplow Analytics Ltd. All rights reserved.
+ *
+ * This program is licensed to you under the Snowplow Community License Version 1.0,
+ * and you may not use this file except in compliance with the Snowplow Community License Version 1.0.
+ * You may obtain a copy of the Snowplow Community License Version 1.0 at https://docs.snowplow.io/community-license-1.0
+ */
+package com.snowplowanalytics.snowplow.sinks.kinesis
+
+import cats.effect.{IO, Resource}
+import cats.effect.testing.specs2.CatsResource
+
+import scala.concurrent.duration.{DurationInt, FiniteDuration}
+
+import org.specs2.mutable.SpecificationLike
+
+import org.testcontainers.containers.localstack.LocalStackContainer
+
+import software.amazon.awssdk.regions.providers.DefaultAwsRegionProviderChain
+import software.amazon.awssdk.regions.Region
+
+import com.snowplowanalytics.snowplow.it.kinesis._
+import com.snowplowanalytics.snowplow.sinks.Sinkable
+
+import Utils._
+
+class KinesisSinkSpec extends CatsResource[IO, (Region, LocalStackContainer, String => KinesisSinkConfig)] with SpecificationLike {
+
+  override val Timeout: FiniteDuration = 3.minutes
+
+  /** Resources which are shared across tests */
+  override val resource: Resource[IO, (Region, LocalStackContainer, String => KinesisSinkConfig)] =
+    for {
+      region <- Resource.eval(IO.blocking((new DefaultAwsRegionProviderChain).getRegion))
+      localstack <- Localstack.resource(region, KinesisSinkSpec.getClass.getSimpleName)
+    } yield (region, localstack, getKinesisSinkConfig(localstack.getEndpoint)(_))
+
+  override def is = s2"""
+  KinesisSinkSpec should
+    write to output stream $e1
+  """
+
+  def e1 = withResource { case (region, localstack, getKinesisSinkConfig) =>
+    val testStream1Name = "test-sink-stream-1"
+    val testPayload     = "test-payload"
+    val testInput       = List(Sinkable(testPayload.getBytes(), Some("myPk"), Map(("", ""))))
+
+    val kinesisClient = getKinesisClient(localstack.getEndpoint, region)
+    createAndWaitForKinesisStream(kinesisClient, testStream1Name, 1): Unit
+    val testSinkResource = KinesisSink.resource[IO](getKinesisSinkConfig(testStream1Name))
+
+    for {
+      _ <- testSinkResource.use(testSink => testSink.sink(testInput))
+      _ <- IO.sleep(3.seconds)
+      result = getDataFromKinesis(kinesisClient, testStream1Name)
+    } yield List(
+      result.events must haveSize(1),
+      result.events must beEqualTo(List(testPayload))
+    )
+  }
+}
+
+object KinesisSinkSpec {}

modules/kinesis-it/src/test/scala/com/snowplowanalytics/snowplow/sources/kinesis/KinesisSourceSpec.scala

@@ -16,45 +16,48 @@ import org.specs2.mutable.SpecificationLike
 
 import org.testcontainers.containers.localstack.LocalStackContainer
 
-import software.amazon.awssdk.services.kinesis.KinesisAsyncClient
+import software.amazon.awssdk.regions.providers.DefaultAwsRegionProviderChain
 
 import com.snowplowanalytics.snowplow.sources.EventProcessingConfig
 import com.snowplowanalytics.snowplow.sources.EventProcessingConfig.NoWindowing
+import com.snowplowanalytics.snowplow.it.kinesis._
 
 import java.time.Instant
 
 import Utils._
+import software.amazon.awssdk.regions.Region
 
-class KinesisSourceSpec
-    extends CatsResource[IO, (LocalStackContainer, KinesisAsyncClient, String => KinesisSourceConfig)]
-    with SpecificationLike {
-  import KinesisSourceSpec._
+class KinesisSourceSpec extends CatsResource[IO, (Region, LocalStackContainer, String => KinesisSourceConfig)] with SpecificationLike {
 
   override val Timeout: FiniteDuration = 3.minutes
 
   /** Resources which are shared across tests */
-  override val resource: Resource[IO, (LocalStackContainer, KinesisAsyncClient, String => KinesisSourceConfig)] =
+  override val resource: Resource[IO, (Region, LocalStackContainer, String => KinesisSourceConfig)] =
     for {
-      region <- Resource.eval(KinesisSourceConfig.getRuntimeRegion[IO])
-      localstack <- Localstack.resource(region, KINESIS_INITIALIZE_STREAMS)
-      kinesisClient <- Resource.eval(getKinesisClient(localstack.getEndpoint, region))
-    } yield (localstack, kinesisClient, getKinesisConfig(localstack.getEndpoint)(_))
+      region <- Resource.eval(IO.blocking((new DefaultAwsRegionProviderChain).getRegion))
+      localstack <- Localstack.resource(region, KinesisSourceSpec.getClass.getSimpleName)
+    } yield (region, localstack, getKinesisSourceConfig(localstack.getEndpoint)(_))
 
   override def is = s2"""
   KinesisSourceSpec should
     read from input stream $e1
   """
 
-  def e1 = withResource { case (_, kinesisClient, getKinesisConfig) =>
-    val testPayload = "test-payload"
+  def e1 = withResource { case (region, localstack, getKinesisSourceConfig) =>
+    val testPayload     = "test-payload"
+    val testStream1Name = "test-source-stream-1"
+
+    val kinesisClient = getKinesisClient(localstack.getEndpoint, region)
+    createAndWaitForKinesisStream(kinesisClient, testStream1Name, 1)
 
     for {
+
       refProcessed <- Ref[IO].of[List[ReceivedEvents]](Nil)
       t1 <- IO.realTimeInstant
       _ <- putDataToKinesis(kinesisClient, testStream1Name, testPayload)
       t2 <- IO.realTimeInstant
       processingConfig = new EventProcessingConfig(NoWindowing)
-      kinesisConfig    = getKinesisConfig(testStream1Name)
+      kinesisConfig    = getKinesisSourceConfig(testStream1Name)
       sourceAndAck <- KinesisSource.build[IO](kinesisConfig)
       stream = sourceAndAck.stream(processingConfig, testProcessor(refProcessed))
       fiber <- stream.compile.drain.start
@@ -71,8 +74,4 @@ class KinesisSourceSpec
   }
 }
 
-object KinesisSourceSpec {
-  val testStream1Name = "test-stream-1"
-  val KINESIS_INITIALIZE_STREAMS: String =
-    List(s"$testStream1Name:1").mkString(",")
-}
+object KinesisSourceSpec {}

modules/kinesis-it/src/test/scala/com/snowplowanalytics/snowplow/sources/kinesis/Localstack.scala

@@ -5,7 +5,7 @@
  * and you may not use this file except in compliance with the Snowplow Community License Version 1.0.
  * You may obtain a copy of the Snowplow Community License Version 1.0 at https://docs.snowplow.io/community-license-1.0
  */
-package com.snowplowanalytics.snowplow.sources.kinesis
+package com.snowplowanalytics.snowplow.it.kinesis
 
 import cats.effect.{IO, Resource}
 import org.slf4j.LoggerFactory
@@ -17,19 +17,21 @@ import software.amazon.awssdk.regions.Region
 
 object Localstack {
 
-  def resource(region: Region, kinesisInitializeStreams: String): Resource[IO, LocalStackContainer] =
+  def resource(
+    region: Region,
+    loggerName: String
+  ): Resource[IO, LocalStackContainer] =
     Resource.make {
       val localstack = new LocalStackContainer(DockerImageName.parse("localstack/localstack:2.2.0"))
       localstack.addEnv("AWS_DEFAULT_REGION", region.id)
-      localstack.addEnv("KINESIS_INITIALIZE_STREAMS", kinesisInitializeStreams)
       localstack.addExposedPort(4566)
       localstack.setWaitStrategy(Wait.forLogMessage(".*Ready.*", 1))
-      IO(startLocalstack(localstack))
+      IO(startLocalstack(localstack, loggerName))
     }(ls => IO.blocking(ls.stop()))
 
-  private def startLocalstack(localstack: LocalStackContainer): LocalStackContainer = {
+  private def startLocalstack(localstack: LocalStackContainer, loggerName: String): LocalStackContainer = {
     localstack.start()
-    val logger = LoggerFactory.getLogger(KinesisSourceSpec.getClass.getSimpleName)
+    val logger = LoggerFactory.getLogger(loggerName)
     val logs   = new Slf4jLogConsumer(logger)
     localstack.followOutput(logs)
     localstack

modules/kinesis-it/src/test/scala/com/snowplowanalytics/snowplow/sources/kinesis/Utils.scala

@@ -5,28 +5,70 @@
  * and you may not use this file except in compliance with the Snowplow Community License Version 1.0.
  * You may obtain a copy of the Snowplow Community License Version 1.0 at https://docs.snowplow.io/community-license-1.0
  */
-package com.snowplowanalytics.snowplow.sources.kinesis
+package com.snowplowanalytics.snowplow.it.kinesis
 
 import cats.effect.{IO, Ref}
 
+import scala.concurrent.duration.FiniteDuration
+import scala.jdk.CollectionConverters._
+
 import eu.timepit.refined.types.numeric.PosInt
 
 import software.amazon.awssdk.core.SdkBytes
 import software.amazon.awssdk.regions.Region
 import software.amazon.awssdk.services.kinesis.KinesisAsyncClient
-import software.amazon.awssdk.services.kinesis.model.{PutRecordRequest, PutRecordResponse}
+import software.amazon.awssdk.services.kinesis.model.{
+  CreateStreamRequest,
+  GetRecordsRequest,
+  GetShardIteratorRequest,
+  PutRecordRequest,
+  PutRecordResponse
+}
+
+import com.snowplowanalytics.snowplow.sources.{EventProcessor, TokenedEvents}
+import com.snowplowanalytics.snowplow.sources.kinesis.KinesisSourceConfig
+import com.snowplowanalytics.snowplow.sinks.kinesis.{BackoffPolicy, KinesisSinkConfig}
 
 import java.net.URI
 import java.nio.charset.StandardCharsets
 import java.util.UUID
 import java.time.Instant
+import java.util.concurrent.TimeUnit
 
-import com.snowplowanalytics.snowplow.sources.{EventProcessor, TokenedEvents}
+import software.amazon.awssdk.services.kinesis.model.DescribeStreamRequest
 
 object Utils {
 
   case class ReceivedEvents(events: List[String], tstamp: Option[Instant])
 
+  def createAndWaitForKinesisStream(
+    client: KinesisAsyncClient,
+    streamName: String,
+    shardCount: Int
+  ) = {
+    val createStreamReq = CreateStreamRequest
+      .builder()
+      .streamName(streamName)
+      .shardCount(shardCount)
+      .build()
+
+    val resp = client.createStream(createStreamReq).get
+
+    // Block till it's active
+    while (
+      resp.sdkHttpResponse().isSuccessful() && (client
+        .describeStream(DescribeStreamRequest.builder().streamName(streamName).build())
+        .get
+        .streamDescription
+        .streamStatusAsString != "ACTIVE")
+    ) {
+      Thread.sleep(500)
+      println(
+        client.describeStream(DescribeStreamRequest.builder().streamName(streamName).build()).get.streamDescription.streamStatusAsString
+      )
+    }
+  }
+
   def putDataToKinesis(
     client: KinesisAsyncClient,
     streamName: String,
@@ -42,7 +84,51 @@ object Utils {
     IO.blocking(client.putRecord(record).get())
   }
 
-  def getKinesisConfig(endpoint: URI)(streamName: String): KinesisSourceConfig = KinesisSourceConfig(
+  /**
+   * getDataFromKinesis gets the last 1000 records from kinesis, stringifies the datta it found, and
+   * returns a ReceivedEvents It can be called at the end of simple tests to return data from a
+   * Kinesis stream.
+   *
+   * If required in future, where more data is used we might amend it to poll the stream for data
+   * and return everything it finds after a period without any data.
+   */
+  def getDataFromKinesis(
+    client: KinesisAsyncClient,
+    streamName: String
+  ): ReceivedEvents = {
+
+    val descStreamResp = client.describeStream(DescribeStreamRequest.builder().streamName(streamName).build()).get
+
+    // We're assuming only one shard here.
+    // Any future test with multiple shards requires us to create one iterator per shard
+    val shIterRequest = GetShardIteratorRequest
+      .builder()
+      .streamName(streamName)
+      .shardIteratorType("TRIM_HORIZON")
+      .shardId(descStreamResp.streamDescription.shards.get(0).shardId)
+      .build();
+
+    val shIter = client.getShardIterator(shIterRequest).get.shardIterator
+
+    val request = GetRecordsRequest
+      .builder()
+      .streamARN(descStreamResp.streamDescription().streamARN())
+      .shardIterator(shIter)
+      .build()
+
+    val out =
+      client
+        .getRecords(request)
+        .get()
+        .records()
+        .asScala
+        .toList
+        .map(record => new String(record.data.asByteArray()))
+
+    ReceivedEvents(out, None)
+  }
+
+  def getKinesisSourceConfig(endpoint: URI)(streamName: String): KinesisSourceConfig = KinesisSourceConfig(
     UUID.randomUUID().toString,
     streamName,
     KinesisSourceConfig.InitialPosition.TrimHorizon,
@@ -53,6 +139,14 @@ object Utils {
     Some(endpoint)
   )
 
+  def getKinesisSinkConfig(endpoint: URI)(streamName: String): KinesisSinkConfig = KinesisSinkConfig(
+    streamName,
+    BackoffPolicy(FiniteDuration(1, TimeUnit.SECONDS), FiniteDuration(1, TimeUnit.SECONDS), None),
+    1000,
+    1000000,
+    Some(endpoint)
+  )
+
   def testProcessor(ref: Ref[IO, List[ReceivedEvents]]): EventProcessor[IO] =
     _.evalMap { case TokenedEvents(events, token, tstamp) =>
       val parsed = events.map(byteBuffer => StandardCharsets.UTF_8.decode(byteBuffer).toString)
@@ -61,12 +155,11 @@ object Utils {
       } yield token
     }
 
-  def getKinesisClient(endpoint: URI, region: Region): IO[KinesisAsyncClient] =
-    IO(
-      KinesisAsyncClient
-        .builder()
-        .endpointOverride(endpoint)
-        .region(region)
-        .build()
-    )
+  def getKinesisClient(endpoint: URI, region: Region): KinesisAsyncClient =
+    KinesisAsyncClient
+      .builder()
+      .endpointOverride(endpoint)
+      .region(region)
+      .build()
+
 }