143 Kafka should be configured to point to the launched zookeeper (#145)

instance.  Previously, if 0 was passed for both kafka and zookeeper
ports, zookeeper would start on an available port; however, kafka was
told that zookeeper was listening on port 0.

I also removed the implicit config from `EmbeddedK`.  It didn't look
like it was being used anywhere...  However, I added it as a required
argument.  This was done to allow Java code like this:

```java
int kafkaPort = 0;
int zookeeperPort = 0;

EmbeddedKafka$ kafka = EmbeddedKafka$.MODULE$;

EmbeddedKafkaConfig cfg = new EmbeddedKafkaConfigImpl(
        kafkaPort,
        zookeeperPort,
        Map$.MODULE$.empty(),
        Map$.MODULE$.empty(),
        Map$.MODULE$.empty()
);
EmbeddedK serverInfo = kafka.start(cfg);
EmbeddedKafkaConfig actualCfg = serverInfo.config();
kafka.publishStringMessageToKafka("topic", "message", actualCfg);
```

Author: crankydillo

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

@@ -61,9 +61,25 @@ object EmbeddedKafka extends EmbeddedKafkaSupport[EmbeddedKafkaConfig] {
     val factory =
       EmbeddedZ(startZooKeeper(config.zooKeeperPort, zkLogsDir), zkLogsDir)
 
-    val kafkaServer = startKafka(config, kafkaLogsDir)
+    val zkPort = zookeeperPort(factory)
+
+    val configWithUsedZookeeperPort = EmbeddedKafkaConfig(
+      config.kafkaPort,
+      zkPort,
+      config.customBrokerProperties,
+      config.customProducerProperties,
+      config.customConsumerProperties
+    )
+    val kafkaServer = startKafka(configWithUsedZookeeperPort, kafkaLogsDir)
+
+    val actualConfig = EmbeddedKafkaConfigImpl(kafkaPort(kafkaServer),
+                                               zkPort,
+                                               config.customBrokerProperties,
+                                               config.customProducerProperties,
+                                               config.customConsumerProperties)
+
     val broker =
-      EmbeddedK(Option(factory), kafkaServer, kafkaLogsDir)
+      EmbeddedK(Option(factory), kafkaServer, kafkaLogsDir, actualConfig)
 
     servers :+= broker
     broker
@@ -93,7 +109,8 @@ object EmbeddedKafka extends EmbeddedKafkaSupport[EmbeddedKafkaConfig] {
     */
   def startKafka(kafkaLogsDir: Directory)(
       implicit config: EmbeddedKafkaConfig): EmbeddedK = {
-    val broker = EmbeddedK(startKafka(config, kafkaLogsDir), kafkaLogsDir)
+    val broker =
+      EmbeddedK(startKafka(config, kafkaLogsDir), kafkaLogsDir, config)
     servers :+= broker
     broker
   }
@@ -168,6 +185,13 @@ object EmbeddedKafka extends EmbeddedKafkaSupport[EmbeddedKafkaConfig] {
       ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG -> false.toString
     ) ++ config.customConsumerProperties
 
+  private[embeddedkafka] def kafkaPort(kafkaServer: KafkaServer): Int =
+    kafkaServer.boundPort(kafkaServer.config.listeners.head.listenerName)
+
+  private[embeddedkafka] def zookeeperPort(zk: EmbeddedZ): Int =
+    zookeeperPort(zk.factory)
+  private[embeddedkafka] def zookeeperPort(fac: ServerCnxnFactory): Int =
+    fac.getLocalPort
 }
 
 private[embeddedkafka] trait EmbeddedKafkaSupport[C <: EmbeddedKafkaConfig] {
@@ -221,10 +245,8 @@ private[embeddedkafka] trait EmbeddedKafkaSupport[C <: EmbeddedKafkaConfig] {
                      zkPort,
                      config.customBrokerProperties,
                      kafkaLogsDir)
-        val kafkaPort =
-          broker.boundPort(broker.config.listeners.head.listenerName)
         val actualConfig =
-          EmbeddedKafkaConfigImpl(kafkaPort,
+          EmbeddedKafkaConfigImpl(EmbeddedKafka.kafkaPort(broker),
                                   zkPort,
                                   config.customBrokerProperties,
                                   config.customProducerProperties,

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

@@ -51,7 +51,8 @@ private[embeddedkafka] trait EmbeddedServerWithKafka extends EmbeddedServer {
   */
 case class EmbeddedK(factory: Option[EmbeddedZ],
                      broker: KafkaServer,
-                     logsDirs: Directory)(implicit config: EmbeddedKafkaConfig)
+                     logsDirs: Directory,
+                     config: EmbeddedKafkaConfig)
     extends EmbeddedServerWithKafka {
 
   /**
@@ -72,7 +73,8 @@ case class EmbeddedK(factory: Option[EmbeddedZ],
 }
 
 object EmbeddedK {
-  def apply(broker: KafkaServer, logsDirs: Directory)(
-      implicit config: EmbeddedKafkaConfig): EmbeddedK =
-    EmbeddedK(factory = None, broker, logsDirs)
+  def apply(broker: KafkaServer,
+            logsDirs: Directory,
+            config: EmbeddedKafkaConfig): EmbeddedK =
+    EmbeddedK(factory = None, broker, logsDirs, config)
 }

embedded-kafka/src/test/scala/net/manub/embeddedkafka/EmbeddedKafkaObjectSpec.scala

@@ -5,6 +5,8 @@ import org.apache.kafka.common.serialization.{
   StringSerializer
 }
 import net.manub.embeddedkafka.EmbeddedKafka._
+import org.apache.kafka.common.network.ListenerName
+import org.apache.kafka.common.security.auth.SecurityProtocol
 
 import scala.collection.JavaConverters._
 import scala.reflect.io.Directory
@@ -61,6 +63,28 @@ class EmbeddedKafkaObjectSpec extends EmbeddedKafkaSpecSupport {
         EmbeddedKafka.stop()
       }
 
+      "start and stop Kafka and Zookeeper successfully on arbitrary available ports" in {
+        val someConfig =
+          EmbeddedKafkaConfig(kafkaPort = 0, zooKeeperPort = 0)
+        val kafka = EmbeddedKafka.start()(someConfig)
+
+        kafka.factory shouldBe defined
+
+        val usedZookeeperPort = EmbeddedKafka.zookeeperPort(kafka.factory.get)
+        val usedKafkaPort = EmbeddedKafka.kafkaPort(kafka.broker)
+
+        kafkaIsAvailable(usedKafkaPort)
+        zookeeperIsAvailable(usedZookeeperPort)
+
+        kafka.config.kafkaPort should be(usedKafkaPort)
+        kafka.config.zooKeeperPort should be(usedZookeeperPort)
+
+        EmbeddedKafka.stop()
+
+        kafkaIsNotAvailable(usedKafkaPort)
+        zookeeperIsNotAvailable(usedZookeeperPort)
+      }
+
       "start and stop multiple Kafka instances on specified ports" in {
         val someConfig =
           EmbeddedKafkaConfig(kafkaPort = 12345, zooKeeperPort = 32111)