Use Testcontainers and remove sbt-docker-compose

.github/autolabeler.yml

@@ -4,4 +4,4 @@ benchmarks: ["/benchmarks"]
 core: ["/core"]
 documentation: ["/docs"]
 testkit: ["/testkit"]
-tests: ["/tests", "docker-compose.yml"]
+tests: ["/tests"]

.travis.yml

@@ -63,10 +63,14 @@ jobs:
       - _JAVA_OPTIONS="-XX:+UnlockExperimentalVMOptions -XX:+UseJVMCICompiler"
       - CMD="++2.13.1 test"
       name: "Run tests with Scala 2.13 and AdoptOpenJDK 11"
-
     - stage: integration
-      env: CMD="dockerComposeTestAll"
-    - env: CMD="benchmarks/It/compile"
+      env: CMD="tests/it:test"
+      name: "Run integration tests"
+
+    - env: CMD="benchmarks/it:test"
+      # TODO: Replace with compile CMD before merge
+      # - env: CMD="benchmarks/it:compile"
+      name: "Compile benchmark tests"
 
     - stage: whitesource
       env: CMD=";whitesourceCheckPolicies ;whitesourceUpdate"

benchmarks/src/it/scala/akka/kafka/benchmarks/Benchmarks.scala

@@ -1,46 +1,46 @@
-/*
+  /*
  * Copyright (C) 2014 - 2016 Softwaremill <http://softwaremill.com>
  * Copyright (C) 2016 - 2019 Lightbend Inc. <http://www.lightbend.com>
  */
 package akka.kafka.benchmarks
 
+import akka.kafka.benchmarks.BenchmarksBase._
 import akka.kafka.benchmarks.PerfFixtureHelpers.FilledTopic
 import akka.kafka.benchmarks.Timed.runPerfTest
 import akka.kafka.benchmarks.app.RunTestCommand
-import akka.kafka.testkit.scaladsl.ScalatestKafkaSpec
-import org.scalatest.FlatSpecLike
-
-import BenchmarksBase._
+import akka.kafka.testkit.internal.TestcontainersKafka.TestcontainersKafkaSettings
+import akka.kafka.testkit.scaladsl.TestcontainersKafkaLike
 
 object BenchmarksBase {
   // Message count multiplier to adapt for shorter local testing
   val factor = 1000
 
-  val topic_50_100 = FilledTopic(50 * factor, 100)
+  // Default settings for Kafka testcontainers cluster
+  val settings = TestcontainersKafkaSettings()
 
-  val topic_100_100 = FilledTopic(100 * factor, 100)
-  val topic_100_5000 = FilledTopic(100 * factor, 5000)
+  import settings._
 
-  val topic_1000_100 = FilledTopic(1000 * factor, 100)
-  val topic_1000_5000 = FilledTopic(1000 * factor, 5 * 1000)
-  val topic_1000_5000_8 = FilledTopic(msgCount = 1000 * factor, msgSize = 5 * 1000, numberOfPartitions = 8)
+  val topic_50_100 = FilledTopic(50 * factor, 100, replicationFactor = numBrokers)
 
-  val topic_2000_100 = FilledTopic(2000 * factor, 100)
-  val topic_2000_500 = FilledTopic(2000 * factor, 500)
-  val topic_2000_5000 = FilledTopic(2000 * factor, 5000)
-  val topic_2000_5000_8 = FilledTopic(2000 * factor, 5000, numberOfPartitions = 8)
+  val topic_100_100 = FilledTopic(100 * factor, 100, replicationFactor = numBrokers)
+  val topic_100_5000 = FilledTopic(100 * factor, 5000, replicationFactor = numBrokers)
 
-}
+  val topic_1000_100 = FilledTopic(1000 * factor, 100, replicationFactor = numBrokers)
+  val topic_1000_5000 = FilledTopic(1000 * factor, 5 * 1000, replicationFactor = numBrokers)
+  val topic_1000_5000_8 = FilledTopic(msgCount = 1000 * factor, msgSize = 5 * 1000, numberOfPartitions = 8, replicationFactor = numBrokers)
 
-abstract class BenchmarksBase() extends ScalatestKafkaSpec(0) with FlatSpecLike {
+  val topic_2000_100 = FilledTopic(2000 * factor, 100, replicationFactor = numBrokers)
+  val topic_2000_500 = FilledTopic(2000 * factor, 500, replicationFactor = numBrokers)
+  val topic_2000_5000 = FilledTopic(2000 * factor, 5000, replicationFactor = numBrokers)
+  val topic_2000_5000_8 = FilledTopic(2000 * factor, 5000, numberOfPartitions = 8, replicationFactor = numBrokers)
+}
 
