Added a way to create topics and brokers by passing a custom configuration (#15)

* Added a way to create topics and brokers by passing a custom configuration.

* moved the local vals to the main trait to avoid multiple instantiation when whe create custom topics

Author: lucapertile

src/main/scala/net/manub/embeddedkafka/EmbeddedKafka.scala

@@ -4,9 +4,11 @@ import java.net.InetSocketAddress
 import java.util.Properties
 import java.util.concurrent.Executors
 
+import kafka.admin.AdminUtils
 import kafka.consumer.{Consumer, ConsumerConfig, Whitelist}
 import kafka.serializer.{Decoder, StringDecoder}
 import kafka.server.{KafkaConfig, KafkaServer}
+import kafka.utils.ZkUtils
 import org.apache.kafka.clients.producer.{KafkaProducer, ProducerConfig, ProducerRecord}
 import org.apache.kafka.common.serialization.{Serializer, StringSerializer}
 import org.apache.zookeeper.server.{ServerCnxnFactory, ZooKeeperServer}
@@ -75,6 +77,10 @@ sealed trait EmbeddedKafkaSupport {
   val executorService = Executors.newFixedThreadPool(2)
   implicit val executionContext = ExecutionContext.fromExecutorService(executorService)
 
+  val zkSessionTimeoutMs  = 10000
+  val zkConnectionTimeoutMs = 10000
+  val zkSecurityEnabled = false
+
   /**
     * Starts a ZooKeeper instance and a Kafka broker, then executes the body passed as a parameter.
     *
@@ -226,6 +232,7 @@ sealed trait EmbeddedKafkaSupport {
     val zkAddress = s"localhost:${config.zooKeeperPort}"
 
     val properties: Properties = new Properties
+    config.customBrokerProperties.foreach { case (key, value) => properties.setProperty(key, value) }
     properties.setProperty("zookeeper.connect", zkAddress)
     properties.setProperty("broker.id", "0")
     properties.setProperty("host.name", "localhost")
@@ -239,4 +246,8 @@ sealed trait EmbeddedKafkaSupport {
     broker.startup()
     broker
   }
+
+  def createCustomTopic(topic: String, topicConfig: Properties)(implicit config: EmbeddedKafkaConfig): Unit = {
+    AdminUtils.createTopic(ZkUtils(s"localhost:${config.zooKeeperPort}", zkSessionTimeoutMs, zkConnectionTimeoutMs, zkSecurityEnabled), topic, 1, 1, topicConfig)
+  }
 }

src/main/scala/net/manub/embeddedkafka/EmbeddedKafkaConfig.scala

@@ -1,6 +1,6 @@
 package net.manub.embeddedkafka
 
-case class EmbeddedKafkaConfig(kafkaPort: Int = 6001, zooKeeperPort: Int = 6000)
+case class EmbeddedKafkaConfig(kafkaPort: Int = 6001, zooKeeperPort: Int = 6000, customBrokerProperties: Map[String, String] = Map.empty)
 
 object EmbeddedKafkaConfig {
   implicit val defaultConfig = EmbeddedKafkaConfig()

src/test/scala/net/manub/embeddedkafka/EmbeddedKafkaSpec.scala

@@ -3,8 +3,10 @@ package net.manub.embeddedkafka
 import java.util.Properties
 import java.util.concurrent.TimeoutException
 
+import kafka.admin.AdminUtils
 import kafka.consumer.{Consumer, ConsumerConfig, Whitelist}
 import kafka.serializer.StringDecoder
+import kafka.utils.ZkUtils
 import org.apache.kafka.clients.producer.{KafkaProducer, ProducerConfig, ProducerRecord}
 import org.apache.kafka.common.serialization.{ByteArraySerializer, StringSerializer}
 import org.scalatest.exceptions.TestFailedException
@@ -114,6 +116,27 @@ class EmbeddedKafkaSpec extends EmbeddedKafkaSpecSupport with EmbeddedKafka {
     }
   }
 
+  "the createCustomTopic method" should {
+    "create a topic with a custom configuration" in {
+      implicit val config = EmbeddedKafkaConfig(customBrokerProperties = Map("log.cleaner.dedupe.buffer.size" -> "2000000"))
+      val topic = "test_custom_topic"
+
+      withRunningKafka {
+        val properties: Properties = new Properties
+        properties.put("cleanup.policy", "compact")
+
+        createCustomTopic(topic, properties)
+
+        val zkSessionTimeoutMs  = 10000
+        val zkConnectionTimeoutMs = 10000
+        val zkSecurityEnabled = false
+
+        AdminUtils.topicExists(ZkUtils(s"localhost:${config.zooKeeperPort}", zkSessionTimeoutMs, zkConnectionTimeoutMs, zkSecurityEnabled), topic) shouldBe true
+
+      }
+    }
+  }
+
   "the consumeFirstStringMessageFrom method" should {
     "return a message published to a topic" in {
       withRunningKafka {