✨ feat: upgrade to Kafka 4.x

Author: gaeljw

README.md

@@ -19,6 +19,7 @@ Versions match the version of Confluent Schema Registry they're built against.
 
 | embedded-kafka-schema-registry version | Confluent Schema Registry version | embedded-kafka & Kafka Kafka version | Scala versions  | Java version |
 |----------------------------------------|-----------------------------------|--------------------------------------|-----------------|--------------|
+| 8.0.0                                  | 8.0.0                             | 4.0.x                                | 2.13, 3.3       | 17+          |
 | 7.9.2                                  | 7.9.2                             | 3.9.x                                | 2,12, 2.13, 3.3 | 17+          |
 | 7.9.1                                  | 7.9.1                             | 3.9.x                                | 2,12, 2.13, 3.3 | 17+          |
 | 7.9.0                                  | 7.9.0                             | 3.9.x                                | 2,12, 2.13, 3.3 | 8+           |
@@ -69,7 +70,7 @@ class MySpec extends AnyWordSpecLike with Matchers with EmbeddedKafka {
 }
 ```
 
-* In-memory Zookeeper, Kafka, and Schema Registry will be instantiated respectively on port 6000, 6001, and 6002 and automatically shutdown at the end of the test.
+* Kafka Broker, Kafka Controller and Schema Registry will be instantiated respectively on port 6001, 6002, and 6003 and automatically shutdown at the end of the test.
 
 ## embedded-kafka-schema-registry-streams
 

build.sbt

@@ -78,7 +78,6 @@ lazy val commonSettings = Seq(
   organization       := "io.github.embeddedkafka",
   scalaVersion       := Versions.Scala213,
   crossScalaVersions := Seq(
-    Versions.Scala212,
     Versions.Scala213,
     Versions.Scala3
   )

embedded-kafka-schema-registry/src/main/scala/io/github/embeddedkafka/schemaregistry/EmbeddedKafka.scala

@@ -7,12 +7,7 @@ import io.github.embeddedkafka.schemaregistry.ops.{
   RunningSchemaRegistryOps,
   SchemaRegistryOps
 }
-import io.github.embeddedkafka.{
-  EmbeddedKafkaSupport,
-  EmbeddedServer,
-  EmbeddedZ,
-  ServerOps
-}
+import io.github.embeddedkafka.{EmbeddedKafkaSupport, EmbeddedServer, ServerOps}
 
 trait EmbeddedKafka
     extends EmbeddedKafkaSupport[EmbeddedKafkaConfig]
@@ -30,26 +25,28 @@ trait EmbeddedKafka
 
   override private[embeddedkafka] def withRunningServers[T](
       config: EmbeddedKafkaConfig,
-      actualZkPort: Int,
       kafkaLogsDir: Path
   )(body: EmbeddedKafkaConfig => T): T = {
-    val broker =
+    val (broker, controller) =
       startKafka(
         config.kafkaPort,
-        actualZkPort,
+        config.controllerPort,
         config.customBrokerProperties,
         kafkaLogsDir
       )
-    val actualKafkaPort = EmbeddedKafka.kafkaPort(broker)
-    val restApp         = startSchemaRegistry(
+
+    val actualBrokerPort     = EmbeddedKafka.kafkaPort(broker)
+    val actualControllerPort = EmbeddedKafka.controllerPort(controller)
+
+    val restApp = startSchemaRegistry(
       config.schemaRegistryPort,
-      actualKafkaPort,
+      actualBrokerPort,
       config.customSchemaRegistryProperties
     )
 
     val configWithUsedPorts = EmbeddedKafkaConfig(
-      actualKafkaPort,
-      actualZkPort,
+      actualBrokerPort,
+      actualControllerPort,
       EmbeddedKafka.schemaRegistryPort(restApp),
       config.customBrokerProperties,
       config.customProducerProperties,
@@ -61,8 +58,13 @@ trait EmbeddedKafka
       body(configWithUsedPorts)
     } finally {
       restApp.stop()
+      // In combined mode, we want to shut down the broker first, since the controller may be
+      // needed for controlled shutdown. Additionally, the controller shutdown process currently
+      // stops the raft client early on, which would disrupt broker shutdown.
       broker.shutdown()
+      controller.shutdown()
       broker.awaitShutdown()
+      controller.awaitShutdown()
     }
   }
 }
@@ -74,32 +76,29 @@ object EmbeddedKafka
   override def start()(
       implicit config: EmbeddedKafkaConfig
   ): EmbeddedKWithSR = {
-    val zkLogsDir    = Files.createTempDirectory("zookeeper-logs")
     val kafkaLogsDir = Files.createTempDirectory("kafka-logs")
 
-    val factory =
-      EmbeddedZ(startZooKeeper(config.zooKeeperPort, zkLogsDir), zkLogsDir)
-
-    val actualZookeeperPort = zookeeperPort(factory)
-    val kafkaBroker         = startKafka(
+    val (broker, controller) = startKafka(
       kafkaPort = config.kafkaPort,
-      zooKeeperPort = actualZookeeperPort,
+      controllerPort = config.controllerPort,
       customBrokerProperties = config.customBrokerProperties,
       kafkaLogDir = kafkaLogsDir
     )
 
-    val actualKafkaPort = EmbeddedKafka.kafkaPort(kafkaBroker)
-    val restApp         = EmbeddedSR(
+    val actualBrokerPort     = EmbeddedKafka.kafkaPort(broker)
+    val actualControllerPort = EmbeddedKafka.controllerPort(controller)
+
+    val restApp = EmbeddedSR(
       startSchemaRegistry(
         config.schemaRegistryPort,
-        actualKafkaPort,
+        actualBrokerPort,
         config.customSchemaRegistryProperties
       )
     )
 
     val configWithUsedPorts = EmbeddedKafkaConfigImpl(
-      kafkaPort = actualKafkaPort,
-      zooKeeperPort = actualZookeeperPort,
+      kafkaPort = actualBrokerPort,
+      controllerPort = actualControllerPort,
       schemaRegistryPort = EmbeddedKafka.schemaRegistryPort(restApp.app),
       customBrokerProperties = config.customBrokerProperties,
       customProducerProperties = config.customProducerProperties,
@@ -108,8 +107,8 @@ object EmbeddedKafka
     )
 
     val server = EmbeddedKWithSR(
-      factory = Option(factory),
-      broker = kafkaBroker,
+      broker = broker,
+      controller = controller,
       app = restApp,
       logsDirs = kafkaLogsDir,
       configWithUsedPorts

embedded-kafka-schema-registry/src/main/scala/io/github/embeddedkafka/schemaregistry/config.scala

@@ -11,7 +11,7 @@ trait EmbeddedKafkaConfig extends OriginalEmbeddedKafkaConfig {
 
 case class EmbeddedKafkaConfigImpl(
     kafkaPort: Int,
-    zooKeeperPort: Int,
+    controllerPort: Int,
     schemaRegistryPort: Int,
     customBrokerProperties: Map[String, String],
     customProducerProperties: Map[String, String],
@@ -22,13 +22,13 @@ case class EmbeddedKafkaConfigImpl(
 }
 
 object EmbeddedKafkaConfig {
-  lazy val defaultSchemaRegistryPort = 6002
+  lazy val defaultSchemaRegistryPort = 6003
 
   implicit val defaultConfig: EmbeddedKafkaConfig = apply()
 
   def apply(
       kafkaPort: Int = OriginalEmbeddedKafkaConfig.defaultKafkaPort,
-      zooKeeperPort: Int = OriginalEmbeddedKafkaConfig.defaultZookeeperPort,
+      controllerPort: Int = OriginalEmbeddedKafkaConfig.defaultControllerPort,
       schemaRegistryPort: Int = defaultSchemaRegistryPort,
       customBrokerProperties: Map[String, String] = Map.empty,
       customProducerProperties: Map[String, String] = Map.empty,
@@ -37,7 +37,7 @@ object EmbeddedKafkaConfig {
   ): EmbeddedKafkaConfig =
     EmbeddedKafkaConfigImpl(
       kafkaPort,
-      zooKeeperPort,
+      controllerPort,
       schemaRegistryPort,
       customBrokerProperties,
       customProducerProperties,

embedded-kafka-schema-registry/src/main/scala/io/github/embeddedkafka/schemaregistry/ops/SchemaRegistryOps.scala

@@ -35,7 +35,7 @@ trait SchemaRegistryOps {
 
     val restAppProperties = Map(
       RestConfig.LISTENERS_CONFIG -> s"http://localhost:$actualSchemaRegistryPort",
-      SchemaRegistryConfig.KAFKASTORE_BOOTSTRAP_SERVERS_CONFIG -> s"localhost:$kafkaPort"
+      SchemaRegistryConfig.KAFKASTORE_BOOTSTRAP_SERVERS_CONFIG -> s"PLAINTEXT://localhost:$kafkaPort"
     ) ++ customProperties
 
     val restApp = new SchemaRegistryRestApplication(

embedded-kafka-schema-registry/src/main/scala/io/github/embeddedkafka/schemaregistry/servers.scala

@@ -1,14 +1,9 @@
 package io.github.embeddedkafka.schemaregistry
 
 import java.nio.file.Path
-
 import io.confluent.kafka.schemaregistry.rest.SchemaRegistryRestApplication
-import kafka.server.KafkaServer
-import io.github.embeddedkafka.{
-  EmbeddedServer,
-  EmbeddedServerWithKafka,
-  EmbeddedZ
-}
+import io.github.embeddedkafka.{EmbeddedServer, EmbeddedServerWithKafka}
+import kafka.server.{BrokerServer, ControllerServer}
 
 import scala.reflect.io.Directory
 
@@ -28,19 +23,22 @@ case class EmbeddedSR(app: SchemaRegistryRestApplication)
 }
 
 case class EmbeddedKWithSR(
-    factory: Option[EmbeddedZ],
-    broker: KafkaServer,
+    broker: BrokerServer,
+    controller: ControllerServer,
     app: EmbeddedSR,
     logsDirs: Path,
     config: EmbeddedKafkaConfig
 ) extends EmbeddedServerWithKafka {
   override def stop(clearLogs: Boolean): Unit = {
     app.stop()
 
+    // In combined mode, we want to shut down the broker first, since the controller may be
+    // needed for controlled shutdown. Additionally, the controller shutdown process currently
+    // stops the raft client early on, which would disrupt broker shutdown.
     broker.shutdown()
+    controller.shutdown()
     broker.awaitShutdown()
-
-    factory.foreach(_.stop(clearLogs))
+    controller.awaitShutdown()
 
     if (clearLogs) {
       val _ = Directory(logsDirs.toFile).deleteRecursively()

embedded-kafka-schema-registry/src/test/scala/io/github/embeddedkafka/schemaregistry/EmbeddedKafkaObjectSpec.scala

@@ -14,14 +14,14 @@ class EmbeddedKafkaObjectSpec extends EmbeddedKafkaSpecSupport {
         val firstBroker = EmbeddedKafka.start()(
           EmbeddedKafkaConfig(
             kafkaPort = 7000,
-            zooKeeperPort = 7001,
+            controllerPort = 7001,
             schemaRegistryPort = 7002
           )
         )
         EmbeddedKafka.start()(
           EmbeddedKafkaConfig(
             kafkaPort = 8000,
-            zooKeeperPort = 8001,
+            controllerPort = 8001,
             schemaRegistryPort = 8002
           )
         )
@@ -47,11 +47,11 @@ class EmbeddedKafkaObjectSpec extends EmbeddedKafkaSpecSupport {
         EmbeddedKafka.stop()
       }
 
-      "start and stop Kafka, Zookeeper, and Schema Registry on different specified ports using an implicit configuration" in {
+      "start and stop Kafka Broker, Kafka Controller, and Schema Registry on different specified ports using an implicit configuration" in {
         implicit val config: EmbeddedKafkaConfig =
           EmbeddedKafkaConfig(
             kafkaPort = 12345,
-            zooKeeperPort = 54321,
+            controllerPort = 54321,
             schemaRegistryPort = 13542
           )
 
@@ -66,7 +66,7 @@ class EmbeddedKafkaObjectSpec extends EmbeddedKafkaSpecSupport {
     }
 
     "invoking the isRunning method" should {
-      "return true when Schema Registry, Kafka, and Zookeeper are all running" in {
+      "return true when Schema Registry, Kafka Broker, and Kafka Controller are all running" in {
         implicit val config: EmbeddedKafkaConfig =
           EmbeddedKafkaConfig()
 
@@ -76,13 +76,10 @@ class EmbeddedKafkaObjectSpec extends EmbeddedKafkaSpecSupport {
         EmbeddedKafka.isRunning shouldBe false
       }
 
-      "return true when Schema Registry, Kafka, and Zookeeper are all running, if started separately" in {
+      "return true when Schema Registry, Kafka Broker, and Kafka Controller are all running, if started separately" in {
         implicit val config: EmbeddedKafkaConfig =
           EmbeddedKafkaConfig()
 
-        EmbeddedKafka.startZooKeeper(
-          Files.createTempDirectory("zookeeper-test-logs")
-        )
         EmbeddedKafka.startKafka(Files.createTempDirectory("kafka-test-logs"))
         EmbeddedKafka.startSchemaRegistry
 
@@ -91,13 +88,10 @@ class EmbeddedKafkaObjectSpec extends EmbeddedKafkaSpecSupport {
         EmbeddedKafka.isRunning shouldBe false
       }
 
-      "return false when only Kafka and Zookeeper are running" in {
+      "return false when only Kafka Broker and Kafka Controller are running" in {
         implicit val config: EmbeddedKafkaConfig =
           EmbeddedKafkaConfig()
 
-        EmbeddedKafka.startZooKeeper(
-          Files.createTempDirectory("zookeeper-test-logs")
-        )
         EmbeddedKafka.startKafka(Files.createTempDirectory("kafka-test-logs"))
         EmbeddedKafka.startSchemaRegistry
         EmbeddedKafka.stopSchemaRegistry()

embedded-kafka-schema-registry/src/test/scala/io/github/embeddedkafka/schemaregistry/EmbeddedKafkaTraitSpec.scala

@@ -28,11 +28,11 @@ class EmbeddedKafkaTraitSpec extends EmbeddedKafkaSpecSupport {
       }
     }
 
-    "start and stop Kafka, Zookeeper, and Schema Registry successfully on non-zero ports" in {
+    "start and stop Kafka Broker, Kafka Controller, and Schema Registry successfully on non-zero ports" in {
       val userDefinedConfig =
         EmbeddedKafkaConfig(
           kafkaPort = 12345,
-          zooKeeperPort = 12346,
+          controllerPort = 12346,
           schemaRegistryPort = 12347
         )
 
@@ -49,12 +49,12 @@ class EmbeddedKafkaTraitSpec extends EmbeddedKafkaSpecSupport {
   private def everyServerIsAvailable(config: EmbeddedKafkaConfig): Assertion = {
     expectedServerStatus(config.kafkaPort, Available)
     expectedServerStatus(config.schemaRegistryPort, Available)
-    expectedServerStatus(config.zooKeeperPort, Available)
+    expectedServerStatus(config.controllerPort, Available)
   }
 
   private def noServerIsAvailable(config: EmbeddedKafkaConfig): Assertion = {
     expectedServerStatus(config.kafkaPort, NotAvailable)
     expectedServerStatus(config.schemaRegistryPort, NotAvailable)
-    expectedServerStatus(config.zooKeeperPort, NotAvailable)
+    expectedServerStatus(config.controllerPort, NotAvailable)
   }
 }

kafka-streams/src/test/scala/io/github/embeddedkafka/schemaregistry/streams/ExampleKafkaStreamsSpec.scala

@@ -31,7 +31,7 @@ class ExampleKafkaStreamsSpec
   implicit val config: EmbeddedKafkaConfig =
     EmbeddedKafkaConfig(
       kafkaPort = 7000,
-      zooKeeperPort = 7001,
+      controllerPort = 7001,
       schemaRegistryPort = 7002
     )
 

project/Dependencies.scala

@@ -13,10 +13,9 @@ object Dependencies {
   object Versions {
     val Scala3            = "3.3.6"
     val Scala213          = "2.13.16"
-    val Scala212          = "2.12.20"
-    val EmbeddedKafka     = "3.9.1"
+    val EmbeddedKafka     = "4.0.1.1"
     val ConfluentPlatform = "8.0.0"
-    val Slf4j             = "1.7.36"
+    val Slf4j             = "2.0.17"
     val ScalaTest         = "3.2.19"
   }