Added ability to override default embedded kafka properties (#50)

* Cannot change default embedded kafka properties

When you use embedded kafka, you may be want to change default properties e.g. you have integration tests and want to start embedded kafka with integration tests in one network node, but an application starts in another network node. Embedded kafka runs with binding to localhost so you cannot to connect to it from another node. This PR fix this problem and you can change default propeties, so now you can set `host.name` and `advertised.host.name` correctly.

* Cannot change default embedded kafka properties

Update README

README.md

@@ -75,8 +75,8 @@ This works for both `withRunningKafka` and `EmbeddedKafka.start()`
 
 Also, it is now possible to provide custom properties to the broker while starting Kafka. `EmbeddedKafkaConfig` has a 
 `customBrokerProperties` field which can be used to provide extra properties contained in a `Map[String, String]`.
-Those properties will be added to the broker configuration, however some properties are set by the library itself and 
-in case of conflict the library values will take precedence. Please look at the source code to see what this properties
+Those properties will be added to the broker configuration, be careful some properties are set by the library itself and
+in case of conflict your values will take precedence. Please look at the source code to see what these properties
 are.
 
 ## Utility methods

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

@@ -278,9 +278,6 @@ 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")
@@ -289,6 +286,9 @@ sealed trait EmbeddedKafkaSupport {
     properties.setProperty("port", config.kafkaPort.toString)
     properties.setProperty("log.dir", kafkaLogDir.toAbsolute.path)
     properties.setProperty("log.flush.interval.messages", 1.toString)
+    config.customBrokerProperties.foreach {
+      case (key, value) => properties.setProperty(key, value)
+    }
 
     val broker = new KafkaServer(new KafkaConfig(properties))
     broker.startup()