Added partitions and replicationFactor options to createCustomTopic

Author: Christian Perez-Llamas

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

@@ -246,11 +246,13 @@ sealed trait EmbeddedKafkaSupport {
     * @param topicConfig per topic configuration [[Map]]
     * @param config      an implicit [[EmbeddedKafkaConfig]]
     */
-  def createCustomTopic(topic: String, topicConfig: Map[String,String] = Map.empty)(implicit config: EmbeddedKafkaConfig): Unit = {
+  def createCustomTopic(topic: String, topicConfig: Map[String,String] = Map.empty,
+    partitions: Int = 1, replicationFactor: Int = 1)(implicit config: EmbeddedKafkaConfig): Unit = {
+
     val zkUtils = ZkUtils(s"localhost:${config.zooKeeperPort}", zkSessionTimeoutMs, zkConnectionTimeoutMs, zkSecurityEnabled)
     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()
+    try AdminUtils.createTopic(zkUtils, topic, partitions, replicationFactor, topicProperties) finally zkUtils.close()
   }
 
 }

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

@@ -121,6 +121,24 @@ class EmbeddedKafkaSpec extends EmbeddedKafkaSpecSupport with EmbeddedKafka {
 
       }
     }
+
+    "create a topic with custom number of partitions" in {
+      implicit val config = EmbeddedKafkaConfig()
+      val topic = "test_custom_topic"
+
+      withRunningKafka {
+
+        createCustomTopic(topic, Map("cleanup.policy"->"compact"), partitions = 2)
+
+        val zkSessionTimeoutMs  = 10000
+        val zkConnectionTimeoutMs = 10000
+        val zkSecurityEnabled = false
+
+        val zkUtils = ZkUtils(s"localhost:${config.zooKeeperPort}", zkSessionTimeoutMs, zkConnectionTimeoutMs, zkSecurityEnabled)
+        try { AdminUtils.fetchTopicMetadataFromZk(topic, zkUtils).partitionsMetadata.size shouldBe 2 } finally zkUtils.close()
+
+      }
+    }
   }
 
   "the consumeFirstStringMessageFrom method" should {