adopted version; not ready for normal use yet

Author: source-c

README.adoc

@@ -17,7 +17,7 @@ Add the following to your http://github.com/technomancy/leiningen[Leiningen's]
 
 [source,clojure]
 ----
-[net.tbt-post/clj-kafka-x "0.3.0"]
+[net.tbt-post/clj-kafka-x "0.3.1"]
 ----
 
 == Usage

project.clj

@@ -1,4 +1,4 @@
-(defproject net.tbt-post/clj-kafka-x "0.3.0"
+(defproject net.tbt-post/clj-kafka-x "0.3.1"
   :description "A Clojure wrapper for Apache Kafka v2.0.0 client"
   :url "https://github.com/source-c/clj-kafka-x"
   :license {:name "Apache License 2.0"

src/clj_kafka_x/admin.clj

@@ -1,15 +1,15 @@
 (ns clj-kafka-x.admin
   (:require [clj-kafka-x.data :refer [map->properties]])
   (:import kafka.admin.AdminUtils
-           kafka.admin.RackAwareMode.Enforced
+           (kafka.admin RackAwareMode RackAwareMode$Enforced$)
            kafka.utils.ZkUtils))
 
 (defn zk-utils
   ""
   [zk-url & {:keys [session-timeout connection-timeout security-enabled]
-                          :or {session-timeout 1000
-                               connection-timeout 1000
-                               security-enabled false}}]
+             :or   {session-timeout    1000
+                    connection-timeout 1000
+                    security-enabled   false}}]
   (ZkUtils/apply zk-url session-timeout connection-timeout (Boolean/valueOf security-enabled)))
 
 
@@ -30,15 +30,15 @@
   (create-topic z-utils \"topic-b\" :topic-config {\"cleanup.policy\" \"compact\"})
   "
   [z-utils topic & {:keys [partitions replication-factor topic-config]
-               :or {partitions 1
-                    replication-factor 1
-                    topic-config nil}}]
+                    :or   {partitions         1
+                           replication-factor 1
+                           topic-config       nil}}]
   (AdminUtils/createTopic z-utils
                           topic
                           (int partitions)
                           (int replication-factor)
                           (map->properties topic-config)
-                          (Enforced)))
+                          RackAwareMode$Enforced$))
 
 (defn topic-exists?
   "Returns true or false dependant on the existance of the given topic"

src/clj_kafka_x/consumers/ballanced.clj

@@ -1,12 +1,9 @@
 (ns clj-kafka-x.consumers.ballanced
   (:require [clojure.string :refer [join]])
   (:import (java.util.concurrent Executors)
-           (kafka.consumer Consumer ConsumerConfig)
-           (org.apache.kafka.clients.producer KafkaProducer)
+           (org.apache.kafka.clients.consumer KafkaConsumer Consumer ConsumerConfig ConsumerRecords)
            (java.util Properties)
-           (kafka.consumer KafkaStream)
-           (clojure.lang PersistentArrayMap)
-           (org.apache.kafka.clients.producer ProducerRecord)))
+           (clojure.lang PersistentArrayMap)))
 
 (defrecord KafkaMessage [topic partition offset key message])
 
@@ -18,15 +15,15 @@
   ([stream thread-num spec]
    (consume-messages stream thread-num spec (:id spec)))
   ([stream thread-num spec id]
-   (let [it (.iterator ^KafkaStream stream)]
+   (let [it (.iterator ^ConsumerRecords stream)]
      (while (.hasNext it)
        (let [msg (.next it)
              kmsg (KafkaMessage.
                     (.topic msg)
                     (.partition msg)
                     (.offset msg)
                     (.key msg)
-                    (.message msg))
+                    (.value msg))
              prc @(resolve (:processor spec))]
 
          (process-msg prc kmsg id)))
@@ -84,7 +81,7 @@
   ([config blist prefix]
    (create-consumer blist prefix (:group (:consumer config))))
   ([config blist prefix grname]
-   (Consumer/createJavaConsumerConnector
+   (KafkaConsumer.
      (create-consumer-config config (->zklist blist prefix) grname))))
 
 (defn- shutdown-topic [topic obj]
@@ -110,10 +107,10 @@
           tpool (Executors/newFixedThreadPool psize)
           consumer (create-consumer zkpool zkpref group)]
       (try (swap! storage merge
-                  {T {:instance (hk/consume-topic
+                  {T {:instance (consume-topic
                                   consumer T
                                   tpool psize
                                   (get topics-list T))
                       :pool     tpool}})
            (catch Exception e (do (shutdown-topic T {:pool tpool}))))))
-  storage)
+  storage)

src/clj_kafka_x/data.clj

@@ -2,10 +2,7 @@
   (:import [java.util HashMap Map Properties]
            [org.apache.kafka.clients.consumer ConsumerRecord ConsumerRecords OffsetAndMetadata]
            org.apache.kafka.clients.producer.RecordMetadata
-           [org.apache.kafka.common Metric MetricName Node PartitionInfo TopicPartition]
-
-           (kafka.consumer Consumer ConsumerConfig KafkaStream)
-           (kafka.message MessageAndMetadata)))
+           [org.apache.kafka.common Metric MetricName Node PartitionInfo TopicPartition]))
 
 (defprotocol ToClojure
   ""
@@ -30,14 +27,6 @@
      :key (.key x)
      :value (.value x)})
 
-  MessageAndMetadata
-  (to-clojure [x]
-    {:topic (.topic x)
-     :partition (.partition x)
-     :offset (.offset x)
-     :key (.key x)
-     :value (.message x)}) ;; FIME: :value/:message
-
   ConsumerRecords
   (to-clojure [x]
     (mapv to-clojure x))