zk/controller to get leader node

src/clj_kafka/zk.clj

@@ -13,3 +13,13 @@
                       :data
                       #(zk/data z (str "/brokers/ids/" %)))
                 (zk/children z "/brokers/ids")))))
+
+(defn controller
+  "Get leader node"
+  [m]
+  (with-resource [z (zk/connect (get m "zookeeper.connect"))]
+    zk/close
+    (-> (zk/data z "/controller")
+        :data
+        String.
+        Integer/valueOf)))

test/clj_kafka/test/zk.clj

@@ -3,9 +3,15 @@
         clj-kafka.zk
         [clj-kafka.test.utils :only (with-test-broker)]))
 
-(given (with-test-broker {:zookeeper-port 2182
-                          :kafka-port 9999
-                          :topic "test"}
+(def config {:zookeeper-port 2182
+             :kafka-port 9999
+             :topic "test"})
+
+(given (with-test-broker config
          (brokers {"zookeeper.connect" "127.0.0.1:2182"}))
        (expect count 1
                first {:host "localhost", :jmx_port -1, :port 9999, :version 1}))
+
+(given (with-test-broker config
+         (controller {"zookeeper.connect" "127.0.0.1:2182"}))
+       (expect identity 0))