Add possibility to override streams config in runStreams method #68 (#94)

Author: msaunier-poctu

kafka-streams/src/main/scala/net/manub/embeddedkafka/streams/EmbeddedKafkaStreams.scala

@@ -35,16 +35,18 @@ trait EmbeddedKafkaStreams extends EmbeddedKafka with TestStreamsConfig {
     * @param builder        the streams builder that will be used to instantiate the streams with
     *                       a default configuration (all state directories are different and
     *                       in temp folders)
+    * @param extraConfig    additional KafkaStreams configuration (overwrite existing keys in
+    *                       default config)
     * @param block          the code block that will executed while the streams are active.
     *                       Once the block has been executed the streams will be closed.
     */
-  def runStreams(topicsToCreate: Seq[String], builder: TopologyBuilder)(
+  def runStreams(topicsToCreate: Seq[String], builder: TopologyBuilder, extraConfig: Map[String, AnyRef] = Map.empty)(
       block: => Any)(implicit config: EmbeddedKafkaConfig): Any =
     withRunningKafka {
       topicsToCreate.foreach(topic => createCustomTopic(topic))
       val streamId = UUIDs.newUuid().toString
       logger.debug(s"Creating stream with Application ID: [$streamId]")
-      val streams = new KafkaStreams(builder, streamConfig(streamId))
+      val streams = new KafkaStreams(builder, streamConfig(streamId, extraConfig))
       streams.start()
       try {
         block