-  override def bootstrapServers: String =
-    (1 to BuildInfo.kafkaScale).map(i => sys.props(s"kafka_${i}_9094")).mkString(",")
+abstract class BenchmarksBase() extends SpecBase with TestcontainersKafkaLike {
 
   override def setUp(): Unit = {
     super.setUp()
     waitUntilCluster() {
-      _.nodes().get().size == BuildInfo.kafkaScale
+      _.nodes().get().size == BenchmarksBase.settings.numBrokers
     }
   }
 }

benchmarks/src/it/scala/akka/kafka/benchmarks/SpecBase.scala

@@ -0,0 +1,15 @@
+package akka.kafka.benchmarks
+
+import akka.kafka.testkit.scaladsl.ScalatestKafkaSpec
+import org.scalatest.concurrent.{Eventually, ScalaFutures}
+import org.scalatest.{FlatSpecLike, Matchers}
+
+abstract class SpecBase(kafkaPort: Int)
+  extends ScalatestKafkaSpec(kafkaPort)
+    with FlatSpecLike
+    with Matchers
+    with ScalaFutures
+    with Eventually {
+
+  protected def this() = this(kafkaPort = -1)
+}

benchmarks/src/it/scala/akka/kafka/benchmarks/TestTester.scala

@@ -0,0 +1,18 @@
+package akka.kafka.benchmarks
+
+// TODO: deleteme
+class TestTester1 extends BenchmarksBase() {
+
+  it should "test" in {
+    log.info("TestTester1 test")
+    succeed
+  }
+}
+
+class TestTester2 extends BenchmarksBase() {
+
+  it should "test" in {
+    log.info("TestTester2 test")
+    succeed
+  }
+}

benchmarks/src/main/scala/akka/kafka/benchmarks/PerfFixtureHelpers.scala

@@ -27,9 +27,9 @@ object PerfFixtureHelpers {
       msgCount: Int,
       msgSize: Int,
       numberOfPartitions: Int = 1,
+      replicationFactor: Int = 1,
       topic: String = randomId()
   ) {
-    def replicationFactor = BuildInfo.kafkaScale
     def freshTopic: FilledTopic = copy(topic = randomId())
   }
 }

benchmarks/src/main/scala/akka/kafka/benchmarks/app/RunTestCommand.scala

@@ -12,5 +12,6 @@ case class RunTestCommand(testName: String, kafkaHost: String, filledTopic: Fill
   val msgCount = filledTopic.msgCount
   val msgSize = filledTopic.msgSize
   val numberOfPartitions = filledTopic.numberOfPartitions
+  val replicationFactor = filledTopic.replicationFactor
 
 }

build.sbt

@@ -32,8 +32,6 @@ val silencer = {
 // that depends on the same Kafka version, as is defined above
 val embeddedKafkaSchemaRegistry = "5.1.2"
 
-val kafkaScale = settingKey[Int]("Number of kafka docker containers")
-
 resolvers in ThisBuild ++= Seq(
   // for Embedded Kafka
   Resolver.bintrayRepo("manub", "maven"),
@@ -149,7 +147,6 @@ lazy val `alpakka-kafka` =
     .settings(commonSettings)
     .settings(
       skip in publish := true,
-      dockerComposeIgnore := true,
       ScalaUnidoc / unidoc / unidocProjectFilter := inProjects(core, testkit),
       onLoadMessage :=
         """
@@ -182,10 +179,10 @@ lazy val `alpakka-kafka` =
             |  test
             |    runs all the tests
             |
-            |  tests/dockerComposeTest it:test
-            |    run integration test backed by Docker containers
+            |  tests/it:test
+            |    run integration tests backed by Docker containers
             |
-            |  benchmarks/dockerComposeTest it:testOnly *.AlpakkaKafkaPlainConsumer
+            |  benchmarks/it:testOnly *.AlpakkaKafkaPlainConsumer
             |    run a single benchmark backed by Docker containers
           """.stripMargin
     )
@@ -222,6 +219,7 @@ lazy val testkit = project
     libraryDependencies ++= Seq(
         "com.typesafe.akka" %% "akka-stream-testkit" % akkaVersion,
         "org.testcontainers" % "kafka" % testcontainersVersion % Provided,
+        "com.dimafeng" %% "testcontainers-scala" % testcontainersScalaVersion % Provided,
         "org.scalatest" %% "scalatest" % scalatestVersion % Provided,
         "junit" % "junit" % "4.12" % Provided,
         "org.junit.jupiter" % "junit-jupiter-api" % JupiterKeys.junitJupiterVersion.value % Provided
@@ -234,7 +232,7 @@ lazy val testkit = project
             "org.apache.commons" % "commons-compress" % "1.19", // embedded Kafka pulls in Avro which pulls in commons-compress 1.8.1
             "io.github.embeddedkafka" %% "embedded-kafka" % kafkaVersion exclude ("log4j", "log4j")
           )
-      },
+      } ++ silencer,
     Compile / unmanagedSources / excludeFilter := {
       if (scalaBinaryVersion.value == "2.13") {
         HiddenFileFilter ||
@@ -243,6 +241,7 @@ lazy val testkit = project
         "EmbeddedKafkaJunit4Test.java"
       } else (Test / unmanagedSources / excludeFilter).value
     },
