Merge pull request #23 from lucapertile/refactor-create-custom-topic

refactored createCustomTopic signature to use a scala Map

Author: manub

README.md

@@ -81,7 +81,7 @@ The `EmbeddedKafka` trait provides also some utility methods to interact with th
         
         def consumeFirstMessageFrom(topic: String): String
 
-        def createCustomTopic(topic: String, topicConfig: Properties): Unit
+        def createCustomTopic(topic: String, topicConfig: Map[String,String]): Unit
         
 ## Custom producers
 

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

@@ -243,11 +243,14 @@ sealed trait EmbeddedKafkaSupport {
     * Creates a topic with a custom configuration
     *
     * @param topic       the topic name
-    * @param topicConfig per topic configuration [[java.util.Properties]]
+    * @param topicConfig per topic configuration [[Map]]
     * @param config      an implicit [[EmbeddedKafkaConfig]]
     */
-  def createCustomTopic(topic: String, topicConfig: Properties)(implicit config: EmbeddedKafkaConfig): Unit = {
+  def createCustomTopic(topic: String, topicConfig: Map[String,String] = Map.empty)(implicit config: EmbeddedKafkaConfig): Unit = {
     val zkUtils = ZkUtils(s"localhost:${config.zooKeeperPort}", zkSessionTimeoutMs, zkConnectionTimeoutMs, zkSecurityEnabled)
-    try AdminUtils.createTopic(zkUtils, topic, 1, 1, topicConfig) finally zkUtils.close()
+    val topicProperties = topicConfig.foldLeft(new Properties){case (props, (k,v)) => props.put(k,v); props}
+
+    try AdminUtils.createTopic(zkUtils, topic, 1, 1, topicProperties) finally zkUtils.close()
   }
+
 }

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

@@ -109,10 +109,8 @@ class EmbeddedKafkaSpec extends EmbeddedKafkaSpecSupport with EmbeddedKafka {
       val topic = "test_custom_topic"
 
       withRunningKafka {
-        val properties: Properties = new Properties
-        properties.put("cleanup.policy", "compact")
 
-        createCustomTopic(topic, properties)
+        createCustomTopic(topic, Map("cleanup.policy"->"compact"))
 
         val zkSessionTimeoutMs  = 10000
         val zkConnectionTimeoutMs = 10000