Make EmbeddedKafkaConfig a trait to allow other implementation (#85)

* Make EmbeddedKafkaConfig a trait with a basic implementation

* Use def instead of val in trait definition

Author: max5599

embedded-kafka/src/main/scala/net/manub/embeddedkafka/EmbeddedKafka.scala

@@ -5,30 +5,20 @@ import java.util.Properties
 import java.util.concurrent.Executors
 
 import kafka.admin.AdminUtils
-import kafka.server.KafkaConfig._
 import kafka.server.{KafkaConfig, KafkaServer}
 import kafka.utils.ZkUtils
 import org.apache.kafka.clients.consumer.{KafkaConsumer, OffsetAndMetadata}
-import org.apache.kafka.clients.producer.{
-  KafkaProducer,
-  ProducerConfig,
-  ProducerRecord
-}
+import org.apache.kafka.clients.producer.{KafkaProducer, ProducerConfig, ProducerRecord}
+import org.apache.kafka.common.serialization.{Deserializer, Serializer, StringDeserializer, StringSerializer}
 import org.apache.kafka.common.{KafkaException, TopicPartition}
-import org.apache.kafka.common.serialization.{
-  Deserializer,
-  Serializer,
-  StringDeserializer,
-  StringSerializer
-}
 import org.apache.zookeeper.server.{ServerCnxnFactory, ZooKeeperServer}
 import org.scalatest.Suite
 
 import scala.collection.JavaConverters._
 import scala.collection.mutable
 import scala.collection.mutable.ListBuffer
 import scala.concurrent.duration._
-import scala.concurrent.{ExecutionContext, TimeoutException}
+import scala.concurrent.{ExecutionContext, ExecutionContextExecutorService, TimeoutException}
 import scala.language.{higherKinds, postfixOps}
 import scala.reflect.io.Directory
 import scala.util.Try
@@ -118,7 +108,7 @@ object EmbeddedKafka extends EmbeddedKafkaSupport {
 
 sealed trait EmbeddedKafkaSupport {
   private val executorService = Executors.newFixedThreadPool(2)
-  implicit private val executionContext =
+  implicit private val executionContext: ExecutionContextExecutorService =
     ExecutionContext.fromExecutorService(executorService)
 
   val zkSessionTimeoutMs = 10000
@@ -136,7 +126,7 @@ sealed trait EmbeddedKafkaSupport {
     withRunningZooKeeper(config.zooKeeperPort) { zkPort =>
       withTempDir("kafka") { kafkaLogsDir =>
         val broker =
-          startKafka(config.copy(zooKeeperPort = zkPort), kafkaLogsDir)
+          startKafka(config.kafkaPort, zkPort, config.customBrokerProperties, kafkaLogsDir)
         try {
           body
         } finally {
@@ -162,11 +152,11 @@ sealed trait EmbeddedKafkaSupport {
     withRunningZooKeeper(config.zooKeeperPort) { zkPort =>
       withTempDir("kafka") { kafkaLogsDir =>
         val broker: KafkaServer =
-          startKafka(config.copy(zooKeeperPort = zkPort), kafkaLogsDir)
+          startKafka(config.kafkaPort, zkPort, config.customBrokerProperties, kafkaLogsDir)
         val kafkaPort =
           broker.boundPort(broker.config.listeners.head.listenerName)
         val actualConfig =
-          config.copy(kafkaPort = kafkaPort, zooKeeperPort = zkPort)
+          EmbeddedKafkaConfigImpl(kafkaPort, zkPort, config.customBrokerProperties, config.customProducerProperties, config.customConsumerProperties)
         try {
           body(actualConfig)
         } finally {
@@ -556,10 +546,12 @@ sealed trait EmbeddedKafkaSupport {
     factory
   }
 
-  def startKafka(config: EmbeddedKafkaConfig,
-                 kafkaLogDir: Directory): KafkaServer = {
-    val zkAddress = s"localhost:${config.zooKeeperPort}"
-    val listener = s"PLAINTEXT://localhost:${config.kafkaPort}"
+  private def startKafka(kafkaPort: Int,
+                         zooKeeperPort: Int,
+                         customBrokerProperties: Map[String, String],
+                         kafkaLogDir: Directory) = {
+    val zkAddress = s"localhost:$zooKeeperPort"
+    val listener = s"PLAINTEXT://localhost:$kafkaPort"
 
     val properties = new Properties
     properties.setProperty("zookeeper.connect", zkAddress)
@@ -577,7 +569,7 @@ sealed trait EmbeddedKafkaSupport {
     // The total memory used for log deduplication across all cleaner threads, keep it small to not exhaust suite memory
     properties.setProperty("log.cleaner.dedupe.buffer.size", "1048577")
 
-    config.customBrokerProperties.foreach {
+    customBrokerProperties.foreach {
       case (key, value) => properties.setProperty(key, value)
     }
 
@@ -586,6 +578,11 @@ sealed trait EmbeddedKafkaSupport {
     broker
   }
 
+  def startKafka(config: EmbeddedKafkaConfig,
+                 kafkaLogDir: Directory): KafkaServer = {
+    startKafka(config.kafkaPort, config.zooKeeperPort, config.customBrokerProperties, kafkaLogDir)
+  }
+
   /**
     * Creates a topic with a custom configuration
     *

embedded-kafka/src/main/scala/net/manub/embeddedkafka/EmbeddedKafkaConfig.scala

@@ -1,11 +1,30 @@
 package net.manub.embeddedkafka
 
-case class EmbeddedKafkaConfig(kafkaPort: Int = 6001,
-                               zooKeeperPort: Int = 6000,
-                               customBrokerProperties: Map[String, String] = Map.empty,
-                               customProducerProperties: Map[String, String] = Map.empty,
-                               customConsumerProperties: Map[String, String] = Map.empty)
+trait EmbeddedKafkaConfig {
+  def kafkaPort: Int
+  def zooKeeperPort: Int
+  def customBrokerProperties: Map[String, String]
+  def customProducerProperties: Map[String, String]
+  def customConsumerProperties: Map[String, String]
+}
+
+case class EmbeddedKafkaConfigImpl(
+                                    kafkaPort: Int,
+                                    zooKeeperPort: Int,
+                                    customBrokerProperties: Map[String, String],
+                                    customProducerProperties: Map[String, String],
+                                    customConsumerProperties: Map[String, String]
+                                  ) extends EmbeddedKafkaConfig
 
 object EmbeddedKafkaConfig {
-  implicit val defaultConfig = EmbeddedKafkaConfig()
+  implicit val defaultConfig: EmbeddedKafkaConfig = apply()
+
+  def apply(
+             kafkaPort: Int = 6001,
+             zooKeeperPort: Int = 6000,
+             customBrokerProperties: Map[String, String] = Map.empty,
+             customProducerProperties: Map[String, String] = Map.empty,
+             customConsumerProperties: Map[String, String] = Map.empty
+           ) : EmbeddedKafkaConfig =
+    EmbeddedKafkaConfigImpl(kafkaPort, zooKeeperPort, customBrokerProperties, customProducerProperties, customConsumerProperties)
 }

embedded-kafka/src/test/scala/net/manub/embeddedkafka/EmbeddedKafkaWithRunningKafkaOnFoundPortSpec.scala

@@ -29,7 +29,7 @@ class EmbeddedKafkaWithRunningKafkaOnFoundPortSpec
           // Confirm both actual configs are running on separate non-zero ports, but otherwise equal
           allConfigs.map(_.kafkaPort).distinct should have size 3
           allConfigs.map(_.zooKeeperPort).distinct should have size 3
-          allConfigs.map(_.copy(kafkaPort = 0, zooKeeperPort = 0)).distinct should have size 1
+          allConfigs.map(config => EmbeddedKafkaConfigImpl(kafkaPort = 0, zooKeeperPort = 0, config.customBrokerProperties, config.customProducerProperties, config.customConsumerProperties)).distinct should have size 1
           actualConfig2
         }
         bothKafkaAndZkAreNotAvailable(actualConfig2)