+    Compile / compile / scalacOptions += "-P:silencer:globalFilters=scala.jdk.CollectionConverters",
     mimaPreviousArtifacts := Set(
         organization.value %% name.value % previousStableVersion.value
           .getOrElse(throw new Error("Unable to determine previous version"))
@@ -251,9 +250,9 @@ lazy val testkit = project
 
 lazy val tests = project
   .dependsOn(core, testkit)
-  .enablePlugins(AutomateHeaderPlugin, DockerCompose, BuildInfoPlugin)
+  .enablePlugins(AutomateHeaderPlugin)
   .disablePlugins(MimaPlugin, SitePlugin)
-  .configs(IntegrationTest)
+  .configs(IntegrationTest.extend(Test))
   .settings(commonSettings)
   .settings(Defaults.itSettings)
   .settings(automateHeaderSettings(IntegrationTest))
@@ -289,12 +288,6 @@ lazy val tests = project
         }
       } ++
       Seq( // integration test dependencies
-        "com.typesafe.akka" %% "akka-stream-testkit" % akkaVersion % IntegrationTest,
-        "org.scalatest" %% "scalatest" % scalatestVersion % IntegrationTest,
-        "com.spotify" % "docker-client" % "8.11.7" % IntegrationTest,
-        "com.typesafe.akka" %% "akka-slf4j" % akkaVersion % IntegrationTest,
-        "ch.qos.logback" % "logback-classic" % "1.2.3" % IntegrationTest,
-        "org.slf4j" % "log4j-over-slf4j" % slf4jVersion % IntegrationTest
       ),
     resolvers += "Confluent Maven Repo" at "https://packages.confluent.io/maven/",
     publish / skip := true,
@@ -320,25 +313,9 @@ lazy val tests = project
         "SerializationTest.java" ||
         "TransactionsExampleTest.java"
       } else (Test / unmanagedSources / excludeFilter).value
-    },
-    kafkaScale := 3,
-    buildInfoPackage := "akka.kafka",
-    buildInfoKeys := Seq[BuildInfoKey](kafkaScale),
-    dockerComposeTestLogging := true,
-    dockerComposeFilePath := (baseDirectory.value / ".." / "docker-compose.yml").getAbsolutePath,
-    dockerComposeTestCommandOptions := {
-      import com.github.ehsanyou.sbt.docker.compose.commands.test._
-      DockerComposeTestCmd(DockerComposeTest.ItTest)
-        .withEnvVar("KAFKA_SCALE", kafkaScale.value.toString)
     }
   )
 
-commands += Command.command("dockerComposeTestAll") { state ⇒
-  val extracted = Project.extract(state)
-  val (_, allTests) = extracted.runTask(tests / IntegrationTest / definedTestNames, state)
-  allTests.map(test => s"tests/dockerComposeTest it:testOnly $test").foldRight(state)(_ :: _)
-}
-
 lazy val docs = project
   .enablePlugins(AkkaParadoxPlugin, ParadoxSitePlugin, PreprocessPlugin, PublishRsyncPlugin)
   .disablePlugins(BintrayPlugin, MimaPlugin)
@@ -382,7 +359,7 @@ lazy val docs = project
 
 lazy val benchmarks = project
   .dependsOn(core, testkit)
