I would like to setup a test environment for Scala project. In addition, I need to start Kafka, that is running in a docker container.
Before the test is going to start, it should start first Kafka container.
I am using Scalatest and thinking about to start the Kafka container in the TestFixture, once before tests run.
The question is, which is the recommended way to start a container before running tests. I considered the Docker API, but not sure, it is the right way or not.
You can use testcontainers-scala, which is just a wrapper around testcontainers.
In your build.sbt add:
libraryDependencies += "com.dimafeng" %% "testcontainers-scala" % "0.25.0" % "test"
libraryDependencies += "org.apache.kafka" % "kafka-clients" % "2.2.0"
And then you can create spec:
import com.dimafeng.testcontainers.{ForAllTestContainer, GenericContainer}
import org.apache.kafka.clients.consumer.KafkaConsumer
import org.apache.kafka.common.serialization.{StringDeserializer, StringSerializer}
import org.scalatest.FlatSpec
import org.testcontainers.containers.Network
import org.testcontainers.utility.Base58
class KafkaSpec extends FlatSpec with ForAllTestContainer {
final val KafkaPort = 9093
override val container = GenericContainer("confluentinc/cp-kafka").configure{ c =>
c.withNetwork(Network.newNetwork())
c.withNetworkAliases("kafka-" + Base58.randomString(6))
c.withExposedPorts(KafkaPort)
c.withEnv("KAFKA_LISTENERS", "PLAINTEXT://0.0.0.0:" + KafkaPort + ",BROKER://0.0.0.0:9092")
c.withEnv("KAFKA_LISTENER_SECURITY_PROTOCOL_MAP", "BROKER:PLAINTEXT,PLAINTEXT:PLAINTEXT")
c.withEnv("KAFKA_INTER_BROKER_LISTENER_NAME", "BROKER")
c.withEnv("KAFKA_BROKER_ID", "1")
c.withEnv("KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR", "1")
c.withEnv("KAFKA_OFFSETS_TOPIC_NUM_PARTITIONS", "1")
c.withEnv("KAFKA_LOG_FLUSH_INTERVAL_MESSAGES", Long.MaxValue.toString)
c.withEnv("KAFKA_GROUP_INITIAL_REBALANCE_DELAY_MS", "0")
}
it should "do something" in {
val properties = new Properties()
properties.put("bootstrap.servers", s"${container.containerIpAddress}:$KafkaPort")
properties.put("group.id", "test")
properties.put("key.deserializer", classOf[StringDeserializer])
properties.put("value.deserializer", classOf[StringDeserializer])
properties.put("key.serializer", classOf[StringSerializer])
properties.put("value.serializer", classOf[StringSerializer])
val kafkaConsumer = new KafkaConsumer[String, String](properties)
....
}
}