Update base Kafka test code

Author: mdedetrich

build.sbt

@@ -22,6 +22,7 @@ val scalaTestScalaCheckVersion = "3.2.9.0"
 val akkaStreamsJson            = "0.8.0"
 val diffxVersion               = "0.5.6"
 val testContainersVersion      = "0.39.8"
+val testContainersJavaVersion  = "1.16.2"
 
 val flagsFor12 = Seq(
   "-Xlint:_",
@@ -88,7 +89,9 @@ lazy val core = project
       "com.softwaremill.diffx"     %% "diffx-scalatest"                % diffxVersion               % Test,
       "com.typesafe.akka"          %% "akka-stream-testkit"            % akkaVersion                % Test,
       "com.typesafe.akka"          %% "akka-http-testkit"              % akkaHttpVersion            % Test,
-      "com.dimafeng"               %% "testcontainers-scala-scalatest" % testContainersVersion      % Test
+      "com.dimafeng"               %% "testcontainers-scala-scalatest" % testContainersVersion      % Test,
+      "com.dimafeng"               %% "testcontainers-scala-kafka"     % testContainersVersion      % Test,
+      "org.testcontainers"          % "kafka"                          % testContainersJavaVersion  % Test
     )
   )
 

core/src/main/scala/io/aiven/guardian/kafka/KafkaClient.scala

@@ -23,12 +23,16 @@ import java.util.Base64
 /** A Kafka Client that uses Alpakka Kafka Consumer under the hood to create a stream of events from a Kafka cluster. To
   * configure the Alpakka Kafka Consumer use the standard typesafe configuration i.e. akka.kafka.consumer (note that the
   * `keySerializer` and `valueSerializer` are hardcoded so you cannot override this).
+  * @param configure
+  *   A way to configure the underlying Kafka consumer settings
   * @param system
   *   A classic `ActorSystem`
   * @param kafkaClusterConfig
   *   Additional cluster configuration that is needed
   */
-class KafkaClient()(implicit system: ActorSystem, kafkaClusterConfig: KafkaCluster)
+class KafkaClient(
+    configure: Option[ConsumerSettings[Array[Byte], Array[Byte]] => ConsumerSettings[Array[Byte], Array[Byte]]] = None
+)(implicit system: ActorSystem, kafkaClusterConfig: KafkaCluster)
     extends KafkaClientInterface
     with StrictLogging {
   override type CursorContext = Committable
@@ -37,8 +41,10 @@ class KafkaClient()(implicit system: ActorSystem, kafkaClusterConfig: KafkaClust
   if (kafkaClusterConfig.topics.isEmpty)
     logger.warn("Kafka Cluster configuration has no topics set")
 
-  private[kafka] val consumerSettings =
-    ConsumerSettings(system, new ByteArrayDeserializer, new ByteArrayDeserializer)
+  private[kafka] val consumerSettings = {
+    val base = ConsumerSettings(system, new ByteArrayDeserializer, new ByteArrayDeserializer)
+    configure.fold(base)(block => block(base))
+  }
 
   private[kafka] val subscriptions = Subscriptions.topics(kafkaClusterConfig.topics)
 

core/src/test/scala/io/aiven/guardian/kafka/Generators.scala

@@ -109,12 +109,18 @@ object Generators {
                                   max: Int = 100,
                                   padTimestampsMillis: Int = 10
   ): Gen[KafkaDataWithTimePeriod] = for {
-    topic   <- Gen.alphaStr
+    topic   <- kafkaTopic
     records <- Generators.kafkaReducedConsumerRecordsGen(topic, min, max, padTimestampsMillis)
     head = records.head
     last = records.last
 
     duration <- Gen.choose[Long](head.timestamp, last.timestamp - 1).map(millis => FiniteDuration(millis, MILLISECONDS))
   } yield KafkaDataWithTimePeriod(records, duration)
 
+  /** Generator for a valid Kafka topic that can be used in actual Kafka clusters
+    */
+  lazy val kafkaTopic: Gen[String] = for {
+    size  <- Gen.choose(1, 249)
+    topic <- Gen.listOfN(size, Gen.oneOf(Gen.alphaChar, Gen.numChar, Gen.const('-'), Gen.const('.'), Gen.const('_')))
+  } yield topic.mkString
 }

core/src/test/scala/io/aiven/guardian/kafka/KafkaClusterTest.scala

@@ -0,0 +1,10 @@
+package io.aiven.guardian.kafka
+
+import akka.testkit.TestKitBase
+import com.dimafeng.testcontainers.ForAllTestContainer
+import com.dimafeng.testcontainers.KafkaContainer
+import org.scalatest.Suite
+
+trait KafkaClusterTest extends ForAllTestContainer with TestKitBase { this: Suite =>
+  override lazy val container: KafkaContainer = new KafkaContainer()
+}