-  .enablePlugins(AutomateHeaderPlugin, DockerCompose, BuildInfoPlugin, DockerPlugin)
+  .enablePlugins(AutomateHeaderPlugin, DockerPlugin)
   .disablePlugins(MimaPlugin, SitePlugin)
   .configs(IntegrationTest)
   .settings(commonSettings)
@@ -398,20 +375,11 @@ lazy val benchmarks = project
         "io.dropwizard.metrics" % "metrics-core" % "3.2.6",
         "ch.qos.logback" % "logback-classic" % "1.2.3",
         "org.slf4j" % "log4j-over-slf4j" % slf4jVersion,
+        "org.testcontainers" % "kafka" % testcontainersVersion % IntegrationTest,
         "com.typesafe.akka" %% "akka-slf4j" % akkaVersion % IntegrationTest,
         "com.typesafe.akka" %% "akka-stream-testkit" % akkaVersion % IntegrationTest,
         "org.scalatest" %% "scalatest" % scalatestVersion % IntegrationTest
       ),
-    kafkaScale := 1,
-    buildInfoPackage := "akka.kafka.benchmarks",
-    buildInfoKeys := Seq[BuildInfoKey](kafkaScale),
-    dockerComposeTestLogging := false,
-    dockerComposeFilePath := (baseDirectory.value / ".." / "docker-compose.yml").getAbsolutePath,
-    dockerComposeTestCommandOptions := {
-      import com.github.ehsanyou.sbt.docker.compose.commands.test._
-      DockerComposeTestCmd(DockerComposeTest.ItTest)
-        .withEnvVar("KAFKA_SCALE", kafkaScale.value.toString)
-    },
     dockerfile in docker := {
       val artifact: File = assembly.value
       val artifactTargetPath = s"/app/${artifact.name}"

core/src/main/scala/akka/kafka/javadsl/Committer.scala

@@ -8,10 +8,10 @@ import java.util.concurrent.CompletionStage
 
 import akka.annotation.ApiMayChange
 import akka.japi.Pair
-import akka.{Done, NotUsed}
-import akka.kafka.ConsumerMessage.{Committable, CommittableOffset, CommittableOffsetBatch}
+import akka.kafka.ConsumerMessage.{Committable, CommittableOffsetBatch}
 import akka.kafka.{scaladsl, CommitterSettings}
 import akka.stream.javadsl.{Flow, FlowWithContext, Sink}
+import akka.{Done, NotUsed}
 
 import scala.compat.java8.FutureConverters.FutureOps
 

core/src/main/scala/akka/kafka/scaladsl/Committer.scala

@@ -5,12 +5,12 @@
 
 package akka.kafka.scaladsl
 
-import akka.dispatch.ExecutionContexts
 import akka.annotation.ApiMayChange
-import akka.{Done, NotUsed}
+import akka.dispatch.ExecutionContexts
 import akka.kafka.CommitterSettings
-import akka.kafka.ConsumerMessage.{Committable, CommittableOffset, CommittableOffsetBatch}
+import akka.kafka.ConsumerMessage.{Committable, CommittableOffsetBatch}
 import akka.stream.scaladsl.{Flow, FlowWithContext, Keep, Sink}
+import akka.{Done, NotUsed}
 
 import scala.concurrent.Future
 

project/plugins.sbt

@@ -13,15 +13,10 @@ addSbtPlugin("org.foundweekends" % "sbt-bintray" % "0.5.5")
 addSbtPlugin("com.typesafe" % "sbt-mima-plugin" % "0.6.0")
 addSbtPlugin("com.typesafe.sbt" % "sbt-site" % "1.4.0")
 addSbtPlugin("com.eed3si9n" % "sbt-unidoc" % "0.4.2")
-addSbtPlugin("com.eed3si9n" % "sbt-buildinfo" % "0.9.0")
-// latest version with https://github.com/ehsanyou/sbt-docker-compose/pull/10
-addSbtPlugin("com.github.ehsanyou" % "sbt-docker-compose" % "67284e73-envvars-2m")
 // patched version of sbt-dependency-graph
 // depend directly on the patched version see https://github.com/akka/alpakka/issues/1388
 addSbtPlugin("net.virtual-void" % "sbt-dependency-graph" % "0.9.2+10-148ba0ff")
 resolvers += Resolver.bintrayIvyRepo("2m", "sbt-plugins")
 
 addSbtPlugin("net.aichler" % "sbt-jupiter-interface" % "0.8.3")
 resolvers += Resolver.jcenterRepo
-
-libraryDependencies += "com.spotify" % "docker-client" % "8.16.0"