fixing embedded subproject

- fixing kafka version issues (upgrading to 0.11.x broke some things)
- commiting a test-config (so the tests do not fail because of a missing config)
- adding a docker/zookeeper to .travis, because tests for the embedded subproject require a running zookeeper

.travis.yml

@@ -1,4 +1,15 @@
+sudo: required
+
 language: clojure
+
+services:
+  - docker
+
+before_install:
+- docker pull wurstmeister/zookeeper
+- docker run -d -p 127.0.0.1:2181:2181 wurstmeister/zookeeper
+- docker ps -a
+
 jdk:
   - oraclejdk8
 

embedded/project.clj

@@ -17,4 +17,5 @@
   [[lein-monolith "1.0.1"]
    [com.roomkey/lein-v "6.2.0"]]
   :profiles {:dev {:dependencies [[midje "1.7.0"]
-                                  [jarohen/nomad "0.7.2"]]}})
+                                  [jarohen/nomad "0.7.2"]]
+                   :resource-paths ["test-resources"]}})

embedded/src/franzy/embedded/broker.clj

@@ -37,8 +37,8 @@
     (-> server
         (.brokerState)
         (.newState ^Byte state)))
-  (bound-port [_ security-protocol]
-    (.boundPort server security-protocol))
+  (bound-port [_ listener-name]
+    (.boundPort server listener-name))
   ZookeeperConnected
   (zk-utils [_]
     "Returns the broker's instance of ZkUtils, which can be used for administration.

embedded/src/franzy/embedded/extensions.clj

@@ -11,9 +11,9 @@
     (some-> server
             (.brokerState)
             (.newState state)))
-  (bound-port [{:keys [^KafkaServer server]} security-protocol]
+  (bound-port [{:keys [^KafkaServer server]} listener-name]
     (some-> server
-            (.boundPort security-protocol)))
+            (.boundPort listener-name)))
   ZookeeperConnected
   (zk-utils [{:keys [^KafkaServer server]}]
     (some-> server

embedded/src/franzy/embedded/protocols.clj

@@ -1,5 +1,6 @@
 (ns franzy.embedded.protocols
-  (:import (org.apache.kafka.common.protocol SecurityProtocol)))
+  (:import (org.apache.kafka.common.protocol SecurityProtocol)
+           (org.apache.kafka.common.network ListenerName)))
 
 (defprotocol KafkaBrokerLifecycle
   "Basic Kafka Broker lifecycle protocol."
@@ -19,4 +20,4 @@
   "Stateful Kafka Broker information."
   (set-state [this ^Byte state])
   ;;TODO: switch to clojure type
-  (bound-port [this ^SecurityProtocol security-protocol]))
+  (bound-port [this ^ListenerName listener-name]))

embedded/test-resources/config/broker.edn

@@ -0,0 +1,15 @@
+{:kakfa/broker-config
+ {:host.name                            "127.0.0.1"
+  :port                                 9092
+  :broker.id                            0
+  :advertised.host.name                 "your-insignificant-host.com"
+  :advertised.port                      9092
+  :num.partitions                       1
+  :controller.socket.timeout.ms         1500
+  :controlled.shutdown.enable           true
+  :replica.socket.timeout.ms            1500
+  :delete.topic.enable                  true
+  :controlled.shutdown.retry.backoff.ms 100
+  :log.cleaner.dedupe.buffer.size       2097152
+  :zookeeper.connect                    "127.0.0.1:2181"}
+ }

embedded/test/franzy/embedded/broker_tests.clj

@@ -5,7 +5,8 @@
             [franzy.embedded.core-test :as core-test])
   (:import (franzy.embedded.broker EmbeddedKafkaBroker StartableEmbeddedKafkaBroker)
            (kafka.utils ZkUtils)
-           (org.apache.kafka.common.protocol SecurityProtocol)))
+           (org.apache.kafka.common.protocol SecurityProtocol)
+           (org.apache.kafka.common.network ListenerName)))
 
 (facts
   "A broker can be created in multiple ways, depending on your needs."
@@ -75,8 +76,8 @@
     "It is possible to check the bound port of a broker that is started per channel."
     (with-open [broker (core-test/make-test-broker)]
       (startup broker)
-      (bound-port broker SecurityProtocol/PLAINTEXT) => (:port (core-test/make-broker-config))))
+      (bound-port broker (ListenerName/forSecurityProtocol SecurityProtocol/PLAINTEXT)) => (:port (core-test/make-broker-config))))
   (fact
     "One way of testing whether a broker is working or started is to check its bound port. It will be an exception if not started."
     (let [broker (core-test/make-test-broker)]
-      (bound-port broker SecurityProtocol/PLAINTEXT) => (throws NullPointerException))))
+      (bound-port broker (ListenerName/forSecurityProtocol SecurityProtocol/PLAINTEXT)) => (throws NullPointerException))))

embedded/test/franzy/embedded/component_tests.clj

@@ -8,7 +8,8 @@
   (:import (org.apache.kafka.common.protocol SecurityProtocol)
            (franzy.embedded.component EmbeddedBroker)
            (kafka.server KafkaServer)
-           (kafka.utils ZkUtils)))
+           (kafka.utils ZkUtils)
+           (org.apache.kafka.common.network ListenerName)))
 
 (facts
   "A broker component can be created in multiple ways, depending on your needs."
@@ -64,5 +65,5 @@
   (fact
     "It is possible to check the bound port of a broker that is started per channel."
     (let [broker (component/start (core-test/make-test-broker-component))]
-      (bound-port broker SecurityProtocol/PLAINTEXT) => (:port (core-test/make-broker-config))
+      (bound-port broker (ListenerName/forSecurityProtocol SecurityProtocol/PLAINTEXT)) => (:port (core-test/make-broker-config))
       (component/stop broker))))